【Java】JavaでPromiseを使った非同期処理を記述したい ～ JavaScriptのPromiseライクな文法の試行～

概要

• Java言語で「JavaScriptのPromiseライクな文法で、非同期処理・並行処理を記述してみたい」と思いやってみました。
• ソースは以下リポジトリにあります
  https://github.com/riversun/java-promise

コード例

「非同期処理１が終わったら、その結果を使った非同期処理２が実行される」処理はJavaではどのように書けば良いでしょうか？

• 解1:Java1.4時代の解：Threadクラスの機能でがんばる

  • Threadを入れ子式に持つとか、joinで終了待ちとか。並列処理黎明期。

• 解2:Java5(1.5)時代の解：Callable/Futureで幸せになれたのかな・・・

  • Future/Callableで結果返せてうれしい、加えセマフォやラッチなど小道具そろったがそれなりに頑張る必要あり。

• 解3:Java8時代の解：CompletableFutureで幸せになる、はず。

  • ついに待望?!のFuture/Promiseパターン的な仕組みが標準で登場！

• 本稿では上の3つの解とは別の切り口で、以下のコードのようにJavaScriptのPromiseライクに書いてみました。

JavaでPromiseのサンプル

Promise.resolve()
 .then((action, data) -> {
   //非同期処理1
   new Thread(() -> {System.out.println("Process1");action.resolve("Result-1");}).start();
 })
 .then((action, data) -> {
   //非同期処理2
   new Thread(() -> {System.out.println("Process2");action.resolve("Result-2");}).start();
 })
 .start();

本稿でやりたいこと

• やりたいことは以下のような処理を「JavaScriptのPromiseライクに記述すること」となります
  • 複数あるAPIを非同期に呼んで、結果を受け取ったら次のAPIを呼ぶ、という一連の処理
  • 複数の処理を同時に（並列に）に動かし、それがすべて完了したら次の処理に移るような処理

本稿で対象としないこと

• (アカデミックな)Future/Promiseパターンの具現化
• Java標準のコンカレント処理の使い方
  • Java5(1.5.0)以降から使えるExecutorServiceやCallable
  • Java8以降から使えるCompletableFutureを使って書く方法

対象環境

• Java5以降
  • ライブラリはJava1.6ベースのAndroidでも動作します
  • Java8のコンカレント系APIはつかっていません

使い方（依存関係)

ライブラリjava-promiseとしてMavenレポジトリにありますので、以下を追加すればすぐに使えます。

Maven

POM.xmlのdependency

<dependency>
	<groupId>org.riversun</groupId>
	<artifactId>java-promise</artifactId>
	<version>1.1.0</version>
</dependency>

Gradle

build.gradle

dependencies {
    compile 'org.riversun:java-promise:1.1.0'
}

build.gradle(Android)

dependencies {
    implementation 'org.riversun:java-promise:1.1.0'
}

本編

JavaScriptで書くPromiseと、本稿で紹介するJavaで書く方法との比較

まず、比較のためにJavaScriptでPromiseを書いてみる

以下のコードは'foo'という文字列に非同期に実行された処理結果('bar')を連結するだけのJavaScriptのサンプルコードとなる。MDNでPromiseのサンプルとして公開されているものから抜粋した。

Example.js

Promise.resolve('foo')
    .then(function (data) {
        return new Promise(function (resolve, reject) {
            setTimeout(function () {
                const newData = data + 'bar';
                resolve(newData);
            }, 1);
        });
    })
    .then(function (data) {
        return new Promise(function (resolve, reject) {
            console.log(data);
            resolve();
        });
    });
console.log("Promise in JavaScript");

実行結果は以下のとおり

Promise in JavaScript
foobar

次にJava8でjava-promiseを使って書く

Example.java

import org.riversun.promise.Promise;

public class Example {

    public static void main(String[] args) {

        Promise.resolve("foo")
                .then(new Promise((action, data) -> {
                    new Thread(() -> {
                        String newData = data + "bar";
                        action.resolve(newData);//#resolveで次の処理に移行
                    }).start();//別スレッドで実行
                }))
                .then(new Promise((action, data) -> {
                    System.out.println(data);
                    action.resolve();
                }))
                .start();//処理開始のトリガー
        System.out.println("Promise in Java");
    }
}

