异步编程：从 `Future` 到 `CompletableFuture`

有了线程池，也有了 Future，为什么 Java 8 还要再加一个 CompletableFuture？异步链条上，任务到底跑在哪条线程上？异常又是从哪一段冒出来的？

Future 提供异步结果的查询能力，但缺少回调与组合接口；Java 8 引入的 CompletableFuture 将结果转换、阶段组合和异常处理纳入同一套 API。这一章聚焦两者的语义边界、线程归属和最容易踩的坑；响应式流只与本专题的 NIO、Netty建立联系，Actor 模型不作为本专题的主线内容。

1. `Future` 的三个主要局限

1.1 有结果，但只能靠阻塞取

Future 是 Java 5 引入的第一代异步结果承载：

java

ExecutorService pool = Executors.newFixedThreadPool(4);
Future<String> f = pool.submit(() -> queryDB());

String r = f.get();     // 主线程在这里被阻塞

submit 返回得很快，但 get() 必须阻塞——线程要么盯着结果原地等，要么就得自己写 isDone() 轮询。"异步执行 + 同步等待"这条路走下去，异步只是名义上的异步。

1.2 没有回调通道

Future 上没有一个能挂"结果就绪后自动做点什么"的钩子。想让下一步逻辑在结果就绪时被"推"到手上，只有两条歪路：

循环 isDone() + sleep——把 CPU 烧完
起一条新线程调 get()——把线程池耗完

不能被通知，就没法把多个异步操作拼成一条流水线。

1.3 组合的成本是嵌套

想表达"A 完成后基于 A 的结果做 B、再基于 B 做 C"，用 Future 只能这样写：

java

Future<String> a = pool.submit(() -> queryA());
Future<String> b = pool.submit(() -> {
    String aResult = a.get();            // 在池内线程上阻塞
    return queryB(aResult);
});
Future<String> c = pool.submit(() -> {
    String bResult = b.get();            // 又阻塞一次
    return queryC(bResult);
});

问题成堆：池内线程被 get() 卡住（可能死锁）、代码嵌套逐层加深、异常传播需要手工写 try/catch/ExecutionException 每一层。

三个短板加起来的结论：Future 只解决了"能拿到异步返回值"，没解决"用得起来"。CompletableFuture 就是补这三个洞。

2. `CompletableFuture` 的两组 API

2.1 API 全景：两类常见操作与三类独立能力

CompletableFuture 方法数看着吓人，但可以从单个 Future 上做转换 和 多个 Future 之间做合并两类常见操作入手，再单独掌握执行器选择、异常处理与完成控制。

2.2 转换：单输入 → 单输出

方法	输入	输出	语义
`thenApply(fn)`	`T → R`	`CF<R>`	同步转换
`thenCompose(fn)`	`T → CF<R>`	`CF<R>`	扁平化异步（避免 `CF<CF<R>>`）
`thenAccept(cons)`	`T → void`	`CF<Void>`	消费结果
`thenRun(runnable)`	忽略结果	`CF<Void>`	只关心"完成了"

关键区别：thenApply 用于"结果 → 新值"，thenCompose 用于"结果 → 新的异步任务"。对应函数式语言里的 map 和 flatMap。混用会得到嵌套的 CompletableFuture<CompletableFuture<R>>——一层多余。thenCompose 负责展开异步依赖，不负责切换执行线程；需要显式指定后续阶段的执行器时，应使用相应的 thenComposeAsync 重载。

java

// ❌ 用 thenApply 处理返回 CF 的函数，得到嵌套
CompletableFuture<CompletableFuture<Order>> bad =
    findUser(id).thenApply(user -> findOrder(user));    // findOrder 返回 CF

// ✅ 用 thenCompose 扁平化
CompletableFuture<Order> good =
    findUser(id).thenCompose(user -> findOrder(user));

2.3 合并：多个输入 → 单输出

方法	输入	语义
`thenCombine(other, bi)`	两个 CF	都完成后合并两个结果
`allOf(cf...)`	N 个 CF	全部完成后触发（返回 `CF<Void>`）
`anyOf(cf...)`	N 个 CF	任一完成后触发（返回 `CF<Object>`）

allOf 表示等待所有输入阶段完成，不表示所有任务都成功。任一输入阶段异常完成时，聚合阶段也会异常完成，其他输入阶段不会因此自动取消。anyOf 返回 CF<Object> 是因为不同输入阶段可能具有不同的结果类型；它在任意一个输入阶段完成时完成，包括以异常方式完成的情况，因此是"任一完成"而非"任一成功"。

