CompletableFuture: 非同期実行と結果の合成

1. 概要

APIでは複数のサービスを呼び出す必要があることがよくあります。たとえば、顧客情報、注文、配送料の取得です。順番に呼び出すと、全体の時間は各呼び出しの遅延の合計になります。CompletableFuture を使うと、互いに独立した呼び出しを並列に実行し、すべて終わった時点で結果を組み合わせられます。

この記事を読むと、非同期タスクの作成方法、処理のつなぎ方（thenApply、thenCompose）、結果の組み合わせ方（thenCombine、allOf、anyOf）がわかります。さらに、エラー処理、タイムアウトの設定、非同期コードのテストの書き方も学べます。

2. サンプルの準備

ここでは3つのダミーのサービスを使います。各サービスは少し待機して、ネットワークの遅延を再現します:

import java.math.BigDecimal;
import java.util.List;

record Customer(long id, String name, String city) {}
record Order(long id, long customerId, BigDecimal total) {}

final class Delay {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }
}

class CustomerService {
    Customer findById(long id) {
        Delay.sleep(300);
        return new Customer(id, "Lan", "Hue");
    }
}

class OrderService {
    List<Order> findByCustomer(long customerId) {
        Delay.sleep(400);
        return List.of(
            new Order(1, customerId, new BigDecimal("150000")),
            new Order(2, customerId, new BigDecimal("90000")));
    }
}

class ShippingService {
    BigDecimal feeFor(String city) {
        Delay.sleep(200);
        return "Hue".equals(city) ? new BigDecimal("30000") : new BigDecimal("45000");
    }
}

これら3つのサービスは JDK だけを使っています。実行するために追加のライブラリは不要です。

3. 非同期タスクの作成

3.1. supplyAsync と runAsync

supplyAsync は Supplier を実行し、CompletableFuture<T> を返します。runAsync は結果を返さない Runnable を実行し、CompletableFuture<Void> を返します:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CreateDemo {
    public static void main(String[] args) {
        var customerService = new CustomerService();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CompletableFuture<Customer> customerFuture =
                CompletableFuture.supplyAsync(() -> customerService.findById(42L), executor);

            CompletableFuture<Void> auditFuture =
                CompletableFuture.runAsync(() -> System.out.println("アクセスログを記録"), executor);

            System.out.println("メインスレッドはブロックされない");
            System.out.println(customerFuture.join());
            auditFuture.join();
        }
    }
}

メインスレッドはブロックされない
アクセスログを記録
Customer[id=42, name=Lan, city=Hue]

最初の2行の順序は入れ替わることがあります。2つのタスクが別々のスレッドで動くためです。次の点に注意してください:

• 2番目の引数は Executor です。省略すると、タスクは ForkJoinPool.commonPool() 上で実行されます。
• Executors.newVirtualThreadPerTaskExecutor()（Java 21）はタスクごとに virtual thread を作成します。HTTP 呼び出しやデータベースへのクエリのような I/O 中心のタスクに向いています。
• ExecutorService は Java 19 から AutoCloseable です。try-with-resources ブロックは、タスクの完了を待ってから executor を閉じます。
• join() は結果が出るまで現在のスレッドをブロックします。タスクが失敗した場合、join() は CompletionException（非チェック例外）を投げます。一方 get() は ExecutionException（チェック例外）を投げます。

3.2. Future を手動で完了させる

ときには結果がラムダ式ではなくコールバックから届くこともあります。その場合は、空の future を作成して手動で完了させます:

CompletableFuture<String> paymentResult = new CompletableFuture<>();

// どこかで、例えば決済ゲートウェイのコールバック内で:
paymentResult.complete("PAID");

System.out.println(paymentResult.join()); // PAID

エラーを通知したい場合は completeExceptionally(new RuntimeException("カードが拒否されました")) を呼び出します。すでに値がある場合は、CompletableFuture.completedFuture(value) で完了済みの future を作るのが最も簡潔です。

4. 処理のつなぎ方

4.1. thenApply と thenAccept

thenApply は結果を変換します。Stream の map に似ています。thenAccept は結果を消費し、何も返しません:

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenApply(Customer::name)
    .thenApply(name -> "こんにちは、" + name);

greeting.thenAccept(System.out::println).join();

こんにちは、Lan

thenApply を呼び出すたびに新しい CompletableFuture が返されます。元の future は変更されません。

4.2. 次の処理も非同期の場合の thenCompose

注文を取得するには、前の処理で得た customerId が必要です。future を返す関数を thenApply に渡すと、結果は CompletableFuture<CompletableFuture<List<Order>>> になってしまいます。thenCompose はこの二重のネストを「平坦化」します。flatMap に似ています:

CompletableFuture<List<Order>> ordersFuture = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenCompose(customer -> CompletableFuture.supplyAsync(
        () -> orderService.findByCustomer(customer.id()), executor));

System.out.println(ordersFuture.join().size()); // 2