実行結果は以下のとおり

Promise in Java
foobar

Promise以下の実行は非同期（別スレッド）で行われるので、この例ではSystem.out.println("Promise in Java");が実行されているのがわかる。

処理の都合最後に.start()を呼び出してPromiseチェインのトリガーをしている以外はJavaScriptのPromiseライクな文法に近づけてみた。

記法

ラムダ式を使わないで書く(Java7以前)

ラムダ式を使わなければ以下のようになる

ラムダ式を使わないで書いた場合

Promise.resolve("foo")
        .then(new Promise(new Func() {
            @Override
            public void run(Action action, Object data) throws Exception {
                new Thread(() -> {
                    String newData = data + "bar";
                    action.resolve(newData);
                }).start();
            }
        }))
        .then(new Promise(new Func() {
            @Override
            public void run(Action action, Object data) throws Exception {
                new Thread(() -> {
                    System.out.println(data);
                    action.resolve();
                }).start();
            }
        }))
        .start();

(action,data)->{}となっていた部分の正体はJavaScriptでいうところの function を表すインタフェースとなる。

Func.java

public interface Func {
    public void run(Action action, Object data) throws Exception;
}

さらにシンプルに書く

Promise.then(new Promise())ではなくPromise.then(new Func())でもOK。new Funcはラムダ式におきかえるとPromise.then((action,data)->{})となり、さらにシンプルになる。

then(Func)をつかって書く

Promise.resolve("foo")
   .then((action, data) -> {
       new Thread(() -> {
           String newData = data + "bar";
           action.resolve(newData);
       }).start();
   })
   .then((action, data) -> {
       System.out.println(data);
       action.resolve();
   })
   .start();

Promiseをつかった並行実行の各種パターン紹介

(1) Promise.then:非同期処理を順番通り実行する

コード:

public class Example20 {

    public static void main(String[] args) {

        // 処理１（別スレッド実行）
        Func function1 = (action, data) -> {
            new Thread(() -> {
                System.out.println("Process-1");
                Promise.sleep(1000);// Thread.sleepと同じ
                action.resolve("Result-1");// ステータスを"fulfilled"にして、次の処理に結果("Result-1")を伝える
            }).start();// 別スレッドでの非同期処理開始
        };

        // 処理２
        Func function2 = (action, data) -> {
            System.out.println("Process-2 result=" + data);
            action.resolve();
        };

        Promise.resolve()// 処理を開始
                .then(function1)// 処理１実行
                .then(function2)// 処理２実行
                .start();// 開始

        System.out.println("Hello,Promise");
    }

実行結果:

Hello,Promise
Process-1
Process-2 result=Result-1

説明：
thenの文法は Promise.then(onFulfilled[, onRejected]); つまり引数を2つまでとることができる
最初の引数onFulfilledは前の実行がfulfilled(≒成功)ステータスで終了した場合に実行される。
2つめの引数onRejectedはオプションだが、こちらは前の実行がrejected(≒失敗)ステータスで終了した場合に実行される。このサンプルは1つめの引数のみを指定している。

処理フロー:

1. Promise.resolveでステータスをfullfilledにしてthenにチェインする。
2. fullfilledなのでthenでは第一引数に指定されたfunction1を実行する
3. function1もaction.resolveによりステータスをfullfilledにする
4. function1はaction.resolveにString型引数**"Result-1"**をセットする
5. 次のthenもステータスがfullfilledなのでfunction2が実行される
6. function2実行時の引数dataにはfunction1の結果**"Result-1"**が格納されている

(2) action.resolve,action.reject：実行結果によって処理を分岐する

コード:

public class Example21 {

