Java 9のFlow APIを使ってリアクティブストリームを実装する方法

Flow APIとは

Flow APIは、Java 9から標準ライブラリに組み込まれた、リアクティブストリーム仕様の公式サポートです。このAPIは、Iterator（イテレータ）パターンとObserver（オブザーバ）パターンを組み合わせた設計になっています。ただし、Flow APIはRxJavaのようなエンドユーザー向けAPIではなく、ライブラリ間の相互運用を実現するための仕様である点に注意が必要です。

Flow APIを構成する4つの基本インターフェース

• Subscriber（サブスクライバ）：Publisherに対して購読を登録し、コールバックを受け取る側のコンポーネントです。
• Publisher（パブリッシャ）：登録されたサブスクライバに対して、データ項目のストリームを発行します。
• Subscription（サブスクリプション）：PublisherとSubscriberをつなぐリンクであり、データの要求やキャンセルを制御します。
• Processor（プロセッサ）：PublisherとSubscriberの中間に位置し、あるストリームを別のストリームへ変換する役割を担います。

以下の例では、「データオブジェクトを1つ要求し、出力した後にもう1つを要求する」という動作を行う基本的なサブスクライバを作成しています。セッションを完了させるには、Javaが標準で提供するPublisher実装であるSubmissionPublisherを使用します。

実装例

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");
    }
}

// メインクラス
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

このように、Flow APIを活用することで、バックプレッシャー（背圧）を考慮した非同期ストリーム処理を外部ライブラリに依存せずに実現できます。サブスクライバ側はrequest()メソッドで受け取れるデータ量を制御できるため、自身の処理能力を超えるデータの流入を防ぎ、安定したデータ処理が可能になります。