CompletableFuture：异步编程的“智能机械臂”

机械臂这个词很亲切，这可是上家公司我们的仓储物流系统大功臣！

如果说Future是一个只会让你“傻等”的取餐牌（拿到牌 -> 去旁边站着 -> 叫号了去取），那么CompletableFuture就是工厂里的智能流水线机械臂

• 传统 Future 痛点：
  • 阻塞：调用future.get()会卡住当前线程，直到任务完成。
  • 无法编排：很难实现“任务 A 做完自动做 B，B 做完并行做 C 和 D，最后汇总”这种复杂逻辑。代码会变成地狱般的嵌套回调（Callback Hell）或者一堆杂乱的线程管理。
• CompletableFuture 核心能力：
  • 非阻塞：任务提交后立即返回，不占用主线程。
  • 函数式编排：像搭积木一样链式调用 (thenApply,thenCompose)。
  • 异常处理：统一的exceptionally或handle机制。
  • 多任务聚合：轻松实现allOf(全做完) 或anyOf(谁快用谁)。

第一部分：核心概念——从“取餐牌”到“流水线”

1. 为什么需要它？（场景对比）

比如练手做小型电商项目，构建一个电商详情页，需要获取：

1. 用户信息(User Service) - 耗时 50ms
2. 商品信息(Product Service) - 耗时 80ms
3. 推荐列表(Recommend Service) - 依赖用户信息，耗时 60ms
4. 优惠券(Coupon Service) - 依赖用户 + 商品，耗时 40ms
5. 当然真实的业务场景不是这么用的，只是用于概念的讲解！！！

// 代码混乱，难以维护，容易忘记关闭线程池，异常处理麻烦 Future<User> f1 = pool.submit(() -> userService.getUser()); Future<Product> f2 = pool.submit(() -> productService.getProduct()); User user = f1.get(); // 阻塞！ Product product = f2.get(); // 阻塞！ ——————————————————————————————————————————————————————————————————————————————————— CompletableFuture (虽是异步编排、环环相扣、逻辑相连 // 总耗时 ≈ max(50, 80) + max(60, 40) ≈ 80 + 60 = 140ms (性能提升 40%+) // 且代码像流水账一样清晰 CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.getUser()); CompletableFuture<Product> productFuture = CompletableFuture.supplyAsync(() -> productService.getProduct()); // 依赖 user 的任务自动触发 CompletableFuture<List<Rec>> recFuture = userFuture.thenApply(user -> recService.getRecs(user)); // 依赖 user 和 product 的任务，等两者都完成后自动触发 CompletableFuture<List<Coupon>> couponFuture = userFuture.thenCombine(productFuture, (u, p) -> couponService.getCoupons(u, p)); // 最后汇总所有结果 CompletableFuture<PageData> allDone = CompletableFuture.allOf(recFuture, couponFuture) .thenApply(v -> assemblePage(recFuture.join(), couponFuture.join()));

这就像饭店上菜一样，作为服务员你不可能一直站在厨房门口等菜。你下单后（submit），厨师（线程池）开始做菜。菜做好了（complete），自动传送到下一个工位（thenApply），最后打包好直接端到你面前（join/get，或者回调通知）

第二部分：核心 API 详解与实战

重要警告：生产环境永远不要使用默认的commonPool()！

• 原因：它是全局共享的。如果你的任务里有 IO 阻塞（查库、调接口），会把公共池的线程占满，导致整个 JVM 其他使用并行流或 CF 的地方全部卡死。
• 最佳实践： always provide a customExecutor(e.g.,ThreadPoolExecutor).自定义

2. 转换结果：thenApply(流水线加工)

场景：上一步的结果，经过计算，变成下一步的输入。