    public static void main(String[] args) {

        Func function1 = (action, data) -> {
            System.out.println("Process-1");
            action.reject();// ステータスを "rejected" にセットして実行完了
        };

        Func function2_1 = (action, data) -> {
            System.out.println("Resolved Process-2");
            action.resolve();
        };

        Func function2_2 = (action, data) -> {
            System.out.println("Rejected Process-2");
            action.resolve();
        };

        Promise.resolve()
                .then(function1)
                .then(
                        function2_1, // ステータスが fulfilled のときに実行される
                        function2_2 // ステータスが rejected のときに実行される
                )
                .start();

        System.out.println("Hello,Promise");

    }
}

実行結果:

Hello,Promise
Process-1
Rejected Process-2

説明:

function1は

action.reject();

で完了しているので、ステータスがrejectedとなる。
次のthenは

 .then(
         function2_1, // ステータスが fulfilled のときに実行される
         function2_2 // ステータスが rejected のときに実行される
 )

としている。
前述のとおり、thenの文法は Promise.then(onFulfilled[, onRejected]);であるので、
function1の完了ステータスがrejectedであるため、ここではthenの2つめの引数であるfunction2_2が実行される。

処理フロー:

(3)Promise.always： resolve、rejectどちらの処理結果も受け取る

コード:

public class Example30 {

    public static void main(String[] args) {
        Func function1 = (action, data) -> {
            action.reject("I send REJECT");
        };
        Func function2 = (action, data) -> {
            System.out.println("Received:" + data);
            action.resolve();
        };
        Promise.resolve()
                .then(function1)
                .always(function2)// ステータスが"fulfilled"でも"rejected"でも実行される
                .start();
    }
}

実行結果:

Received:I send REJECT

説明:

.always(function2)

のように**always((action,data)->{})**は、その前の処理が resolvedによるステータスfulfilledであろうと、rejectedによるステータスrejectedであろうと必ず実行される。

処理フロー:

(4)Promise.all：複数の並列な非同期処理の完了待ちをして次に進む

コード:

public class Example40 {
    @SuppressWarnings("unchecked")
    public static void main(String[] args) {
        //非同期処理１
        Func function1 = (action, data) -> {
            new Thread(() -> {
                Promise.sleep(1000); System.out.println("func1 running");action.resolve("func1-result");
            }).start();
        };
        //非同期処理２        
        Func function2 = (action, data) -> {
            new Thread(() -> {
            Promise.sleep(500);System.out.println("func2 running"); action.resolve("func2-result");
            }).start();
        };
        //非同期処理３
        Func function3 = (action, data) -> {
            new Thread(() -> {
            Promise.sleep(100);System.out.println("func3 running");action.resolve("func3-result");
            }).start();
        };
        //最後に結果を受け取る処理
        Func function4 = (action, data) -> {
            System.out.println("結果を受け取りました");
            List<Object> resultList = (List<Object>) data;
            for (int i = 0; i < resultList.size(); i++) {
                Object result = resultList.get(i);
                System.out.println("非同期処理" + (i + 1) + "の結果は " + result);
            }
            action.resolve();
        };
        
        Promise.all(function1, function2, function3)
                .always(function4)
                .start();
    }
}

実行結果:

func3 running
func2 running
func1 running
非同期処理の結果を受け取りました
非同期処理1の結果は func1-result
非同期処理2の結果は func2-result
非同期処理3の結果は func3-result

説明:

• Promise.all(function1,function2,・・・・functionN)はfunction1～functionNの複数の処理を引数にとることができ、それらを並列実行する

• 並列実行が終わると、チェインされたthen（ここではalways)に処理が移行する。

• 上の例では function1,function2,function3が並列に実行されるが、function1～function3すべてがfulfilledで完了した場合は、各function1～function3の結果がListに格納されthenに渡る。その際、格納順序は、引数に指定された function1,function2,function3の順番となる。(この仕様もJavaScriptのPromiseと同一)

• function1～function3のうち、どれか1つでも失敗≒rejectになった場合、いちばん最初にrejectになったfunctionの結果(reject reason)が次のthenに渡る。（fail-fast原則）

処理フロー:

(5)Promise.all：その２スレッドプールを自分で指定する

(4)で説明したとおり、Promise.allでFuncを並列動作をさせることができるが、事前に並列動作を行うときのスレッド生成ポリシーをExecutorをつかって定義可能。また、既に別の用途で使うために用意したスレッドプールをPromise.allに転用しても良い。

