跳至主要内容

CompletableFuture：组合、汇聚与超时的真实边界

IT枫斗者大约 6 分钟技术进阶JavaCompletableFuture异步编程并发

一次请求要先查用户，再按用户编号查权益，还要并行读取两份摘要。CompletableFuture 能表达这些依赖，但它代表的是结果的完成状态，不能把“结果不再等待”理解成“工作已经停止”。本文按 Java 21 说明，示例使用本地模拟任务，不依赖数据库或 Redis。

thenApply 转换值，thenCompose 连接异步阶段

假设上游是 CompletableFuture<Integer>，得到用户编号后，拼接显示名称只需 thenApply；如果下一步本身返回 Future，则用 thenCompose 接上它。

var source = CompletableFuture.supplyAsync(() -> 7, executor);
var name = source.thenApply(id -> "user-" + id);
var nested = source.thenApply(id -> CompletableFuture.supplyAsync(
        () -> "user-" + id, executor));
var flat = source.thenCompose(id -> CompletableFuture.supplyAsync(
        () -> "user-" + id, executor));

name 的结果是字符串；nested 的结果仍是一个 Future，外层完成只说明内层对象已产生，读取字符串还要再等一次；flat 要等内层完成才完成。Compose 消除的是一层嵌套，不会凭空增加并发，也不会改变“先有编号，后查名称”的依赖。官方 thenCompose 契约要求函数返回的阶段最终能够完成。

不要在回调里用 join() 假装实现扁平化：它会占住执行回调的线程，小线程池中若等待的任务也排在同一个池里，可能互相等待。用依赖关系表达等待，最后只在明确的同步边界读取结果。

executor 决定在哪里执行

不带 Async 的回调可能由完成上游的线程执行；上游已经完成时，也可能直接在注册回调的线程执行。thenApply 并不保证切换线程。普通 CompletableFuture 的无显式 executor 的异步方法通常使用公共 ForkJoinPool；公共池并行度不足二时有另建线程的规则，子类还可改写默认执行器。具体规则见官方执行策略。

完整源码用以下线程池包住整个示例，并在退出时关闭：

try (ExecutorService executor = new ThreadPoolExecutor(
        2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(16),
        Thread.ofPlatform().name("cf-demo-", 0).factory(),
        new ThreadPoolExecutor.AbortPolicy())) {
    var result = CompletableFuture.supplyAsync(() -> 7, executor)
            .thenApplyAsync(id -> "user-" + id, executor);
    System.out.println(result.join());
}

上游指定 executor，不代表后面的所有 Async 都自动继承它；每个需要指定位置的异步阶段都应显式传入。这里的容量只为演示：实际还要考虑拒绝、排队与下游连接数。提交被拒绝也需要处理，不能认为所有失败都会自动装入 Future。

allOf 是完成屏障，不是结果列表

两个独立任务应先提交，再汇聚，避免启动一个就立即 join()，把并行流程写成串行：

var a = CompletableFuture.supplyAsync(() -> "A", executor);
var b = CompletableFuture.supplyAsync(() -> "B", executor);
var combined = CompletableFuture.allOf(a, b)
        .thenApply(ignored -> a.join() + b.join());

allOf 返回 CompletableFuture<Void>；各项正常完成后，才能在成功回调中逐个取值。只要有一项异常，屏障最终也异常，但仍等待其余输入完成，不承诺首个错误到来就返回，更不会取消其他任务。多个输入同时失败时，不应依赖它选择某一个固定异常；要保留全部错误，应先把每项通过 handle 转成带成功或失败信息的业务结果，再汇聚。空输入的屏障立即正常完成，结果为 null。参见官方 allOf 契约。

异常处理的位置改变后续语义

方法触发条件对后续结果的影响
exceptionally上游失败返回替代值，恢复为正常结果
handle上游成功或失败把两种状态映射成新结果
whenComplete上游成功或失败观察结果，通常继续传播原状态

异常会沿依赖传播，成功型回调通常跳过。恢复放在 allOf 之前，是允许单项降级；放在汇聚之后，是对整体失败兜底，二者业务含义不同。别把任意错误都变成空列表，否则权限错误、程序缺陷也会被隐藏。

whenComplete 适合记录日志，但记录动作自己抛异常也可能导致其返回阶段失败；若上游也失败，优先传播上游异常。判断失败应检查异常参数，因为正常结果也允许为 null。join() 常把失败包装为 CompletionException，get() 则通过 ExecutionException 暴露原因并允许中断等待；取消还有 CancellationException 的特殊情况。处理规则见CompletionStage 官方说明。

timeout 不等于停止底层任务

操作改变什么是否停止正在执行的任务
get(timeout, unit)当前调用者的等待期限否，Future 可继续完成
orTimeout原 Future 超时后异常完成否
completeOnTimeout原 Future 超时后以默认值正常完成否
cancel(true)CompletableFuture 的取消状态不保证；参数在此类中不负责中断计算

orTimeout 和 completeOnTimeout 返回的就是原对象，会影响共享该对象的其他消费者。想给某个消费者单独设期限，可对 copy() 的返回值设超时，但底层工作仍需要独立治理。

示例先等待任务报告“已启动”，再设 50 毫秒超时。工作线程被门闩挡住，主线程观察到 TimeoutException 时，任务尚未结束；主线程释放门闩后，任务才结束，但迟到的返回值无法覆盖既有超时状态。这证明的是状态与执行生命周期不同，不是精确的实时调度保证。

真正回收资源，要让底层协议支持取消或截止时间，例如 HTTP 请求超时、数据库查询超时，或任务主动检查中断和取消标记。即便另持 Future 并请求中断，也需要任务配合；已经发生的写入不会因此撤销。参见官方 orTimeout与cancel。

可运行示例与延伸阅读

完整源码位于 examples/java-advanced/completablefuture/CompletableFutureDemo.java，命令及实际输出位于 docs/upgrade-100/checks/completablefuture/。JDK 21 编译运行已执行；EXPECTED_FAILURE 表示主动触发并捕获的预期异常，程序最终应输出 PASS。示例验证组合、汇聚、异常恢复与超时后继续执行，不涉及真实数据库、Redis 或 HTTP 调用，这些集成未执行。

线程池容量与排队问题可继续阅读线程池过载边界；背景材料保留在Java 线程池最佳实践。本文 API 语义引用均来自 Oracle Java SE 21 官方文档。