リアクティブストリームとCompletableFutureの改善

Java 9で「変わること」と、Javaのこれまで 第5回

拡張されたCompletableFuture

　サンプルコード内でもPublisherの処理が終了することを待つために、CompletableFutureを使用しました。このクラスはJava 8で追加されたFuture機能の拡張クラスです。Java 9ではこのクラスがさらに拡張され、いくつかの並行処理における面倒な処理が簡単に記述できるように改善されました。その中でもよく使うと思われる機能を紹介します。

終了後に行う処理を定義する

　リスト5のサンプルコード内では、getメソッドを用いて終了を待ってしまいました。しかし、getメソッドはメインスレッドを止めてしまうコードです。並列処理を行っているケースではメインスレッドを止めたくないはずです。この場合にはthenAcceptメソッドを使えば、記述した終了処理をメインスレッドとは別で実行することができます。

リスト5　thenAcceptメソッドの利用例

CompletableFuture<Integer> f = publisher.consume(s -> {});
f.thenAccept(s -> {
    // 以下のコードはメインスレッドを止めずに実行されます
    System.out.println(Thread.currentThread().getName() + ":thenAccept()");
});

処理にタイムアウトを追加する

　リスト6のサンプルコード内では、TimerTask内で終了をコントロールしていましたが、実際のケースではコントロールできない場合があります。特にデバイスからデータを取得している場合、デバイス側の動作をコントロールすることができないケースが多々あります。そのため、ある程度の時間が経過したら処理を中断したいこともあるかと思います。この場合は、リスト6のようにorTimeoutメソッドを用いると便利です。orTimeoutメソッドは正常終了しない場合に限り、指定した時間が経過するとjava.util.concurrent.TimeoutException例外を投げます。これとよく似たメソッドにcompleteOnTimeoutというメソッドもありますが、こちらはタイムアウトが発生した場合にデフォルト値を指定することができるため、getメソッドでその値を取得することが可能です。

リスト6　orTimeoutメソッドの利用例

try {
    CompletableFuture<Integer> f = publisher.consume(s -> {});
    f.orTimeout(30, TimeUnit.SECONDS);  // 30秒でタイムアウトにする

}
catch(Exception ex){
    publisher.closeExceptionally(ex);
    timer.cancel();
    System.out.println("error : "+ ex.getMessage());
}

例外を発生させるCompletedFutureを作成する

　CompletableFutureを戻り値に持つメソッドを実装しているケースでも、事前条件により即時に値を返したい場合があります。リスト7のサンプルコード（1）のように正常な値を返す場合にはcompletedFutureメソッドがすでに用意されていましたが、例外を発生させるメソッドはありませんでした。Java 9では、（2）のように引数で指定した例外を発生させるfailedFutureメソッドが追加されました。

リスト7　failedFutureメソッドの利用例

private int n = -1;
public CompletableFuture<Integer> getFuture(){
    if(n == 1){
        // (1) 1を返すCompletableFutureを返す
        return CompletableFuture.completedFuture(1);
    }
    else if(n < 0){
        // (2) 指定したエラーを発生させるCompletableFutureを返す
        return CompletableFuture.failedFuture(new IllegalAccessException("access error"));
    }

    //  何らかの並列処理
}

最後に

　リアクティブストリーム機能の追加やCompletableFutureの改善、そして、次回紹介するコレクション操作に関する変更など、Java 9のライブラリの変更は並行処理に関する変更が多く見られます。

　並行処理の記述は実際に結果を確認しても、その内部の動作制限などを把握することは難しく、また、機能上の違いも確認することは難しいです。そのため、プログラミングに慣れていても、あまり得意ではない方も多いのではないでしょうか。

　しかし、現在のトレンドのひとつとしてマイクロサービスがあり、マイクロサービスを実現する上で、これらのモデルは重要な要素であると言っても過言ではありません。筆者もJavaで1つの大きなWebサービスを作るという機会はなくなり、複数のサービスがデータ連携をして1つのシステムを構成するケースが多くなっています。

　その中でも並列処理に関する部分にはいつも頭を悩ませていますが、Java 9によって記述しやすくなったことに気づかされました。

　次回は、今回紹介できなかった残りのライブラリの変更について紹介します。

参考資料

• Java 9のAPIドキュメント
• RxJava