completableFuture 异步编排实战 —— 多任务并行、结果汇总、超时控制与线程池协作
如果说前 1–9 篇解决的是“线程池如何安全、稳定地跑”，
那么这一篇解决的是：
如何把多个异步任务“编排”成一个可读、可控、可维护的并发流程。
这正是现代 Java 并发从ThreadPoolExecutor → CompletableFuture的进化方向。
一、为什么需要 CompletableFuture？
先看一个你一定写过的代码：
Future<User> f1 = pool.submit(() -> this.loadUser()); Future<User> f2 = pool.submit(() -> this.loadUser()); User user = f1.get(); Order order = f2.get();问题很明显：
- get() 阻塞
- 顺序代码读起来像同步
- 异常处理零散
- 任务依赖一多，代码迅速失控
CompletableFuture 的核心价值只有一句话：
用“声明式”的方式，描述异步任务之间的关系，而不是用 get() 等结果。
二、CompletableFuture 和线程池的关系（先搞清楚）
1️⃣ CompletableFuture ≠ 线程池
CompletableFuture 是异步任务编排工具
线程池是执行引擎
2️⃣ 默认线程池是 ForkJoinPool（不推荐直接用）
CompletableFuture.supplyAsync(() -> work());默认使用：
ForkJoinPool.commonPool()👉生产环境强烈建议：显式传入你自己的线程池（第五篇）
CompletableFuture.supplyAsync(() -> work(), ioPool);三、最核心的三种编排模式（80% 场景）
1️⃣ thenApply —— 单任务链式变换
CompletableFuture .supplyAsync(() -> 1, pool) .thenApply(x -> x + 1) .thenApply(x -> x * 2) .thenAccept(System.out::println);- 同一条任务链
- 上一步完成 → 执行下一步
- 适合数据转换
2️⃣ thenCompose —— 依赖型异步（避免嵌套）
CompletableFuture<User> f = loadUserAsync(id) .thenCompose(user -> loadProfileAsync(user));等价于（但比它优雅得多）：
CompletableFuture<User> f = loadUserAsync(id) .thenApply(user -> loadProfileAsync(user)) .get(); // ❌一句话：
thenCompose = “异步版的 flatMap”
3️⃣ thenCombine —— 并行任务结果合并（非常常用）
CompletableFuture<User> userFuture = loadUserAsync(id); CompletableFuture<Order> orderFuture = loadOrderAsync(id); CompletableFuture<UserInfo> result = userFuture.thenCombine(orderFuture, (user, order) -> new UserInfo(user, order));✔ 并行执行
✔ 都完成后才合并
✔ 没有 get()
四、allOf / anyOf：真正的“并行编排”
allOf：全部完成
CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3); all.thenRun(() -> System.out.println("all done"));⚠ 注意：allOf不帮你收集结果，你需要自己 get（或 join）
List<Result> results = List.of(f1, f2, f3) .stream() .map(CompletableFuture::join) .toList();anyOf：任意一个完成
CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3); all.thenRun(() -> System.out.println("all done"));⚠ 注意：allOf不帮你收集结果，你需要自己 get（或 join）
List<Result> results = List.of(f1, f2, f3) .stream() .map(CompletableFuture::join) .toList();anyOf：任意一个完成
CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2); any.thenAccept(r -> System.out.println("first = " + r));常用于：
- 多数据源兜底
- 多节点竞速
五、异常处理：这是 CompletableFuture 的强项
1️⃣ exceptionally —— 兜底恢复
CompletableFuture .supplyAsync(() -> risky(), pool) .exceptionally(e -> { log.error("error", e); return defaultValue; });2️⃣ handle —— 成功 / 失败都处理
future.handle((r, e) -> { if (e != null) { return fallback; } return r; });一句工程经验：
CompletableFuture 的异常是“流的一部分”，不是打断流程。
六、超时控制（非常关键）
Java 9+ 推荐方式
future .orTimeout(2, TimeUnit.SECONDS) .exceptionally(e -> fallback);或者：
future .completeOnTimeout(fallback, 2, TimeUnit.SECONDS);比Future.get(timeout)的优势：
- 不阻塞线程
- 超时是异步语义的一部分
七、CompletableFuture + 线程池的最佳实践
✔ 1）明确线程池职责
IO 任务 → ioPool
CPU 任务 → cpuPool
定时 → scheduledPool
CompletableFuture.supplyAsync(this::loadData, ioPool)
✔ 2）不要在 CompletableFuture 里 get()
// ❌ 反模式 future.thenApply(r -> anotherFuture.get());✔ 3）异常必须收敛在链路末端
future .thenApply(...) .thenApply(...) .exceptionally(this::fallback);✔ 4）避免在 commonPool 跑重任务
ForkJoinPool 是共享资源，容易拖垮 JVM。
八、一个完整实战 Demo
补充知识点：
Java 的“高阶函数”到底是什么：Runnable / Callable 就是函数参数
CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(this::loadUser, ioPool); CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(this::loadOrder, ioPool); CompletableFuture<UserInfo> result = userFuture .thenCombine(orderFuture, UserInfo::new) .orTimeout(2, TimeUnit.SECONDS) .exceptionally(e -> { log.error("timeout or error", e); return UserInfo.empty(); }); result.thenAccept(info -> render(info));这个 Demo 覆盖了：
- 并行
- 合并
- 超时
- 异常
- 自定义线程池
九、和前 9 篇的“闭环关系”
你现在拥有的是一套完整体系：
- 线程池（1–5）
- 任务提交与异常（6–7）
- 可观测性（8）
- 背压（9）
- 异步编排（10） ← 收官
一句总结：
ThreadPoolExecutor 决定“系统能不能跑”，
CompletableFuture 决定“并发代码能不能写得优雅、可维护”。
十、全专栏终极总结
线程池是并发执行的基础设施
背压决定系统是否稳定
监控决定问题是否可见
CompletableFuture 决定异步逻辑是否可维护
并发不是“多开线程”，而是“正确组织任务关系”
到这里，已经是一个完整、工程级、的体系。
