Java 9のFlow APIをPublisher-Subscriberパターンで実装する方法

Flow API（java.util.concurrent.Flow）は、Java 9で新たに導入されたAPIです。このAPIを利用することで、Publisher（パブリッシャー）とSubscriber（サブスクライバー）というインターフェースがどのように連携し、目的の処理を実現するのかを理解しやすくなります。

Flow APIは、リアクティブストリーム仕様に基づいたPublisher、Subscriber、Subscription、Processorの4つのインターフェースで構成されています。

以下の例では、Publisher-Subscriberインターフェースを使用してFlow APIを実装する方法を紹介します。

実装例

import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;

public class FlowAPITest {
    public static void main(String args[]) {
        Publisher<Integer> publisherSync = new Publisher<Integer>() {   // パブリッシャーの作成
            @Override
            public void subscribe(Subscriber<? super Integer> subscriber) {
                for(int i = 0; i < 10; i++) {
                    System.out.println(Thread.currentThread().getName() + " | Publishing = " + i);
                    subscriber.onNext(i);
                }
                subscriber.onComplete();
            }
        };
        Subscriber<Integer> subscriberSync = new Subscriber<Integer>() {   // サブスクライバーの作成
            @Override
            public void onSubscribe(Subscription subscription) {
            }
            @Override
            public void onNext(Integer item) {
                System.out.println(Thread.currentThread().getName() + " | Received = " + item);
                try {
                    Thread.sleep(100);
                } catch(InterruptedException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void onError(Throwable throwable) {
            }
            @Override
            public void onComplete() {
            }
        };
        publisherSync.subscribe(subscriberSync);
    }
}

コードの解説

この例では、まず0から9までの整数を発行するPublisherを作成しています。subscribe()メソッド内のループで各値をsubscriber.onNext()経由で通知し、すべての発行が完了した時点でonComplete()を呼び出してストリームを終了させます。

一方、Subscriber側では、onNext()メソッドで受け取った値をコンソールに出力し、Thread.sleep(100)によって1件ごとに100ミリ秒の遅延を設けています。これにより、データが1件ずつ順番に流れる様子を確認できます。

実行結果

main | Publishing = 0
main | Received = 0
main | Publishing = 1
main | Received = 1
main | Publishing = 2
main | Received = 2
main | Publishing = 3
main | Received = 3
main | Publishing = 4
main | Received = 4
main | Publishing = 5
main | Received = 5
main | Publishing = 6
main | Received = 6
main | Publishing = 7
main | Received = 7
main | Publishing = 8
main | Received = 8
main | Publishing = 9
main | Received = 9

実行結果を見ると、「Publishing」と「Received」が交互に出力されていることが分かります。これは、この実装がmainスレッド上で同期的に動作しており、パブリッシャーが値を発行するたびにサブスクライバーがその処理を完了してから次の値を受け取っているためです。