CompletableFuture：异步运行和结果合并

1. 总体概述

一个 API 通常需要调用多个服务：获取客户信息、订单信息、运费。如果依次调用，总时间等于每次调用延迟的总和。CompletableFuture 允许你并行运行独立的调用，然后在全部完成时合并结果。

读完这篇文章，你将学会如何创建异步任务、连接处理步骤（thenApply、thenCompose）和合并结果（thenCombine、allOf、anyOf）。你也会学到如何处理错误、设置超时和为异步代码编写测试。

2. 准备示例

我们使用三个模拟服务。每个服务都会“睡眠”一段时间来模拟网络延迟：

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");
    }
}

这三个服务只使用 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]

前两行的顺序可能互换，因为它们在不同的线程上运行。你需要注意以下几点：

• 第二个参数是 Executor。如果省略，任务将在 ForkJoinPool.commonPool() 上运行。
• Executors.newVirtualThreadPerTaskExecutor()（Java 21）为每个任务创建一个虚拟线程。这个选择适合 I/O 任务，如 HTTP 调用或数据库查询。
• ExecutorService 从 Java 19 开始是 AutoCloseable 的。try-with-resources 块将等待任务完成后才关闭 executor。
• join() 阻塞当前线程直到有结果。如果任务出错，join() 抛出 CompletionException（未检查异常）。而 get() 抛出 ExecutionException（检查异常）。

3.2. 手动完成一个 future

有时结果来自回调而不是 lambda。此时你创建一个空 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

这两步相互依赖，所以仍然按顺序运行。总时间约为 700 毫秒。thenCompose 不会加速。它帮助你在不阻塞任何线程的情况下连接步骤。

5. 合并结果

对于独立任务，CompletableFuture 才真正发挥作用。

5.1. 对两个 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) + " 毫秒");

270000
耗时 405 毫秒

两个调用并行运行。总时间大约等于最慢的调用（400 毫秒），而不是按顺序调用的 600 毫秒。thenCombine 的第二个参数中的函数只在两个 future 都完成时才运行。

5.2. 对多个 future 使用 allOf

当有超过两个任务时，你使用 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 完成时立即完成。当你调用同一服务的多个副本时，这很有用：

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+ 为此提供了两个方法：

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()); // 45000 后约 500 毫秒

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join() 抛出 CompletionException，其原因是 TimeoutException

completeOnTimeout 在超时时返回默认值。orTimeout 以 TimeoutException 结束 future。

注意： 超时仅完成 future。它不会停止底层任务。线程仍会继续“睡眠” 2 秒。cancel(true) 也不会中断正在运行的线程。如果需要真正取消，你必须在任务中自己检查取消标志，或使用 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>

首先，将逻辑提取到一个类中以便于测试：

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 秒然后检查值。这样如果代码出错，测试不会挂起。在第二个测试中，来自 ShippingService 的错误传播通过 thenCombine 并被包装在 CompletionException 中。

在测试中，你可以传递 Runnable::run 作为 Executor，使所有内容在测试线程上同步运行。这有助于加快测试速度和稳定结果。

8. 常见错误

• 在 commonPool 上运行 I/O。 这个池的线程数等于 CPU 数减 1，并被整个应用共享（包括 parallel stream）。几个缓慢的 HTTP 调用可能会阻塞整个应用。对于 I/O，你应该传递专用的 executor，最好是虚拟线程。
• 过早调用 join()。 写 supplyAsync(a).join() 然后再 supplyAsync(b).join() 仍然是顺序的。你需要先启动所有 future，然后再合并或等待。
• 吞掉异常。 如果没人调用 join()、get() 或附加 exceptionally/whenComplete，错误将无声地消失。
• 混淆 thenApply 和 thenApplyAsync。 thenApply 可能在完成前一步的线程上运行，或如果 future 已完成则在调用线程上运行。如果下一步很重或阻塞，使用 ...Async(fn, executor) 变体。
• 不设置超时。 一个挂起的服务会导致你的请求也挂起。
• 程序提前退出。 commonPool 的线程是守护线程。如果 main 在任务完成前结束，结果永远不会被打印出来。

注意： 当只有单一调用且你立即等待结果时，不应使用 CompletableFuture。从 Java 21 开始，对于在虚拟线程上运行的代码，通常同步编写既更易读又不失性能。CompletableFuture 仅在需要并行运行多个独立任务，或需要非阻塞管道时才值得使用。

9. 结论

CompletableFuture 帮助你并行运行独立任务并合并结果，而无需自己管理线程。快速总结：

• 创建任务：supplyAsync/runAsync，始终传递合适的 executor。
• 连接步骤：当转换值时用 thenApply，当下一步返回 future 时用 thenCompose。
• 合并：对两个 future 用 thenCombine，对多个 future 用 allOf，用 anyOf 获取最快的结果。
• 错误和超时：exceptionally、handle、orTimeout、completeOnTimeout。

在测试中，使用 AssertJ 的 succeedsWithin 检查结果，而无需担心测试挂起。