CompletableFuture の組み合わせ: thenCombine, allOf, anyOf

1. CompletableFuture を組み合わせる

現実のアプリではすべてが 1 つのソースから来ることはまれです。ユーザープロフィールと注文を並行して読み込んだり、2 つのマイクロサービスからのデータを結合したり、時には最初に返ってきた応答だけを処理したいこともあります。同期的なやり方だと順番に待つ必要がありますが、CompletableFuture クラスなら同時に起動し、結果をエlegantに合成できます。そのための専用メソッドが thenCombine、allOf、anyOf です。順に見ていきましょう。

並列タスク: 2 つの非同期リクエスト

同期だとこうなります:

String name = loadUserName();    // 時間がかかる
int balance = loadUserBalance(); // 時間がかかる
System.out.println("名前: " + name + "、残高: " + balance);

問題点: 2 つ目の呼び出しは 1 つ目が終わってからでないと始まりません.

非同期のやり方

CompletableFuture を使えば両方のタスクを同時に開始できます:

CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> loadUserName());
CompletableFuture<Integer> balanceFuture = CompletableFuture.supplyAsync(() -> loadUserBalance());

では、両方の結果を取り出して一緒に処理するにはどうすればよいでしょうか？そのために thenCombine を使います。

2. thenCombine: 2 つのタスクの結果を結合する

メソッド thenCombine は 2 つの CompletableFuture を結合し、両タスクが完了したら処理を実行できます。結合結果を持つ新しい CompletableFuture を返します。

シグネチャ:

<A, B, C> CompletableFuture<C> thenCombine(
    CompletionStage<? extends B> other,
    BiFunction<? super A, ? super B, ? extends C> fn
)

• A — 最初の Future の結果型,
• B — 2 つ目の結果型,
• C — 結合結果の型.

例

CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> loadUserName());
CompletableFuture<Integer> balanceFuture = CompletableFuture.supplyAsync(() -> loadUserBalance());

CompletableFuture<String> resultFuture = nameFuture.thenCombine(
    balanceFuture,
    (name, balance) -> "名前: " + name + "、残高: " + balance
);

resultFuture.thenAccept(System.out::println);

仕組み:

• 両方の Future が並行に開始されます。
• 両方が完了すると、関数 (name, balance) -> ... が呼ばれます。
• 最終的な Future は結果の文字列を持ちます。

数値のミニ例

CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> 3);
CompletableFuture<Integer> sum = f1.thenCombine(f2, Integer::sum);

sum.thenAccept(result -> System.out.println("合計: " + result));

出力:

合計: 5

非同期版

結合処理が重い場合は thenCombineAsync を使います:

f1.thenCombineAsync(f2, (a, b) -> a * b);

3. allOf: タスクが多いとき

もしタスクが 2 つではなく 10 個あったらどうでしょう？たとえば 10 人分のユーザーデータを並列に読み込みたい場合です。 そのためのメソッドが CompletableFuture.allOf です。

説明

CompletableFuture.allOf(f1, f2, ..., fn) は、渡したタスクが すべて 完了したときに完了する新しい Future を返します。ただし注意点があります。この Future 自体は結果を持ちません — 型は常に CompletableFuture<Void> です。結果を得るには、元の Future から個別に「取り出す」必要があります。

例

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "一番目");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "二番目");

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2);

all.thenRun(() -> {
    // すべてのタスクが完了！
    String s1 = f1.join(); // join() — get() に似ているが、unchecked 例外をスローする
    String s2 = f2.join();
    System.out.println(s1 + " & " + s2);
});

出力:

一番目 & 二番目

タスク配列の例

List<CompletableFuture<String>> futures = new ArrayList<>();
for (int i = 0; i < 5; i++) {
    int id = i;
    futures.add(CompletableFuture.supplyAsync(() -> "ユーザー " + id));
}

CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

all.thenRun(() -> {
    for (CompletableFuture<String> f : futures) {
        System.out.println(f.join());
    }
});

何が起きるか:

• すべてのタスクが並行に開始されます。
• allOf は、すべてのタスクが完了したときに完了します。
• thenRun ブロック内で join() により結果を取得できます。

ビジュアル図

[Future1]   \
[Future2] ----> [allOf] ---> thenRun
[Future3]   /

4. anyOf: 最初に完了したタスクを待つ

ときには全件を待つのではなく、最も速く完了したタスクの結果だけが欲しいことがあります。たとえば 2 台のサーバーに同時に問い合わせて、先に応答した方のデータを使うといったケースです。 そのためのメソッドが CompletableFuture.anyOf です。

説明

CompletableFuture.anyOf(f1, f2, ..., fn) は、渡したタスクのうち いずれか が完了した時点で完了する Future を返します。結果の型は CompletableFuture<Object> です。これはそれぞれのタスクの型が異なり得るためです。

例

CompletableFuture<String> fast = CompletableFuture.supplyAsync(() -> {
    sleep(500);
    return "速いサーバー";
});
CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
    sleep(2000);
    return "遅いサーバー";
});

CompletableFuture<Object> any = CompletableFuture.anyOf(fast, slow);

any.thenAccept(result -> System.out.println("受け取った: " + result));

出力:

受け取った: 速いサーバー

異なる型の例
異なる型のタスクを組み合わせることもできますが、その場合の結果は Object 型になるため、明示的なキャストが必要です。

5. 役立つ注意点

推奨事項

• allOf は結果の配列を返しません。元の Future を保持しておき、join() や get() で値を取得する必要があります。
• anyOf は最初に完了した結果を返しますが、型は Object です。すべて同じ型であればキャストできます。
• allOf 内のいずれかのタスクがエラーで終了した場合、最終的な Future もエラーで終了します。
• thenCombine では両方のタスクが成功する必要があり、そうでない場合は例外になります。

メソッド比較表

thenCombine

allOf

Void

anyOf

6. CompletableFuture を組み合わせる際の典型的なミス

エラー №1: メインスレッドで get()/join() により結果を待ってしまう。
非同期コードを書いているのに、最後に結局 get() や join() を呼んでしまうと、スレッドをブロックして非同期性の利点を失います。ブロックせずに結果を処理するには thenAccept/thenRun を使うのが良いです。

エラー №2: allOf 使用時に元の Future の参照を保持しない。
CompletableFuture.allOf(f1, f2, f3) を呼んだのに、f1、f2、f3 を保存していないと、結果を取得できません。allOf が返すのは Void だけです！

エラー №3: チェーン内のエラーを処理しない。
いずれかのタスクがエラーで終わると、allOf や thenCombine 全体もエラーで終わります。例外を見逃さないように、エラー処理用メソッド（exceptionally、handle、whenComplete）を使いましょう。

エラー №4: anyOf で型が合わない。
anyOf は Object を返します。Future が異なる型を返す場合、最初に来たものが何かを見極める必要があります。可能なら同じ型にそろえるのが良いです。

エラー №5: コメントなしの複雑すぎるチェーン。
コード量が多くなると、Future のチェーンがスパゲッティ化しがちです。遠慮せずにチェーンを適切に変数へ分解し、各ステップにコメントを書きましょう。

1

タスク

JAVA 25 SELF, レベル 55, レッスン 2

ロック未解除

配送の調整: すべての荷物の到着を待つ

配送の調整: すべての荷物の到着を待つ

オープン

1

タスク

JAVA 25 SELF, レベル 55, レッスン 2

ロック未解除

挨拶の組み立て: フレーズの断片を結合する

挨拶の組み立て: フレーズの断片を結合する

オープン