コード例:

public class Example41 {
    @SuppressWarnings("unchecked")
    public static void main(String[] args) {

        final ExecutorService myExecutor = Executors.newFixedThreadPool(2);

        // 非同期処理１
        Func function1 = (action, data) -> {
            System.out.println("No.1 " + Thread.currentThread());
            new Thread(() -> {
                Promise.sleep(1000);System.out.println("func1 running");action.resolve("func1-result");
            }).start();
        };

        // 非同期処理２
        Func function2 = (action, data) -> {
            System.out.println("No.2 " + Thread.currentThread());
            new Thread(() -> {
                Promise.sleep(500);System.out.println("func2 running");action.resolve("func2-result");
            }).start();
        };

        // 非同期処理３
        Func function3 = (action, data) -> {
            System.out.println("No.3 " + Thread.currentThread());
            new Thread(() -> {
                Promise.sleep(100);System.out.println("func3 running");action.resolve("func3-result");
            }).start();
        };
        
        // 最後に結果を受け取る処理
        Func function4 = (action, data) -> {
            System.out.println("No.4 final " + Thread.currentThread());
            System.out.println("結果を受け取りました");
            List<Object> resultList = (List<Object>) data;
            for (int i = 0; i < resultList.size(); i++) {
                Object result = resultList.get(i);
                System.out.println("非同期処理" + (i + 1) + "の結果は " + result);
            }
            myExecutor.shutdown();
            action.resolve();
        };

        Promise.all(myExecutor, function1, function2, function3)
                .always(function4)
                .start();
    }
}

実行結果:

No.1 Thread[pool-1-thread-2,5,main]
No.2 Thread[pool-1-thread-2,5,main]
No.3 Thread[pool-1-thread-2,5,main]
func3 running
func2 running
func1 running
No.4 final Thread[pool-1-thread-1,5,main]
結果を受け取りました
非同期処理1の結果は func1-result
非同期処理2の結果は func2-result
非同期処理3の結果は func3-result

結果から、Funcは同じスレッドプールから取り出されたスレッドで実行されていることがわかる。
（Funcの中であえてさらに非同期処理(new Thread)しているので、その非同期処理は指定したスレッドプールの外側になる）

説明:

• Promise.allの実行に使うExecutorを定義する。以下はプールサイズが2のスレッドプール。

final ExecutorService myExecutor = Executors.newFixedThreadPool(2);

• Promise.all(executor,func1,func2,func3,・・・・funcN)のようにしてExecutorを指定できる

 Promise.all(myExecutor, function1, function2, function3)
         .always(function4)
         .start();

• 独自にExecutorを指定した場合は、忘れずにshutdownする

Func function4 = (action, data) -> {
   //中略
    myExecutor.shutdown();
    action.resolve();
};

スレッド生成ポリシー：

• スレッドプールのサイズは2以上を指定する必要がある。（つまり、singleThreadExecutorは利用不可。）
• java-promiseでは、Promise.allを行う場合、非同期実行のため1スレッドを使う。
• さらに、Promise.allで並列実行をおこなうため、並列実行用に最低1スレッドが必要。(1スレッドだと並列とはいわないが）
• この2つを合計すると2スレッド以上必要になる。

まとめ

• JavaでPromiseを「JavaScriptライクに記述」する方法を試行しました

  • Java8ラムダ式をうまくとりいれるとJavaScriptの記法に近いカタチでPromiseを実行できました
  • 簡潔な記法で気の利いた処理ができるという点はJavaScript(ES)他スクリプト系言語の進化に学びたいとおもいます
    (非同期実行もJavaScriptではasync/awaitまで進化しました)

• JavaでPromise(java-promise)のライブラリ側ソースコードは以下にあります
  https://github.com/riversun/java-promise

  • git clone https://github.com/riversun/java-promise.git して
  • mvn test すると単体テスト動確ができます

• また、本稿内に掲載したサンプルコードは以下にあります
  https://github.com/riversun/java-promise-examples/tree/master-ja