Java9でpubsubを実装する

Java9でpubsubを実装する
1. 関連

Java9でpubsubを実装する

Java9からpubsubを簡単に実装できるようになっています。サブスクライバーを実装します。

オーバーライド
onSubscribe()
onNext()
onError()
onComplete()

Subscriberインタフェースからオーバーライドするメソッドは4つです。

package jp.co.confrage;

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

public class Subscriber implements Flow.Subscriber{
  private Flow.Subscription sub;  
  @Override  
  public void onSubscribe(Subscription sub) {  
    this.sub = sub; // 登録
    sub.request(1); // 最初のデータをリクエスト
    System.out.println("登録");
  }
  @Override  
  public void onNext(T item) {
    System.out.println(item);
    sub.request(1); // 次のデータをリクエスト
    }
  @Override  
  public void onComplete() {  
    System.out.println("完了");  // 完了処理
  }
  @Override  
  public void onError(Throwable t) {  
    t.printStackTrace();  // エラー発生処理
  } 
}

SubmissionPublisherクラスを使用してパブリッシャーのインスタンスを作成します。

subscribeメソッドが用意されているので引数にサブスクライバーのインスタンスをセットします。これでパブサブの設定は完了です。

あとはSubmissionPublisherクラスのsubmitメソッドでpublishします。

以下は簡単なパブサブの実装です。

package jp.co.confrage;

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

public class Sample {
  public static void main(String[] args) {
    SubmissionPublisher publisher = new SubmissionPublisher<>();  
    Subscriber subscriber = new Subscriber<>();  
      publisher.subscribe(subscriber); // サブスクライバーを登録する
      List items = List.of("1", "2", "3", "4", "5", "6");
      items.forEach(publisher::submit); // パブリッシュする
      publisher.close(); 

      CompletableFuture f = publisher.consume(s -> {});
      try {
        f.get(); // pubsubが終了するまで待つ
      }catch(Exception ex){}
    }
}

ちゃんと実行されているのが結果からわかります。