はじめに
この記事はExecutorフレームワークのありがたみを知るシリーズ第６弾、CompletableFutureに関する記事です。
関連記事：
- Executors#newSingleThreadExecutor編
- Executors#newFixedThreadPool編
- Executors#newCachedTheadPool編
- Callable/Future編
- ExecutorCompletionService編
参考：
【マルチスレッド】CompletableFutureについて
CompletableFutureを使って非同期処理をやってみた
Futureの問題点
CompletionService編に引き続き、CompletableFutureはFutureの抱える問題を解決してくれるものです。
例によって、以下の「受け取った数値になんらかの計算を施す」タスクについて考えます。
public class MyTask implements Callable<Integer> {
private int value;
public MyTask(int value) {
this.value = value;
}
@Override
public Integer call() throws Exception {
System.out.println("[start] value:" + value + " thread-id:" + Thread.currentThread().getId());
// 何か複雑な処理
System.out.println("[ end ] value:" + value + " thread-id:" + Thread.currentThread().getId());
return value + 10;
}
}
依存関係の管理が面倒
あるタスクが完了したら、その結果を次のタスクに渡して処理する場合を考えます。
今回は一つのタスクを例にしているので、同じ計算を複数回かけていきます。
Futureを使って、次のように書けます。
public class App {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executor = Executors.newFixedThreadPool(3);
List<Integer> valueList = Arrays.asList(1, 2, 3, 4, 5);
List<Future<Integer>> results = new ArrayList<>();
for (int value : valueList) {
results.add(executor.submit(new MyTask(value)));
}
List<Future<Integer>> results2 = new ArrayList<>();
for (Future<Integer> result : results) {
try {
int nextValue = result.get();
results2.add(executor.submit(new MyTask(nextValue)));
} catch (Exception e) {
throw new RuntimeException("1回目の実行で失敗");
}
}
List<Future<Integer>> results3 = new ArrayList<>();
for (Future<Integer> result : results2) {
try {
int nextValue = result.get();
results3.add(executor.submit(new MyTask(nextValue)));
} catch (Exception e) {
throw new RuntimeException("2回目の実行で失敗");
}
}
for (Future<Integer> result : results3) {
try {
int nextValue = result.get();
System.out.println(nextValue);
} catch (Exception e) {
throw new RuntimeException("3回目の実行で失敗");
}
}
}
}
// コンソールの出力：
// [start] value:2 thread-id:21
// [start] value:3 thread-id:22
// [start] value:1 thread-id:20
// [ end ] value:2 thread-id:21
// [ end ] value:1 thread-id:20
// [ end ] value:3 thread-id:22
// [start] value:4 thread-id:21
// [start] value:5 thread-id:22
// [ end ] value:4 thread-id:21
// [ end ] value:5 thread-id:22 ※1回目の実行ここまで
// [start] value:11 thread-id:21
// [ end ] value:11 thread-id:21
// [start] value:12 thread-id:22
// [ end ] value:12 thread-id:22
// [start] value:13 thread-id:20
// [start] value:14 thread-id:21
// [start] value:15 thread-id:22
// [ end ] value:13 thread-id:20
// [ end ] value:14 thread-id:21
// [ end ] value:15 thread-id:22 ※2回目の実行ここまで
// [start] value:21 thread-id:20
// [start] value:22 thread-id:21
// [start] value:23 thread-id:22
// [ end ] value:21 thread-id:20
// [ end ] value:22 thread-id:21
// [ end ] value:23 thread-id:22
// [start] value:24 thread-id:20
// 31
// 32
// 33
// [start] value:25 thread-id:21
// [ end ] value:24 thread-id:20
// [ end ] value:25 thread-id:21
// 34
// 35
依存関係（タスク１が終わったらタスク２を実行する）が読み取りにくいですね。
依存関係に関係ないコードが多く、やりたいことの本体が見えにくいです。
遅いタスクに足を引っ張られる
上記のコードだと、Future#getで待つことになります。
例えば、「1」「11」「21」の処理が10秒、それ以外が1秒かかるタスクだとしましょう。
本来、「1」の処理は合計で30秒、「2,3,4,5」の処理はそれぞれ合計3秒で終わるはずと言えます。
ただ、上記のコードだと1回目、2回目、3回目それぞれの実行で「1」を待つことになるため、「2,3,4,5」の処理も合計で30秒かかってしまうことになります。
CompletableFutureを使って解決する
CompletableFutureを使うと、タスク間の依存関係を流れるように実装できます。
依存関係の表現が簡単にでき、管理が楽
上記のコードをCompletableFutureを使って書いてみます
public class App {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executor = Executors.newFixedThreadPool(3);
List<Integer> valueList = Arrays.asList(1, 2, 3, 4, 5);
List<CompletableFuture<Integer>> results = valueList
.stream().map(value -> CompletableFuture.supplyAsync(wrap(new MyTask(value), 1), executor)
.thenCompose(
result1 -> CompletableFuture.supplyAsync(wrap(new MyTask(result1), 2), executor))
.thenCompose(
result2 -> CompletableFuture.supplyAsync(wrap(new MyTask(result2), 3), executor)))
.toList();
for (CompletableFuture<Integer> result : results) {
System.out.println(result.get());
}
}
private static <T> Supplier<T> wrap(Callable<T> task, int runCount) {
return () -> {
try {
return task.call();
} catch (Exception e) {
throw new RuntimeException(runCount + "回目の実行で失敗");
}
};
}
}
valueListをStreamで処理しています。各値に対して実行する内容がthenComposeで繋がっており依存関係が明確に記述できています。
今回はMyTaskの例外処理を各実行回ごとにやるため、wrapというメソッドを作っていますが、単純に「MyTaskを実行する、例外が発生したら何回目の実行で失敗したかわかる例外をスローする」としているだけです。
今回は、タスクが完了したらその結果を次のタスクに渡して次のタスクを実行する、ということをしたかったのでthenComposeを使っています。
CompletableFutureはタスクの依存関係を柔軟に表現するためのメソッドが充実しており、思いつく限りのパターンが網羅されていますので、一度ドキュメントを参照してください。
Java8 - CompletableFuture
他のタスクによらず処理を進められる
上記コードの実行結果は以下です。
// コンソールの出力：
// [start] value:1 thread-id:20
// [start] value:3 thread-id:22
// [start] value:2 thread-id:21
// [ end ] value:1 thread-id:20
// [ end ] value:2 thread-id:21
// [ end ] value:3 thread-id:22
// [start] value:5 thread-id:21
// [start] value:4 thread-id:22
// [start] value:12 thread-id:20 ※1回目の実行が終わっていなくても2回目の実行が始まっている
// [ end ] value:12 thread-id:20
// [ end ] value:5 thread-id:21
// [ end ] value:4 thread-id:22
// [start] value:11 thread-id:20
// [start] value:13 thread-id:21
// [ end ] value:11 thread-id:20
// [start] value:22 thread-id:22 ※2回目の実行が終わっていなくても3回目実行が始まっている
// [ end ] value:13 thread-id:21
// [start] value:14 thread-id:21
// [ end ] value:14 thread-id:21
// [start] value:15 thread-id:20
// [ end ] value:15 thread-id:20
// [ end ] value:22 thread-id:22
// [start] value:21 thread-id:21
// [start] value:23 thread-id:20
// [ end ] value:23 thread-id:20
// [start] value:25 thread-id:20
// [start] value:24 thread-id:22
// [ end ] value:21 thread-id:21
// [ end ] value:25 thread-id:20
// [ end ] value:24 thread-id:22
// 31
// 32
// 33
// 34
// 35
上記の通り、1回目の実行が全て終わらなくても、2回目の実行が始まっています。他のタスクの進捗によらず、タスクを進める事ができるため、「非同期」の実現には欠かせないものになっています。
おわりに
依存関係を表現するために「一つのクラスに全部書く」という方法も実はあります。
でもそれはJavaのオブジェクト型志向に反して、手続き型的な書き方になり、巨大なクラスと成り果てます。クラスの独立性や再利用性を著しく阻害する悪しきコードになると考えられます。
CompletableFutureを使うことで、並列処理を楽に実装できると同時に、クラスやプロジェクトの健康を保ってくれるものだと感じます。