allOf 返回 Void——想拿到各个 CF 的结果，还得在完成后手动 join 每一个。这是初学者最容易忽视的一处：

java

CompletableFuture<User>  fu = findUser(id);
CompletableFuture<Order> fo = findOrder(id);
CompletableFuture<Cart>  fc = findCart(id);

// ❌ allOf 只告诉你"全部完成"，不给你结果
CompletableFuture<Void> all = CompletableFuture.allOf(fu, fo, fc);
all.join();     // 拿不到 user / order / cart

// ✅ 完成后自己 join 每个
CompletableFuture<Profile> profile = all.thenApply(v ->
    new Profile(fu.join(), fo.join(), fc.join())    // 此时都已完成，join 立即返回
);

2.4 一条真实的异步流水线

用户查询 → 订单列表 → 每单查物流 → 汇总：

java

CompletableFuture<String> pipeline =
    findUser(id)                                    // 第 1 步
        .thenCompose(u -> findOrders(u))            // 第 2 步：等第 1 步
        .thenCompose(orders -> {                    // 第 3 步：并行查物流
            List<CompletableFuture<String>> fs = orders.stream()
                .map(o -> findTracking(o))
                .toList();
            return CompletableFuture
                .allOf(fs.toArray(new CompletableFuture[0]))
                .thenApply(v -> fs.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.joining(", ")));
        });

String result = pipeline.join();

三步的时间线：

txt

时间 →
[findUser 100ms] → [findOrders 150ms] → [findTracking 并行 80ms]
                                          ├─ ORD-001 80ms
                                          └─ ORD-002 80ms
总耗时 ≈ 330ms（串行需要 100 + 150 + 80 × 2 = 410ms）

thenCompose 用于串行依赖，allOf + join 用于扇出并行——两者组合就能表达大部分业务流水线。

3. 执行线程之谜

thenApply、thenApplyAsync、传 Executor 与不传 Executor——到底跑在哪条线程上是 CompletableFuture 最让人迷惑的一处。

3.1 三种触发方式

写法	执行线程
`thenApply(fn)`	不指定执行器。回调可能在完成上游阶段的线程执行；若上游已完成，也可能在注册回调的线程执行
`thenApplyAsync(fn)`	默认异步执行器，OpenJDK 21 常规配置下为 `ForkJoinPool.commonPool()`
`thenApplyAsync(fn, executor)`	提交到指定的 `executor`

同步版本 thenApply 的"急切执行"：如果上一个 CF 在调用 thenApply 时已经完成，回调立即在当前线程运行——不再是异步。这是很多"看起来应该异步、实际在业务线程上跑了阻塞任务"的根源。

3.2 `commonPool` 的默认坑

supplyAsync(fn) / thenApplyAsync(fn) 不传 executor 时，默认使用 ForkJoinPool.commonPool()（第 11 章的 ForkJoinPool 一节讨论过）。这个池由 JVM 中的任务共享；默认并行度是 max(1, 可用处理器数 - 1)，也可由 java.util.concurrent.ForkJoinPool.common.parallelism 覆盖。在 OpenJDK 21 中，当公共池并行度低于 2 时，CompletableFuture 的默认异步执行路径会回退到按任务创建线程的执行器，而不是依靠公共池的常规工作线程调度；显式传入 Executor 的重载不使用这条默认回退路径。

java

// ❌ 阻塞 IO 塞进 commonPool：同池任务的延迟会互相放大
CompletableFuture.supplyAsync(() -> httpClient.get(url));

// ✅ 阻塞任务用独立线程池
ExecutorService ioPool = new ThreadPoolExecutor(
    20, 40, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(200),
    new NamedThreadFactory("io"),
    new ThreadPoolExecutor.CallerRunsPolicy());

CompletableFuture.supplyAsync(() -> httpClient.get(url), ioPool);

默认建议：阻塞 I/O、任务时长差异大或有独立服务目标的 stage 使用专用 Executor。短小、无阻塞且可接受共享池调度的任务，可以显式接受 commonPool 的默认行为；是否共享应由压测和隔离目标决定。

3.3 `Async` 变体的选型

看到方法名带 Async 后缀就要问自己两个问题：

上一个 stage 的执行线程是否适合承担这一步？ 不适合就用 Async 换线程
需要指定线程池吗？ 需要就传 executor；不传时遵循默认执行器规则

java

// 假设 dbPool 用于数据库、cpuPool 用于计算
CompletableFuture
    .supplyAsync(() -> queryDB(id), dbPool)                // 明确用 dbPool
    .thenApplyAsync(this::heavyCompute, cpuPool)           // 换到 cpuPool 计算
    .thenAccept(this::logResult);                          // 同步：谁完成 heavyCompute 就谁记日志

