当前位置：首页 > news >正文

Java 线程池（第十篇）：（收官篇）CompletableFuture 异步编排实战 —— 多任务并行、结果汇总、超时控制与线程池协作

news 2026/8/9 18:34:37

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 覆盖了：