この2つの処理は互いに依存しているため、依然として順番に実行されます。全体の時間は約700msです。thenCompose によって速くなるわけではありません。スレッドをブロックせずに処理をつなげられる、という点に意味があります。

5. 結果の合成

互いに独立したタスクでこそ、CompletableFuture の真価が発揮されます。

5.1. 2つの future に対する thenCombine

顧客の都市がわかっていて、注文一覧と配送料を同時に取得したい場合を考えます:

long start = System.currentTimeMillis();

CompletableFuture<List<Order>> orders =
    CompletableFuture.supplyAsync(() -> orderService.findByCustomer(42L), executor);
CompletableFuture<BigDecimal> shippingFee =
    CompletableFuture.supplyAsync(() -> shippingService.feeFor("Hue"), executor);

CompletableFuture<BigDecimal> grandTotal = orders.thenCombine(shippingFee,
    (orderList, fee) -> orderList.stream()
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .add(fee));

System.out.println(grandTotal.join());
System.out.println("処理時間: " + (System.currentTimeMillis() - start) + " ms");

270000
処理時間: 405 ms

2つの呼び出しは並列に実行されます。全体の時間は最も遅い呼び出しとほぼ同じ（400ms）で、順番に呼び出した場合の600msではありません。thenCombine の2番目の引数の関数は、両方の future が完了したときにだけ実行されます。

5.2. 複数の future に対する allOf

タスクが2つより多い場合は allOf を使います。戻り値は CompletableFuture<Void> なので、各子 future の結果は自分で取り出す必要があります:

List<String> cities = List.of("Hue", "Hanoi", "Danang");

List<CompletableFuture<BigDecimal>> fees = cities.stream()
    .map(city -> CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor))
    .toList();

CompletableFuture<List<BigDecimal>> allFees = CompletableFuture
    .allOf(fees.toArray(CompletableFuture[]::new))
    .thenApply(ignored -> fees.stream()
        .map(CompletableFuture::join)
        .toList());

System.out.println(allFees.join());

[30000, 45000, 45000]

thenApply の中では、すべての子 future がすでに完了しています。そのため join() はブロックせずにすぐ戻ります。結果の順序は cities の順序と同じに保たれます。

5.3. anyOf で最速の結果を取得する

anyOf は子 future のいずれか1つが完了した時点で完了します。同じサービスのレプリカを複数呼び出すときに便利です:

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> { Delay.sleep(500); return "replica-A"; }, executor),
    CompletableFuture.supplyAsync(() -> { Delay.sleep(100); return "replica-B"; }, executor));

System.out.println(fastest.join()); // replica-B

戻り値の型は CompletableFuture<Object> なので、自分で型キャストする必要があります。また、最速の future が例外で終わった場合、anyOf もその例外で終わることに注意してください。

6. エラー処理とタイムアウト

6.1. exceptionally と handle

どの処理で発生した例外も、後続の処理へ伝播します。その間の処理はスキップされ、エラーを処理するステップに到達するまで実行されません:

CompletableFuture<BigDecimal> fee = CompletableFuture
    .supplyAsync(() -> {
        throw new IllegalStateException("Shipping service down");
    }, executor)
    .thenApply(value -> (BigDecimal) value)          // スキップされる
    .exceptionally(ex -> {
        System.out.println("エラー: " + ex.getMessage());
        return new BigDecimal("50000");              // フォールバック値
    });

System.out.println(fee.join());

エラー: java.lang.IllegalStateException: Shipping service down
50000

引数 ex は通常、元の例外をラップした CompletionException です。そのため、メッセージにはクラス名の接頭辞が付きます。実際のエラーを取り出すには ex.getCause() を使ってください。

handle は結果とエラーの両方を受け取ります。タスクが成功しても失敗しても、必ず呼び出されます:

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> orderService.findByCustomer(42L), executor)
    .handle((orderList, ex) -> ex == null
        ? "注文 " + orderList.size() + " 件あり"
        : "注文の取得に失敗しました");

ほかに whenComplete もあります。こちらも結果とエラーを受け取りますが、結果は変更しません。ログの記録に向いたメソッドです。

6.2. orTimeout と completeOnTimeout

外部サービスを無期限に待つことは絶対に避けてください。Java 9 以降には、このための2つのメソッドが用意されています:

import java.util.concurrent.TimeUnit;

CompletableFuture<BigDecimal> withDefault = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .completeOnTimeout(new BigDecimal("45000"), 500, TimeUnit.MILLISECONDS);

System.out.println(withDefault.join()); // 約500ms後に 45000

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join() は cause が TimeoutException の CompletionException を投げる

completeOnTimeout は時間切れのとき既定値を返します。orTimeout は future を TimeoutException で終了させます。