这种“每一步都在合适的池上”的写法便于控制线程池边界，适合任务类型差异明显的生产调用链；阶段少且执行很快时，也可通过明确注释接受默认执行器。

4. 异常传播的三条路径

4.1 传播规则

异常沿着链条向后透传：链条中任何一步失败，后续 stage 会跳过正常回调，直到一个能够观察或恢复异常的方法。exceptionally 和 handle 能产生替代结果；whenComplete 只观察结果和异常，不恢复异常。

4.2 三个方法的差异

方法	能感知异常	能替换结果	触发时机
`exceptionally(fn)`	✅	✅ 异常时给出替代值	只在上游异常时执行
`handle(bi)`	✅	✅ 正常和异常都能替换	无论上游正常还是异常都执行
`whenComplete(bi)`	✅	❌ 只做副作用，不改变结果	正常和异常都执行

whenComplete 用于观察结果、记录日志或执行清理动作，它不会主动恢复上游异常；但回调本身也应尽量避免抛出异常。上游正常完成而 whenComplete 回调抛出异常时，返回阶段会异常完成；上游已经异常完成时，原异常优先传播。

java

future
    .thenApply(this::riskyStep)
    // 只关心异常：给个默认值
    .exceptionally(ex -> {
        log.warn("failed, use default", ex);
        return "default";
    });

future
    .thenApply(this::riskyStep)
    // 统一处理正常/异常：能改结果
    .handle((result, ex) -> {
        if (ex != null) return "fallback";
        return result.toUpperCase();
    });

future
    .thenApply(this::riskyStep)
    // 打日志 / 释放资源，不改变结果
    .whenComplete((result, ex) -> {
        if (ex != null) log.error("failed", ex);
        span.finish();
    });

4.3 异常在链条中的形状

异常传播路径上，异常会被包装成 CompletionException（get() 时是 ExecutionException）：

txt

supplyAsync ──▶ thenApply ──▶ thenApply ──▶ ...
     │                                       │
     │       抛 IOException                  │
     └──────────┬──────────────────────────┘
                │
                ▼
        包装成 CompletionException
                │
                ▼
        跳过后续 thenApply，直到 exceptionally / handle

CompletableFuture 会通过异常完成状态在依赖链中传播失败。对于执行过程中抛出的异常，依赖阶段和 join() 常见的观察形式是 CompletionException；调用 get() 时，执行失败通常通过 ExecutionException 暴露。异常的包装层次取决于具体的完成路径，排查时应检查 getCause()，必要时逐层展开原因链，而不是假定所有异常都只有一层包装。

4.4 超时保护（JDK 9+）

JDK 9 起 CompletableFuture 支持异步超时，不再需要外挂 ScheduledExecutorService：

java

// 超时作为异常进入后续处理
CompletableFuture<String> failOnTimeout = CompletableFuture
    .supplyAsync(() -> query())
    .orTimeout(3, TimeUnit.SECONDS)
    .exceptionally(ex -> "unavailable");

// 超时直接返回默认值
CompletableFuture<String> fallbackOnTimeout = CompletableFuture
    .supplyAsync(() -> query())
    .completeOnTimeout("fallback", 3, TimeUnit.SECONDS);

orTimeout 会把超时以异常方式注入链条，随后可用 exceptionally 或 handle 决定降级；completeOnTimeout 则直接给出替代值，让链条按正常结果继续。两者控制的是 CompletableFuture 的完成状态，不等同于取消底层任务。超时后，原始计算可能仍在运行；如果业务要求停止数据库查询、HTTP 请求或其他耗时操作，还需要调用底层客户端提供的取消机制，并处理取消与任务完成之间的竞态。

5. 常见反模式

线上 CompletableFuture 的 bug 高度集中在这五种模式上。

5.1 链尾忘记处理结果

java

// ❌ 没人观察完成状态、异常和取消结果
CompletableFuture.supplyAsync(() -> queryDB(id))
    .thenAccept(this::process);
// 调用方不保存返回的 future，也无法在业务边界判断是否完成

// ✅ 需要等待结果时，保留返回值并显式处理
CompletableFuture<Void> future = CompletableFuture
    .supplyAsync(() -> queryDB(id))
    .thenAccept(this::process);
try {
    future.join();
} catch (CompletionException ex) {
    // 记录 ex.getCause()，再决定重试、回滚或降级
}