• 签名：thenApply(Function<T, U>)
• 行为：当前任务完成后，在当前线程（或指定线程）执行转换函数，返回新结果。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { System.out.println("Step 1: 获取原始数据 (Thread: " + Thread.currentThread().getName() + ")"); return 10; }, customExecutor); // 链式调用：10 -> 20 CompletableFuture<Integer> result = future.thenApply(data -> { System.out.println("Step 2: 数据翻倍 (Thread: " + Thread.currentThread().getName() + ")"); return data * 2; }); // 注意：thenApply 默认复用上一个任务的线程（如果上一个刚结束），除非指定 executor

3. 依赖另一个异步任务：thenCompose(扁平化/串联)

场景：第二步也是一个异步任务（返回CompletableFuture），你需要把两个未来“拍平”成一个

• 签名：thenCompose(Function<T, CompletableFuture<U>>)
• 比喻：第一步拿到了“订单 ID”，第二步要用这个 ID 去异步查询“订单详情”。
• 区别：
  • thenApply:T->U(同步转换)
  • thenCompose:T->CompletableFuture<U>(异步依赖，避免嵌套CompletableFuture<CompletableFuture<U>>)

// 模拟异步获取用户 ID CompletableFuture<String> userIdFuture = CompletableFuture.supplyAsync(() -> "U1001"); // 错误写法 (会得到 CompletableFuture<CompletableFuture<User>>) // userIdFuture.thenApply(id -> getUserAsync(id)); // 正确写法 (thenCompose 拍平) CompletableFuture<User> userFuture = userIdFuture.thenCompose(id -> { System.out.println("拿到 ID: " + id + ", 开始异步查询用户详情..."); return getUserAsync(id); // 返回一个新的 CompletableFuture });

4. 合并两个任务：thenCombine(并联汇聚)

任务 C 依赖 任务 A和任务 B 的结果。A 和 B 并行执行，都完成后触发 C

• 签名：thenCombine(OtherFuture, BiFunction<T, U, V>)

CompletableFuture<Integer> taskA = CompletableFuture.supplyAsync(() -> 10); CompletableFuture<Integer> taskB = CompletableFuture.supplyAsync(() -> 20); // 等 A 和 B 都做完，执行相加 CompletableFuture<Integer> sumFuture = taskA.thenCombine(taskB, (a, b) -> { System.out.println("A=" + a + ", B=" + b + ", 计算总和"); return a + b; });

5. 等待所有/任意任务：allOf/anyOf

• allOf(f1, f2, ...): 等待所有任务完成。返回CompletableFuture<Void>。
  • 用法：通常配合join()提取各个任务的结果。
• anyOf(f1, f2, ...):任意一个完成任务即返回（常用于多源兜底，谁快用谁）

6. 异常处理：不让流水线崩塌

传统的try-catch在异步链式中很难写。CF 提供了专门的钩子。

