CompletableFuture 异步编排

一个接口要聚合用户信息、订单列表、积分余额三个下游服务，串行调用要 300 ms，并行只要 100 ms 出头。Future 能并行，但只能阻塞 get()，无法表达依赖关系和回调。CompletableFuture 是 JDK 8 给出的答案：它既是一个 Future，也是一个可组合的 CompletionStage。本文讲清常用 API 的语义差异、编排模式，以及最容易被忽视的线程池与异常问题。

一、两种创建方式与线程池陷阱

// 1. 使用默认线程池 ForkJoinPool.commonPool()
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "hello");

// 2. 指定自定义线程池（推荐）
ExecutorService pool = Executors.newFixedThreadPool(8);
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "hello", pool);

默认的 commonPool 并行度为 CPU 核数 - 1，而且是整个 JVM 共享的。只要有一个任务阻塞（比如远程调用、数据库查询），就会拖慢所有使用默认池的代码。IO 密集型任务必须传入自定义线程池。

// 反例：IO 任务用 commonPool，可能拖垮全 JVM
CompletableFuture.supplyAsync(() -> httpCall());   // 不要这样写

runAsync 接收 Runnable，没有返回值；supplyAsync 接收 Supplier，有返回值。这两组方法都各有带 Executor 参数的重载，务必使用带池的版本。

还有一个容易忽略的差异：new CompletableFuture<>() 创建的是一个未完成的 future，需要手动 complete()；而 completedFuture(value) 直接返回已完成状态，适合构造常量或测试桩。

CompletableFuture<String> manual = new CompletableFuture<>();
// 另一个线程稍后完成它
executor.execute(() -> manual.complete("done"));
String v = manual.get(1, TimeUnit.SECONDS);

二、四类编排 API 的语义

CompletableFuture 的方法名有规律可循，读懂命名就懂了一半：

thenApply 和 thenCompose 最容易混淆：

// thenApply：结果是 CompletableFuture<CompletableFuture<String>>，嵌套了
CompletableFuture<CompletableFuture<String>> nested =
        CompletableFuture.supplyAsync(() -> "user")
                         .thenApply(u -> CompletableFuture.supplyAsync(() -> u + "-detail"));

// thenCompose：扁平化，结果是 CompletableFuture<String>
CompletableFuture<String> flat =
        CompletableFuture.supplyAsync(() -> "user")
                         .thenCompose(u -> CompletableFuture.supplyAsync(() -> u + "-detail"));

规则：映射函数返回普通值时用 thenApply，返回 CompletableFuture 时用 thenCompose。这与 Stream 中 map 和 flatMap 的关系完全一致。

thenApply 与 thenApplyAsync 的区别不在语义而在执行线程：不带 Async 时，如果 future 已完成，回调直接在调用 thenApply 的线程上同步执行；如果尚未完成，则由完成它的线程执行。带 Async 时总是提交到线程池。这意味着同一段代码在不同时序下可能跑在完全不同的线程上，排查线程相关问题时必须考虑这一点。

thenCombine 用于两个独立任务的汇合，不需要先 allOf：

CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> queryName(id), pool);
CompletableFuture<Integer> b = CompletableFuture.supplyAsync(() -> queryAge(id), pool);

CompletableFuture<String> merged =
        a.thenCombine(b, (name, age) -> name + ":" + age);

它比 allOf 加手动 join 更紧凑，适合只有两个依赖的场景；三个以上用 allOf 更清晰。

三、并行聚合的实战写法

假设要并行查三个服务，全部完成后组装结果：

import java.util.concurrent.*;
import java.util.function.Supplier;

public class Aggregator {
    private final ExecutorService pool = Executors.newFixedThreadPool(16);

    public UserProfile load(long userId) throws Exception {
        CompletableFuture<User> userF =
                CompletableFuture.supplyAsync(() -> queryUser(userId), pool);
        CompletableFuture<Orders> ordersF =
                CompletableFuture.supplyAsync(() -> queryOrders(userId), pool);
        CompletableFuture<Points> pointsF =
                CompletableFuture.supplyAsync(() -> queryPoints(userId), pool);

        // 全部完成后再组装，allOf 的返回值是 CompletableFuture<Void>
        CompletableFuture<Void> all = CompletableFuture.allOf(userF, ordersF, pointsF);

        return all.thenApply(v -> {
            // allOf 不直接给结果，需要从各 future 里取
            return new UserProfile(userF.join(), ordersF.join(), pointsF.join());
        }).get(2, TimeUnit.SECONDS);
    }

    private User queryUser(long id)     { return new User(id); }
    private Orders queryOrders(long id) { return new Orders(id); }
    private Points queryPoints(long id) { return new Points(id); }

    record User(long id) {}
    record Orders(long id) {}
    record Points(long id) {}
}

几个要点：

• allOf 的返回类型是 CompletableFuture<Void>，它只表示"都完成了"，不携带结果，需要自己从各个 future 中 join() 取。
• join() 和 get() 的区别是 join() 不抛受检异常，失败时抛 CompletionException。在 thenApply 内部使用 join() 更简洁。
• 阻塞式 get(timeout) 必须设超时，否则下游卡住会拖死整个接口。

如果只需要最快的那个结果，用 anyOf：

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
        CompletableFuture.supplyAsync(() -> queryFromA(), pool),
        CompletableFuture.supplyAsync(() -> queryFromB(), pool)
);
Object result = fastest.get(500, TimeUnit.MILLISECONDS);