不能根据调用方是否持有返回值，断言任务会因失去引用而被 GC 掉。异步链提交后通常仍由其内部依赖关系驱动。这里真正的问题是调用方没有消费结果：任务完成、失败或取消后，业务边界既无法及时获知，也无法一致地处理异常。若调用必须等待链完成，应 join()；若确实采用 fire-and-forget，应显式管理执行器、异常、超时和任务生命周期。

5.2 在回调里做阻塞 IO

java

// ❌ commonPool 里塞阻塞 IO
CompletableFuture.supplyAsync(this::queryDB)
    .thenApply(user -> httpClient.get(user.avatarUrl()))  // 阻塞！还在 commonPool 上
    .thenAccept(this::save);

thenApply 不负责切换执行线程，其回调可能在完成上游阶段的线程执行，也可能在注册回调的线程执行。如果上游阶段由公共池中的 Worker 完成，阻塞回调可能占用该 Worker；如果上游已经完成，回调也可能直接在注册回调的线程执行。含阻塞且有独立服务目标的 stage 应使用 thenApplyAsync(fn, executor) 换到专用池，避免执行线程受到上游完成时机的意外影响。

5.3 未指定 Executor

跨业务共享 commonPool 会让任务时长和故障相互影响。默认建议是为阻塞、重计算或有隔离要求的 stage 显式传 executor；短小且可接受共享调度的 stage 可以省略，但团队应统一约定哪些方法允许省略。

5.4 混用 `get()` 与 `join()`

两者行为几乎一样，但异常类型不同：

方法	检查异常	未检查异常包装
`get()`	`throws InterruptedException, ExecutionException`	业务异常包在 `ExecutionException.getCause()` 里
`join()`	不抛检查异常	业务异常包在 `CompletionException.getCause()` 里

Stream 里用 .map(CompletableFuture::join) 是标准写法（不能用 .map(CompletableFuture::get)——检查异常无法穿过 lambda）。异常处理时要记住 join() 拿到的是 CompletionException，get() 拿到的是 ExecutionException——两者都需要 .getCause()。

5.5 `allOf` 后忘记 `join` 各个 CF

java

// 先构造三个已完成的 Future
CompletableFuture<String> futureA = CompletableFuture.completedFuture("A");
CompletableFuture<String> futureB = CompletableFuture.completedFuture("B");
CompletableFuture<String> futureC = CompletableFuture.completedFuture("C");

// ❌ 只等待完成，没有读取三个 Future 的结果
CompletableFuture.allOf(futureA, futureB, futureC)
    .thenApply(v -> "missing results");

// ✅ allOf 完成后再读取各自结果
CompletableFuture.allOf(futureA, futureB, futureC)
    .thenApply(v -> String.join(",", futureA.join(), futureB.join(), futureC.join()));

allOf 的语义是“全部完成”，不是“结果汇集”。要合并结果，应在 thenApply 中读取每个已完成的 Future，或使用 thenCombine / thenCompose 表达依赖关系。此时各 Future 已完成，join() 通常立即返回；但仍应传播其失败结果，而不是只返回占位值。

6. 其他并发范式：只点名思想

CompletableFuture 解决的是"单值异步 + 编排"。生产上还有两类相邻范式，它们的主体归其他专题：

6.1 响应式编程（延伸阅读）

模型：Publisher / Subscriber / Subscription / Processor（JDK 9 的 java.util.concurrent.Flow；主流实现 Reactor、RxJava）
核心能力：数据流（而非单值），背压（onBackpressureBuffer / Drop / Latest / Error），冷流 / 热流
和 CompletableFuture 的关系：CompletableFuture = 单值 + 一次；响应式流 = 多值 + 持续
与本专题的连接：Publisher/Subscriber 协议、request(n) 拉取语义和 Scheduler 与非阻塞 I/O 密切相关，可从 Java NIO与 Netty建立上下文

一句话结论：业务里只是“异步取一次结果 + 编排”，不必上响应式；需要流处理、背压或事件驱动时，应另行评估响应式框架，而不是仅凭本页决定。

6.2 Actor 模型（延伸阅读）

核心思想：不共享状态，只传消息。每个 Actor 有独立邮箱，串行处理消息，天然无锁
能力项：监督策略（Resume / Restart / Stop / Escalate）、位置透明（本地和远程 Actor 用同一 API 调用）、事件溯源
代表：Akka（Scala 主导，Java API 完整）
主要场景：分布式系统、聊天/游戏这类"多个独立实体"的天然模型、事件驱动架构