• exceptionally(Function<Throwable, T>): 类似 catch，返回一个默认值。
• handle(BiFunction<T, Throwable, U>): 无论成功还是异常都会执行（类似 finally + 判断）。最推荐

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { if (true) throw new RuntimeException("服务挂了!"); return "OK"; }).handle((result, ex) -> { if (ex != null) { System.err.println("出错了: " + ex.getMessage()); return "默认降级数据"; // 返回兜底值 } return result; // 正常返回 });

第三部分：微服务实战——编排复杂调用链

场景：构建一个聚合接口，需要并行调用三个下游服务，其中两个有依赖关系，最后汇总。

• Task A: 获取基础配置 (独立)
• Task B: 获取用户信息 (独立)
• Task C: 获取用户订单 (依赖 B)
• Task D: 组装最终结果 (依赖 A, C)

import java.util.concurrent.*; import java.util.stream.Collectors; public class MicroserviceOrchestration { // 自定义线程池 (关键！隔离业务，控制资源) private static final ExecutorService executor = new ThreadPoolExecutor( 10, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadFactoryBuilder().setNameFormat("async-pool-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy() ); public CompletableFuture<ResponseDTO> buildPage(String userId) { // 1. 并行启动独立任务 A 和 B CompletableFuture<Config> configFuture = CompletableFuture .supplyAsync(() -> mockConfigService(), executor) .exceptionally(ex -> Config.defaultConfig()); // 降级 CompletableFuture<User> userFuture = CompletableFuture .supplyAsync(() -> mockUserService(userId), executor) .exceptionally(ex -> User.guestUser()); // 降级 // 2. 任务 C 依赖 B (thenCompose) CompletableFuture<Order> orderFuture = userFuture.thenCompose(user -> { if ("guest".equals(user.getType())) { return CompletableFuture.completedFuture(Order.empty()); // 快速返回，不再调下游 } return CompletableFuture.supplyAsync(() -> mockOrderService(user.getId()), executor); }); // 3. 任务 D 依赖 A 和 C (thenCombine) CompletableFuture<ResponseDTO> finalResult = configFuture.thenCombine(orderFuture, (config, order) -> { // 这里执行最后的组装逻辑 return new ResponseDTO(config, order); }); return finalResult; } // 模拟调用 private Config mockConfigService() { /* sleep 50ms */ return new Config(); } private User mockUserService(String id) { /* sleep 80ms */ return new User(id); } private Order mockOrderService(String uid) { /* sleep 60ms */ return new Order(uid); } public static void main(String[] args) throws Exception { MicroserviceOrchestration service = new MicroserviceOrchestration(); long start = System.currentTimeMillis(); // 发起请求 (非阻塞) CompletableFuture<ResponseDTO> future = service.buildPage("U123"); // 在主线程等待结果 (实际 Web 容器中，框架会帮你处理这个等待，直接返回 DeferredResult) ResponseDTO response = future.join(); System.out.println("总耗时: " + (System.currentTimeMillis() - start) + "ms"); System.out.println("结果: " + response); executor.shutdown(); } }

1. 自定义线程池：避免了污染commonPool，且可以针对该业务调整队列大小和拒绝策略。
2. 异常降级：每个远程调用都加了exceptionally，保证单个服务挂掉不会导致整个页面白屏，而是显示默认值。
3. 短路优化：在thenCompose中判断如果是 Guest 用户，直接返回空订单，不再发起多余的 RPC 调用。
4. 自动并行：configFuture和userFuture同时启动，互不阻塞。

第四部分：避坑指南与实践

慎用get()和join()

• 原则：尽量将逻辑写在链式调用 (thenApply,thenAccept) 中，让框架自动回调。
• 例外：只有在最外层（如 Controller 出口，或单元测试）才调用join()等待最终结果。在链条中间调用join()会阻塞当前线程，破坏异步优势。

线程池隔离

• IO 密集型（调 RPC、查 DB）：线程数可以设大一点（如 CPU 核数 * 2 或更多），因为线程大部分时间在 wait。
• CPU 密集型（计算）：线程数 = CPU 核数 + 1。
• 不同业务隔离：核心业务（如下单）和非核心业务（如推荐）使用不同的线程池，防止非核心业务把线程池占满，拖垮核心业务。

上下文传递 (ThreadLocal)

• 问题：supplyAsync会切换线程，导致ThreadLocal(如 TraceID, UserContext)丢失。
• 解决：
  • 方案 A：在supplyAsync之前手动把变量取出来，作为参数传进去（推荐，最简单）。
  • 方案 B：使用InheritableThreadLocal(仅限线程池复用线程时有效，且有风险)。
  • 方案 C：使用阿里TransmittableThreadLocal(TTl) 等专门库，在包装 Runnable/Supplier 时传递上下文。

避免回调地狱 (Callback Hell)

虽然 CF 比原生 Callback 好，但如果链式调用超过 5-6 层，代码也会难读。

• 建议：将长链条拆分成多个方法，每个方法返回一个CompletableFuture

// 坏：超长链 f1.thenApply(...).thenCompose(...).thenApply(...).thenCombine(...)... // 好：拆分 CompletableFuture<A> step1() { ... } CompletableFuture<B> step2(A a) { ... } public CompletableFuture<Result> orchestrate() { return step1().thenCompose(this::step2)... }