注意: タイムアウトが完了させるのは future だけです。内部のタスクは停止しません。スレッドは2秒間スリープを続けます。cancel(true) も実行中のスレッドを interrupt しません。本当にキャンセルしたい場合は、タスク内でキャンセルフラグを自分で確認するか、HTTP クライアント自体のタイムアウトを使ってください。

7. テストを書く

Spring Boot を使っている場合、spring-boot-starter-test には JUnit 5 と AssertJ が含まれています。純粋な Java プロジェクトの場合は、次の依存関係を追加してください:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.26.3</version>
    <scope>test</scope>
</dependency>

まず、テストしやすいようにロジックを1つのクラスにまとめます:

class CheckoutService {
    private final OrderService orderService;
    private final ShippingService shippingService;
    private final Executor executor;

    CheckoutService(OrderService o, ShippingService s, Executor executor) {
        this.orderService = o;
        this.shippingService = s;
        this.executor = executor;
    }

    CompletableFuture<BigDecimal> grandTotal(long customerId, String city) {
        var orders = CompletableFuture.supplyAsync(() -> orderService.findByCustomer(customerId), executor);
        var fee = CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor);
        return orders.thenCombine(fee, (list, f) -> list.stream()
            .map(Order::total)
            .reduce(BigDecimal.ZERO, BigDecimal::add)
            .add(f));
    }
}

次に、成功した場合と失敗した場合の両方のテストを書きます:

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import org.junit.jupiter.api.Test;

class CheckoutServiceTest {

    @Test
    void shouldCombineOrdersAndShippingFee() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), new ShippingService(), executor);

            assertThat(service.grandTotal(42L, "Hue"))
                .succeedsWithin(Duration.ofSeconds(2))
                .isEqualTo(new BigDecimal("270000"));
        }
    }

    @Test
    void shouldPropagateShippingFailure() {
        ShippingService broken = new ShippingService() {
            @Override
            BigDecimal feeFor(String city) {
                throw new IllegalStateException("down");
            }
        };

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), broken, executor);

            assertThatThrownBy(() -> service.grandTotal(42L, "Hue").join())
                .isInstanceOf(CompletionException.class)
                .hasCauseInstanceOf(IllegalStateException.class);
        }
    }
}

succeedsWithin は最大2秒待ってから値を検証します。これにより、コードに問題があってもテストが止まってしまうことはありません。2番目のテストでは、ShippingService からのエラーが thenCombine を通って伝播し、CompletionException でラップされます。

テストでは Runnable::run を Executor として渡すこともできます。そうすると、すべての処理がテストのスレッド上で同期的に実行されます。テストが速くなり、結果も安定します。

8. よくある間違い

• I/O を commonPool 上で実行する。 このプールのスレッド数は CPU 数から 1 を引いた数で、アプリケーション全体（parallel stream を含む）で共有されます。遅い HTTP 呼び出しがいくつかあるだけで、プール全体が詰まることがあります。I/O には専用の executor を渡してください。できれば virtual thread が最適です。
• join() を早く呼びすぎる。 supplyAsync(a).join() を実行してから supplyAsync(b).join() を実行すると、依然として順次実行になります。先にすべての future を起動してから、組み合わせるか待つ必要があります。
• 例外を握りつぶす。 join()、get() のどれも呼ばず、exceptionally/whenComplete も付けなければ、エラーは痕跡を残さずに消えてしまいます。
• thenApply と thenApplyAsync を混同する。 thenApply は、直前の処理を完了させたスレッド上で実行されることがあります。future がすでに完了していれば、呼び出し元のスレッド上で実行されます。次の処理が重い、またはブロッキングする場合は、...Async(fn, executor) のバリアントを使ってください。
• タイムアウトを設定しない。 あるサービスが応答しなくなると、あなたのリクエストも一緒に止まってしまいます。
• プログラムが早く終了してしまう。 commonPool のスレッドは daemon スレッドです。タスクが終わる前に main が終了すると、結果は決して出力されません。

注意: 呼び出しが1つだけで、すぐに結果を待つ場合は CompletableFuture を使うべきではありません。Java 21 以降、virtual thread 上で動くコードでは、通常の同期的なコードのほうが性能を落とさずに読みやすいことが多いです。CompletableFuture が本当に役立つのは、互いに独立した複数のタスクを並列に実行する必要がある場合や、ブロックしないパイプラインが必要な場合だけです。

9. まとめ

CompletableFuture を使うと、スレッドを自分で管理することなく、独立したタスクを並列に実行して結果を組み合わせられます。要点をまとめます:

• タスクの作成: supplyAsync/runAsync。適切な executor を必ず渡してください。
• 処理のつなぎ: 値を変換する場合は thenApply、次の処理が future を返す場合は thenCompose。
• 合成: 2つの future には thenCombine、複数の future には allOf、最速の結果を得るには anyOf。
• エラーとタイムアウト: exceptionally、handle、orTimeout、completeOnTimeout。

テストでは、AssertJ の succeedsWithin を使うと、テストが止まる心配をせずに結果を検証できます。