Actor 与前述所有模型的分野在编程思维：从“共享内存 + 加锁”切换到“消息传递 + 无共享”。这个思维底座与分布式理论的一致性和 CAP 问题相关，但不属于 Java 语言或 JVM 主线。

6.3 四种模型的选型速览

模型	适合场景	本专题位置
线程 + 锁	CPU 密集、简单并发	本专题第 6、8 章
`CompletableFuture`	单值异步 + 编排、微服务扇出	本章
响应式流	高并发 I/O、流处理、背压	网络与通信专题的延伸
虚拟线程	大量阻塞 I/O、同步风格代码	本专题第 13 章
Actor	分布式系统、事件驱动、高容错	分布式专题的延伸

选型的一条起手线：能用同步风格 + 虚拟线程解决的场景，就不要引入响应式或 Actor（第 13 章给了理由）。剩下真正需要流处理和分布式容错的场景，再各自展开。

7. 本章小结

问题	根源	解决方案
`Future.get()` 阻塞	`Future` 只提供拉取通道	`CompletableFuture` 提供回调
异步任务无法组合	缺少链式 API	`thenApply` / `thenCompose` / `thenCombine`
得到 `CF<CF<R>>` 嵌套	`thenApply` 用错	改 `thenCompose`
`allOf` 后拿不到各自结果	`allOf` 返回 `Void`	手动 `join` 每个 CF
阻塞 IO 拖垮 `commonPool`	默认走 `ForkJoinPool.commonPool`	显式传 executor
链尾完成或失败状态不可见	链末端没有消费 future	按需要 `join()`，或补齐异常、监控和生命周期处理
异常常见形式为 `CompletionException`	未展开 `getCause()` 原因链	处理异常时逐层取 `ex.getCause()`
异步超时	JDK 9 之前没原生 API	`orTimeout` / `completeOnTimeout`

8. 下一步

CompletableFuture 解决组合问题，不改变底层线程模型。阻塞式异步任务可继续评估虚拟线程；需要理解事件循环和非阻塞 I/O 时进入 Java NIO，生产上的丢任务、超时和 commonPool 问题从并发性能优化开始。

异步编程：从 Future 到 CompletableFuture ​

1. Future 的三个主要局限 ​

1.1 有结果，但只能靠阻塞取 ​

1.2 没有回调通道 ​

1.3 组合的成本是嵌套 ​

2. CompletableFuture 的两组 API ​

2.1 API 全景：两类常见操作与三类独立能力 ​

2.2 转换：单输入 → 单输出 ​

2.3 合并：多个输入 → 单输出 ​

2.4 一条真实的异步流水线 ​

3. 执行线程之谜 ​

3.1 三种触发方式 ​

3.2 commonPool 的默认坑 ​

3.3 Async 变体的选型 ​

4. 异常传播的三条路径 ​

4.1 传播规则 ​

4.2 三个方法的差异 ​

4.3 异常在链条中的形状 ​

4.4 超时保护（JDK 9+） ​

5. 常见反模式 ​

5.1 链尾忘记处理结果 ​

5.2 在回调里做阻塞 IO ​

5.3 未指定 Executor ​

5.4 混用 get() 与 join() ​

5.5 allOf 后忘记 join 各个 CF ​

6. 其他并发范式：只点名思想 ​

6.1 响应式编程（延伸阅读） ​

6.2 Actor 模型（延伸阅读） ​

6.3 四种模型的选型速览 ​

7. 本章小结 ​

8. 下一步 ​

异步编程：从 `Future` 到 `CompletableFuture`

1. `Future` 的三个主要局限

1.1 有结果，但只能靠阻塞取

1.2 没有回调通道

1.3 组合的成本是嵌套

2. `CompletableFuture` 的两组 API

2.1 API 全景：两类常见操作与三类独立能力

2.2 转换：单输入 → 单输出

2.3 合并：多个输入 → 单输出

2.4 一条真实的异步流水线

3. 执行线程之谜

3.1 三种触发方式

3.2 `commonPool` 的默认坑

3.3 `Async` 变体的选型

4. 异常传播的三条路径

4.1 传播规则

4.2 三个方法的差异

4.3 异常在链条中的形状

4.4 超时保护（JDK 9+）

5. 常见反模式

5.1 链尾忘记处理结果

5.2 在回调里做阻塞 IO

5.3 未指定 Executor

5.4 混用 `get()` 与 `join()`

5.5 `allOf` 后忘记 `join` 各个 CF

6. 其他并发范式：只点名思想

6.1 响应式编程（延伸阅读）

6.2 Actor 模型（延伸阅读）

6.3 四种模型的选型速览

7. 本章小结

8. 下一步