Java 9のSubscriberインターフェースのルールとは？Reactive Streamsの規約を徹底解説

Java 9で導入されたReactive Streams（Flow API）において、SubscriberインターフェースはPublisherを購読（サブスクライブ）し、データを受け取る役割を担います。具体的には、onNext()メソッドでアイテムを順次受け取り、onError()メソッドでエラーメッセージを受け取り、onComplete()メソッドで「これ以上アイテムは届かない」ことを示す完了シグナルを受け取ります。

これらのシグナルが発生する前に、必ずPublisher側からonSubscribe()メソッドが呼び出されます。これは、SubscriberとPublisherの間でSubscription（購読契約）が確立されたことを意味します。

Subscriberインターフェースの定義

Subscriberインターフェースは、以下の4つのメソッドで構成されています。

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

Subscriberインターフェースが守るべきルール

Reactive Streamsの仕様では、Subscriberの実装が従うべき厳格なルールが定められています。主なルールを目的別に整理して紹介します。

リクエストとシグナル受信に関するルール

• Subscriberは、onNext()シグナルを受け取るために、Subscription.request(long n)メソッドを呼び出さなければなりません。これはバックプレッシャー（流量制御）の中核となる仕組みです。
• Subscriberは、事前にSubscription.request(long n)を呼び出していたかどうかにかかわらず、onComplete()シグナルを受け取らなければなりません。
• 同様に、request(long n)の呼び出しの有無にかかわらず、onError()シグナルを受け取らなければなりません。
• Subscriberは、シグナルメソッドへのすべての呼び出しが、対応するシグナルの処理よりも先に発生すること（happens-before関係）を保証しなければなりません。つまり、受信したシグナルを自らの処理ロジックへ適切に伝播させる責任を負います。

完了・エラー時の振る舞いに関するルール

• onComplete()およびonError(Throwable t)メソッド内では、SubscriptionやPublisherのいかなるメソッドも呼び出してはなりません。
• onComplete()およびonError(Throwable t)メソッドは、シグナルを受信した時点で、そのSubscriptionはキャンセル済みであるとみなさなければなりません。

キャンセルに関するルール

• Subscriberは、onSubscribe()シグナルを受け取った時点ですでにアクティブなSubscriptionを持っている場合、新しく渡されたSubscriptionに対してSubscription.cancel()メソッドを呼び出さなければなりません。
• Subscriptionが不要になった場合は、Subscription.cancel()メソッドを呼び出さなければなりません。
• Subscriberは、自身のSubscriptionのrequestおよびcancelメソッドへのすべての呼び出しが逐次的（シリアル）に実行されることを保証しなければなりません。
• Subscription.cancel()を呼び出した後でも、すでに要求済みで未処理の要素が残っている場合、1つ以上のonNext()シグナルを受け取る可能性があります。また、cancel()は基盤となるリソースのクリーンアップ処理が即座に実行されることを保証しないため注意が必要です。

その他の重要な規約

• onSubscribe()は、同一のSubscriberに対して最大1回しか呼び出されてはなりません。
• onSubscribe()、onNext()、onError()、onComplete()の各メソッドは、原則として正常に制御を返さなければなりません。ただし、引数として渡されたパラメータがnullである場合は例外で、呼び出し元に対してNullPointerExceptionをスローしなければなりません。

これらのルールは、PublisherとSubscriberの間で非同期処理を安全かつ予測可能に行うための契約（仕様）です。特にバックプレッシャーを扱うrequest(long n)と、リソース解放に関わるcancel()の挙動は、リアクティブアプリケーションを正しく実装する上で非常に重要なので、しっかりと理解しておきましょう。