anyOf 返回 Object，类型信息丢失，需要自己强转，使用时注意类型安全。

四、异常处理的三套机制

CompletableFuture 的异常传播规则是：异常会沿着 then 链向后传递，跳过所有正常回调，直到遇到 exceptionally 或 handle。

CompletableFuture.supplyAsync(() -> {
            if (Math.random() > 0.5) throw new IllegalStateException("boom");
            return 42;
        }, pool)
        .thenApply(v -> v * 2)                       // 异常时被跳过
        .thenApply(v -> v + 1)                       // 异常时被跳过
        .exceptionally(ex -> {
            log.warn("fallback", ex);
            return -1;                               // 提供降级值
        })
        .thenAccept(v -> System.out.println("result=" + v));

三种机制的区别：

// whenComplete 不能把异常"吃掉"，异常仍会向下传递
future.whenComplete((r, ex) -> {
    if (ex != null) log.error("failed", ex);
});   // 返回的 future 仍然是异常状态

// handle 可以吞掉异常并恢复正常
future.handle((r, ex) -> ex != null ? defaultValue : r);

一个常见误区是以为 whenComplete 能兜底。它只做通知，不能改变完成状态。

注意：如果异常一直未被处理，最终会在调用 get()/join() 时抛出，或者被丢弃。CompletableFuture 不会主动打印未处理异常，这是排查问题时容易踩的坑。

五、超时控制

JDK 8 没有内置的 orTimeout，需要自己组合。JDK 9 起提供了两个方法：

// JDK 9+：超时后以 TimeoutException 完成
future.orTimeout(1, TimeUnit.SECONDS);

// JDK 9+：超时后以指定值完成，不抛异常
future.completeOnTimeout(defaultValue, 1, TimeUnit.SECONDS);

JDK 8 的等价写法：

public static <T> CompletableFuture<T> withTimeout(
        CompletableFuture<T> future, long timeout, TimeUnit unit,
        ScheduledExecutorService scheduler) {
    CompletableFuture<T> result = new CompletableFuture<>();
    ScheduledFuture<?> timer = scheduler.schedule(
            () -> result.completeExceptionally(new TimeoutException()),
            timeout, unit);
    future.whenComplete((v, ex) -> {
        timer.cancel(false);
        if (ex != null) result.completeExceptionally(ex);
        else result.complete(v);
    });
    return result;
}

注意 orTimeout 只是让 future 以超时状态完成，不会中断底层任务。底层任务仍在跑，可能继续占用线程、继续打下游。要真正取消，需要把底层调用做成可中断的，并调用 cancel(true)。

六、并发场景分析

线程池耗尽。编排链中每个 thenApplyAsync 都可能提交新任务到同一个池。如果池大小固定且任务互相等待，就会死锁。

// 危险：同一池内嵌套提交并等待
CompletableFuture.supplyAsync(() ->
        CompletableFuture.supplyAsync(() -> slowCall(), pool).join(), pool);

外层任务占着线程等内层，内层排不上队，池满即死锁。解决方式：IO 任务和 CPU 任务分池，或者避免在池内任务中阻塞等待同池任务。

不带 Async 的回调在哪个线程执行。如果 future 完成时调用线程正好在设置回调，回调就在该线程执行；否则由完成 future 的线程执行。这意味着 thenApply 可能跑在 IO 线程上，耗时操作会阻塞它。涉及阻塞或耗时的回调一律用 thenApplyAsync(..., pool) 显式指定线程池。

上下文丢失。ThreadLocal（如 MDC、用户上下文）不会自动传递到异步线程。需要在提交时捕获、在回调中恢复：

String traceId = MDC.get("traceId");
CompletableFuture.supplyAsync(() -> {
    MDC.put("traceId", traceId);
    try {
        return doWork();
    } finally {
        MDC.remove("traceId");
    }
}, pool);

Spring 的 TaskDecorator 或阿里 TransmittableThreadLocal 可以自动化这一步。

七、常见坑 / 最佳实践

1. IO 任务必须传自定义线程池，不要用 commonPool。默认池只有 CPU-1 个线程且全局共享，一个阻塞任务会拖垮全 JVM。
2. thenApply 与 thenCompose 别混用。返回 CompletableFuture 时用 thenCompose，否则得到嵌套结构。
3. allOf 不返回结果列表，需要自己 join() 各子 future，且要处理其中某个失败的情况。
4. 必须设置超时。用 orTimeout（JDK 9+）或自定义定时器，并注意超时不会中断底层任务。
5. whenComplete 不能兜底异常，要恢复默认值用 handle 或 exceptionally。
6. 不要在异步任务中阻塞等待同一个池的任务，会造成线程池死锁。
7. 显式传递上下文（traceId、用户身份），ThreadLocal 不会自动跨线程。

八、小结

• CompletableFuture 同时是 Future 和 CompletionStage，支持链式编排与回调。
• supplyAsync/runAsync 必须指定自定义线程池，commonPool 只适合纯 CPU 的短任务。
• thenApply 用于值转换，thenCompose 用于扁平化嵌套 future，对应 map 与 flatMap。
• allOf 等待全部完成但不携带结果，anyOf 取最快结果但返回 Object。
• 异常沿链传播，exceptionally/handle 可恢复，whenComplete 只通知不改变状态。
• 超时、线程池隔离、上下文传递是生产环境必须处理的三件事。