Java9でFlowAPIを使用してリアクティブストリームを実装するにはどうすればよいですか？

フローAPI はJava9以降のリアクティブストリーム仕様の公式サポートです。これは両方のイテレータの組み合わせです。 およびオブザーバー パターン。 フローAPI は相互運用仕様であり、 RxJavaのようなエンドユーザーAPIではありません 。

フローAPI 4つの基本的なインターフェースで構成されています：

• サブスクライバー ： サブスクライバーは、コールバックのためにパブリッシャーにサブスクライブします。
• 発行者 ： パブリッシャーは、登録されたサブスクライバーにデータアイテムのストリームを公開します。
• サブスクリプション ： 出版社と購読者の間のリンク。
• プロセッサ ： プロセッサはパブリッシャーとサブスクライバーの間に位置し、あるストリームを別のストリームに変換します。

以下の例では、1つのデータオブジェクトを要求し、それを印刷して、もう1つを要求する基本的なサブスクライバーを作成しました。 Javaが提供するパブリッシャー実装を使用できます（ SubmitPublisher ）セッションを完了します。

例

import java.util.concurrent.Flow;
import java.util.List;
import java.util.concurrent.SubmissionPublisher;

class MySubscriber<T>implements Flow.Subscriber<T> {
   private Flow.Subscription subscription;
   @Override
   public void onSubscribe(Flow.Subscription subscription) {
      this.subscription = subscription;
      this.subscription.request(1);
   }
   @Override
   public void onNext(T item) {
      System.out.println(item);
      subscription.request(1);
   }
   @Override
   public void onError(Throwable throwable) {
      throwable.printStackTrace();
   }
   @Override
   public void onComplete() {
      System.out.println("Done");
   }
}

// main class
public class FlowTest {
   public static void main(String args[]) {
      List<String> items = List.of("1", "2", "3", "4", "5", "6", "7", "8", "9", "10");
      SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
      publisher.subscribe(new MySubscriber<>());
      items.forEach(s -> {
         try {
            Thread.sleep(1000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         publisher.submit(s);
      });
      publisher.close();
   }
}

1
2
3
4
5
6
7
8
9
10
Done