# CompletableFuture 实战:招式和坑,一张表讲完 > 面向读者:对 Java 并发有一定经验、想快速把 `CompletableFuture` 用对的工程师。 > 完整叙述版见 `completablefuture-tutorial.md`,可运行示例见 `src/main/java/space/anyi/Cf*.java`。 先给个定位:`CompletableFuture` 是 JDK 8 加进来的,把异步、回调、组合揉在一起。说白了就是把 JS Promise 和 Stream 的 map/flatMap 搬到了 Java。 为什么需要它?因为老的 `Future` 只能阻塞 get、只能跑单任务。串行拼接、多任务聚合、异常传递,全得自己手写。`CompletableFuture` 就是为补这些短板来的。 ## 速查表 | 需求 | 方法 | 回调 | 返回 | | --- | --- | --- | --- | | 无返回异步 | `runAsync(Runnable)` | `Runnable` | `CF` | | 有返回异步 | `supplyAsync(Supplier)` | `Supplier` | `CF` | | 无需结果直接过 | `completedFuture(v)` | — | `CF` | | 拿上个结果继续加工 | `thenApply` | `Function` | `CF` | | 消费上个结果 | `thenAccept` | `Consumer` | `CF` | | 只等上一环跑完 | `thenRun` | `Runnable` | `CF` | | 两任务都完成再合并 | `thenCombine` | `BiFunction` | `CF` | | 两任务都完成再消费 | `thenAcceptBoth` | `BiConsumer` | `CF` | | 两任务都完成等时间点 | `runAfterBoth` | `Runnable` | `CF` | | 谁先完成用谁再加工 | `applyToEither` | `Function` | `CF` | | 谁先完成消费谁 | `acceptEither` | `Consumer` | `CF` | | 谁先完成等时间点 | `runAfterEither` | `Runnable` | `CF` | | 展平嵌套异步 | `thenCompose` | `Function→CF` | `CF` | | 全部完成 | `allOf(...)` | — | `CF`(各自再 join 汇总) | | 任一完成 | `anyOf(...)` | — | `CF`(需转类型) | | 失败降级 | `exceptionally` | `Function` | `CF` | | 异常后异步续接(展平) | `exceptionallyCompose` | `Function→CF` | `CF` | | 成败都观察 | `whenComplete` | `BiConsumer` | 不改结果 | | 成败都能处理 | `handle` | `BiFunction` | `CF` | | 限时等待,超时快速失败 | `orTimeout(超时, 单位)` | — | `CF` | | 超时补默认值 | `completeOnTimeout(v, 超时, 单位)` | — | `CF` | | 带超时取结果 | `get(超时, 单位)` | — | `T` | | 是否已结束 | `isDone` | — | `boolean` | | 手动定结果 | `complete` / `completeExceptionally` | — | `boolean` | | 强制覆盖/强制异常 | `obtrudeValue` / `obtrudeException` | — | `void`(慎用) | | 取消 | `cancel(true)` | — | `boolean` | 记忆口诀:Apply=加工出结果,Accept=消费不出结果,Run=不管结果只管流程;后缀 Both=Either 只差"都要"还是"谁先"。 ## 六条最佳实践 1. I/O 一定配独立线程池。默认 `ForkJoinPool.commonPool()` 线程数是 `CPU核数-1`,回调里再嵌 `supplyAsync` 就会互相等(Thread Starvation),直接把池子卡死。给耗时调用 `new ThreadPoolExecutor(...)` 并传进 Async 方法。 2. 组装 BFF / RPC 并发调用用 `allOf`。一次性把下游都发出去真并行,别用 `for + get()` 退化成串行。注意 `allOf` 返回 `Void`,汇总前再逐个 `join()`。 3. 回调里禁止 `join()/get()`。它阻塞的是当前执行线程;在一个 worker 上阻塞等另一个 worker,就是 2 号事故的引信。 4. 统一异常出口。想兜底恢复用 `exceptionally`/`handle`,只想记录日志用 `whenComplete`(它不会吞异常,异常仍会沿链传播,最终 `join()` 抛 `CompletionException`,要用 `getCause()` 取原始异常)。 5. 链式只加一次终端操作。`future.thenApply(...).thenApply(...)` 是合法的,"上了链就别回头",避免把同一个 Future 反复 `join()` 造成多次阻塞。 6. 对外异步调用一律配超时。`orTimeout(2, SECONDS)` 让调用方快速失败(注意它不会取消底层任务,只是放弃等待);"过期给默认值也能接受"的场景用 `completeOnTimeout(v, 2, SECONDS)` 更省心。没有超时的异步外呼,等于给自己埋一颗"线程挂死"的雷。 ## 三个高频坑 `whenComplete` 只是旁观。很多同学以为加个 `whenComplete` 就算异常处理完了,实测它观察完异常后,链下游/`join()` 依旧抛异常。要"处理掉"请用 `handle` 或 `exceptionally`。 - `Async` 不等于更快。无 Async 的方法在"触发它的线程"上执行(可能就是主线程)。它提供的不是算力,而是"指定线程池"的能力。 - 取结果首选 `join()`。`get()` 抛受检异常,把链式代码弄得很啰嗦;`join()` 统一包成 `CompletionException`,链路里不打断。 ## 与隔壁方案的边界 | 方案 | 优点 | 局限 | | --- | --- | --- | | `Future` | 简单、JDK5 就有 | 只能阻塞取结果,无法链式/聚合 | | `CompletableFuture` | JDK 内置,聚合强,学习成本低 | 回调多、线程池管理靠自觉、无背压 | | 响应式流(Reactive/Reactor) | 背压、流控、切线程优雅,操作符强大 | 上手曲线陡,全链路改造成本高 | 服务内部异步编排,`CompletableFuture` 是最划算的选择。一旦涉及边缘/流式/背压,再上响应式。别为了"帅"把一个简单聚合演进成全套响应式。 ## 实战案例:把 API 用到真实场景 下面每个案例都可以直接复制运行,对应 `src/main/java/space/anyi/Cf*.java`。 ### 案例一:订单处理流水线(thenApply / thenAccept / thenRun) 典型电商下单流程:创建订单 → 扣库存 → 记日志 → 发通知,每一步都依赖上一步的结果。用 `thenXxx` 串成一条链,代码就像在写流程图: ```java CompletableFuture future = CompletableFuture.supplyAsync(() -> { System.out.println("1. 创建订单"); return "订单号 2024001"; }).thenApply(orderId -> { System.out.println("2. 扣减库存,关联订单: " + orderId); return orderId + " 已扣库存"; }).thenApply(info -> { System.out.println("3. 记录日志: " + info); return info + " 日志已落库"; }).thenAccept(result -> { System.out.println("4. 打印流程结果: " + result); // 只消费结果,无返回值 }).thenRun(() -> { System.out.println("5. 发送短信通知(不关心数据)"); // 连结果都不关心 }); future.join(); ``` 这里其实藏着选型逻辑:只要还想继续加工,就用 `thenApply`;结果到此为止,用 `thenAccept` 消费或者 `thenRun` 等个时间点。哪个用得多,取决于你这链条到底有多长。 ### 案例二:并发聚合两个独立接口(thenCombine) 页面要同时展示"用户信息"和"订单信息",两个请求互不依赖,谁先回来无所谓,关键是两个都回来后拼在一起: ```java CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> { sleep(100); return "用户:李雷"; }); CompletableFuture orderFuture = CompletableFuture.supplyAsync(() -> { sleep(200); return "订单:O-10008"; }); // thenCombine: 等两个都完成,把两个结果合并,返回一个新结果 CompletableFuture combined = userFuture.thenCombine(orderFuture, (user, order) -> user + " | " + order); System.out.println("合并结果: " + combined.join()); ``` 这个 API 的网络耗时就是在这里体现的:两个 sleep 是同时进行的,总共约 200ms 而不是 300ms。回调参数是 `(user, order)` 两个结果,这也是 `thenCombine` 和 `thenApply` 的本质区别:前者拿两个,后者拿一个。 ### 案例三:缓存优先读取(applyToEither) 数据优先读本地缓存(快),缓存没有就等远程(慢),谁先返回就用谁,避免干等最慢的那一路: ```java CompletableFuture cache = CompletableFuture.supplyAsync(() -> { sleep(100); return "缓存数据"; }); CompletableFuture remote = CompletableFuture.supplyAsync(() -> { sleep(400); return "远程数据"; }); // applyToEither: 谁先完成用谁的结果,并继续加工 CompletableFuture result = cache.applyToEither(remote, data -> "优先使用 -> " + data); System.out.println("结果: " + result.join()); // 输出: 优先使用 -> 缓存数据 ``` `applyToEither` 只关心"谁先来",两个 Future 的角色是对等的,谁先完成就用谁的结果。适合缓存+远程、主备容灾、多数据源"先到先得"。 ### 案例四:展平嵌套异步调用(thenCompose) 当 `thenApply` 的回调里返回的又是 `CompletableFuture` 时,会得到嵌套的 `CF>`,取结果得 `join().join()`。`thenCompose` 帮你展平这层嵌套,和 `Stream` 里的 `flatMap` 一模一样: ```java CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> { System.out.println("1. 查询用户"); return "用户:李雷"; }); // thenApply 会产生嵌套:CompletableFuture> CompletableFuture> nested = userFuture.thenApply(Cf05_ThenCompose::getOrder); System.out.println("两层 join 取结果: " + nested.join().join()); // thenCompose 自动展平,保持链式扁平 CompletableFuture flat = userFuture.thenCompose(Cf05_ThenCompose::getOrder); System.out.println("展平结果: " + flat.join()); // 辅助方法:拿到用户后再异步查他的订单 private static CompletableFuture getOrder(String user) { return CompletableFuture.supplyAsync(() -> { System.out.println("2. 异步查询 \"" + user + "\" 的订单"); return "订单 O-10008"; }); } ``` `thenApply` 原样透传回调返回值,`thenCompose` 要求回调返回 `CompletionStage` 并帮你展平。前后依赖且每一环都是异步的调用链,用 `thenCompose` 保持扁平。 ### 案例五:批量并发聚合(allOf / anyOf) 一个页面需要同时调"基础信息、库存、优惠"三个接口,全部返回后才渲染。别傻傻写三次 `get()`,那会退化成串行: ```java CompletableFuture baseInfo = CompletableFuture.supplyAsync(() -> { sleep(100); return "基础信息"; }); CompletableFuture stock = CompletableFuture.supplyAsync(() -> { sleep(200); return "库存信息"; }); CompletableFuture coupon = CompletableFuture.supplyAsync(() -> { sleep(300); return "优惠信息"; }); // allOf: 等全部完成(返回 Void),再各自 join 汇总,让三路请求真正并行 CompletableFuture all = CompletableFuture.allOf(baseInfo, stock, coupon); all.join(); List data = Arrays.stream(new CompletableFuture[]{baseInfo, stock, coupon}) .map(CompletableFuture::join) .map(Object::toString) .collect(Collectors.toList()); System.out.println("全部就绪,汇总: " + String.join(" | ", data)); // anyOf: 任一路先完成即返回(结果是 Object),用于"谁先回来先给谁出结果" CompletableFuture any = CompletableFuture.anyOf(baseInfo, stock, coupon); System.out.println("最先返回: " + any.join()); ``` `allOf` 返回 `CompletableFuture`,本身不带汇总结果,得在 `join()` 后逐个取。`anyOf` 返回 `CompletableFuture`,需要做类型转换。这两货名字看着像兄弟,用起来完全是两码事。 ### 案例六:异常降级与恢复(exceptionally / whenComplete / handle) 异步任务挂了怎么办?`CompletableFuture` 把异常也当作"一种结果"来处理,三种方式各有所长: ```java // exceptionally: 任务失败时给出降级结果(异常被"恢复") CompletableFuture fallback = failTask().exceptionally(e -> { System.out.println("exceptionally 捕获: " + e.getClass().getSimpleName()); return -1; // 降级返回 }); System.out.println("降级结果: " + fallback.join()); // -1 // whenComplete: 只"观察"结果或异常,不消化异常 CompletableFuture observed = failTask(); try { observed.whenComplete((res, ex) -> { if (ex != null) { System.out.println("whenComplete 观察到异常: " + ex.getClass().getSimpleName()); } else { System.out.println("whenComplete 观察到结果: " + res); } }).join(); } catch (Exception e) { System.out.println("whenComplete 不吞异常, join() 抛: " + e.getClass().getSimpleName()); } // handle: 无论成功失败都给一个函数处理,并可返回新结果(可恢复) CompletableFuture handled = failTask().handle((res, ex) -> ex == null ? "正常结果: " + res : "出错被恢复: " + ex.getMessage()); System.out.println("handle 结果: " + handled.join()); private static CompletableFuture failTask() { return CompletableFuture.supplyAsync(() -> { throw new RuntimeException("模拟异常"); }); } ``` 三种方式对比: | 方法 | 成功时触发 | 失败时触发 | 能改变结果 | 会不会吞异常 | | --- | --- | --- | --- | --- | | `exceptionally` | ✗ | ✓ | ✓ 返回降级值 | 会 | | `whenComplete` | ✓ | ✓ | ✗ 只观察 | 不会 | | `handle` | ✓ | ✓ | ✓ 可恢复 | 会 | `whenComplete` 管看不管埋,它只做旁观者。要真正处理异常,请用 `handle` 或 `exceptionally`。还有一点坑:所有异常入口抛的都是 `CompletionException`,`.getCause()` 才是原始异常,别对着 `CompletionException` 猜原因。 ### 案例七:手动完成与状态查询(complete / isDone / cancel) 异步世界里有一种情况:结果的给出不由任务决定,而由外部决定。等用户点击、等第三方回调、等消息推送,都可以用空的 `CompletableFuture` 当闸门: ```java // 用空 CF 当"闸门",等外部事件触发后手动 complete CompletableFuture gate = new CompletableFuture<>(); new Thread(() -> { sleep(500); gate.complete("外部线程触发了完成"); }).start(); System.out.println("阻塞等待手动完成: " + gate.join()); // completeExceptionally: 手动让任务以异常结束,常用于"超时兜底" CompletableFuture timeout = new CompletableFuture<>(); timeout.completeExceptionally(new RuntimeException("调用超时,兜底结束")); try { timeout.join(); } catch (Exception e) { System.out.println("completeExceptionally 抛给调用方: " + e.getCause()); } // isDone: 查询任务是否已结束(成功、失败、取消都算 done) CompletableFuture pending = new CompletableFuture<>(); System.out.println("刚创建未完成 isDone: " + pending.isDone()); // false pending.complete("已赋值"); System.out.println("完成后 isDone: " + pending.isDone()); // true // cancel: 取消任务 CompletableFuture cancelling = new CompletableFuture<>(); System.out.println("cancel() 返回值: " + cancelling.cancel(true)); System.out.println("取消后 isCancelled: " + cancelling.isCancelled() + ", isCompletedExceptionally: " + cancelling.isCompletedExceptionally()); ``` `new CompletableFuture<>()` 当栅栏/闸门,等外部事件到来后再 `complete`,让等待方放行。`obtrudeValue` / `obtrudeException` 不管状态强制覆盖,一般不要在生产乱用。 顺带说一句,`isDone` 不等于"成功"。取消、异常、正常完成都算 done,判断成败要看 `isCompletedExceptionally` 或直接 `join()` 接异常。 ### 案例八:自定义线程池(Async 变体与线程分配) 所有 `xxxAsync` 方法不指定线程池时,都跑在 `ForkJoinPool.commonPool()` 上。这是最容易踩的坑: ```java // 不指定线程池时,默认全部任务都跑在 commonPool 上 CompletableFuture defaultOne = CompletableFuture.supplyAsync(() -> { System.out.println("默认线程池: " + Thread.currentThread().getName()); return "OK"; }); defaultOne.join(); // 重活建议显式指定自定义线程池,避免抢占公共池 ExecutorService pool = Executors.newFixedThreadPool(4); CompletableFuture custom = CompletableFuture.supplyAsync(() -> { System.out.println("自定义池 supplyAsync: " + Thread.currentThread().getName()); return "结果"; }, pool).thenApplyAsync(r -> { System.out.println("自定义池 thenApplyAsync: " + Thread.currentThread().getName()); return r + " 加工完成"; }, pool); System.out.println("最终结果: " + custom.join()); pool.shutdown(); // 关键区别: 不带 Async 后缀的方法,在"触发者"所在的线程执行 CompletableFuture noAsync = CompletableFuture.supplyAsync(() -> { System.out.println("supplyAsync : " + Thread.currentThread().getName()); return "r"; }).thenApply(r -> { System.out.println("thenApply(无Async) : " + Thread.currentThread().getName()); return r; }).thenApplyAsync(r -> { System.out.println("thenApplyAsync(无池): " + Thread.currentThread().getName()); return r; }); noAsync.join(); ``` 典型输出: ``` 默认线程池: ForkJoinPool.commonPool-worker-1 自定义池 supplyAsync: pool-1-thread-1 自定义池 thenApplyAsync: pool-1-thread-2 supplyAsync : ForkJoinPool.commonPool-worker-1 thenApply(无Async) : space.anyi.Cf09_CustomExecutor.main() thenApplyAsync(无池): ForkJoinPool.commonPool-worker-1 ``` `commonPool` 线程数 = CPU核数 - 1,I/O 阻塞任务会把它饿死(Thread Starvation)。不带 `Async` 的方法在"触发线程"上执行,带 `Async` 且不传池子才提交到 `commonPool`。遇到 I/O 阻塞、长耗时任务,务必 `new ThreadPoolExecutor(...)` 并传给带 `Async` 的方法。 ### 案例九:超时控制与异步降级(orTimeout / completeOnTimeout / exceptionallyCompose) 异步调用最怕挂死:下游接口慢、网络闪断,调用方还在傻等。生产里一切异步外呼都必须有超时兜底: ```java // orTimeout: 限时等待,超时后以 TimeoutException 失败 // 注意:它不会取消底层任务,慢任务仍会继续执行 CompletableFuture slow = CompletableFuture.supplyAsync(() -> { sleep(1500); System.out.println("底层慢任务执行完毕(并未被 orTimeout 取消)"); return "慢任务结果"; }); try { slow.orTimeout(500, TimeUnit.MILLISECONDS).join(); } catch (Exception e) { System.out.println("orTimeout 超时失败: " + e.getCause()); } // completeOnTimeout: 超时就给默认值完成,不抛异常 CompletableFuture withDefault = CompletableFuture.supplyAsync(() -> { sleep(1500); return "真实数据"; }).completeOnTimeout("超时兜底数据", 500, TimeUnit.MILLISECONDS); System.out.println("completeOnTimeout 结果: " + withDefault.join()); // 输出: 超时兜底数据 // exceptionallyCompose: 异常后拼接一个异步降级任务,结果自动展平 CompletableFuture recovered = CompletableFuture.supplyAsync(() -> { throw new RuntimeException("主任务失败"); }).exceptionallyCompose(ex -> CompletableFuture.supplyAsync(() -> "降级方案返回的数据")); System.out.println("exceptionallyCompose 降级结果: " + recovered.join()); ``` `orTimeout` 让调用方快速失败,适合"拿不到就报错让上层重试/降级"的业务。`completeOnTimeout` 让调用方拿到兜底值,适合"过期用默认数据也行"的业务(如缓存兜底)。`exceptionallyCompose` 补上了异常处理谱系的最后一块拼图: | 能力 | 正常路径 | 异常路径 | | --- | --- | --- | | 同步续接 | `thenApply` | `exceptionally` | | 异步续接 | `thenCompose` | `exceptionallyCompose` | | 观察兜底 | `whenComplete` / `handle` | — | ## 综合实战:一个完整的异步编排场景 把上面的 API 组合起来,写一个真实场景:电商商品详情页需要并发调用 5 个下游服务,拿到数据后组装成页面 DTO,任何一路失败都给降级默认值,整体有 3 秒超时。 ```java public class ProductDetailPage { private final ExecutorService pool = Executors.newFixedThreadPool(8); public CompletableFuture loadDetail(String productId) { // 五个下游服务并发调用 CompletableFuture infoFuture = CompletableFuture .supplyAsync(() -> remoteCall("商品基础信息", 200), pool) .exceptionally(ex -> new ProductInfo("默认商品", 0.0)); CompletableFuture stockFuture = CompletableFuture .supplyAsync(() -> remoteCall("库存信息", 150), pool) .exceptionally(ex -> new StockInfo(0)); CompletableFuture couponFuture = CompletableFuture .supplyAsync(() -> remoteCall("优惠信息", 300), pool) .exceptionally(ex -> new CouponInfo("无可用优惠")); CompletableFuture sellerFuture = CompletableFuture .supplyAsync(() -> remoteCall("商家信息", 100), pool) .exceptionally(ex -> new SellerInfo("未知商家")); CompletableFuture> commentFuture = CompletableFuture .supplyAsync(() -> remoteCall("评论列表", 250), pool) .exceptionally(ex -> Collections.emptyList()); // allOf 等全部完成,再各自 join 汇总 CompletableFuture all = CompletableFuture.allOf( infoFuture, stockFuture, couponFuture, sellerFuture, commentFuture); // 整体 3 秒超时 return all.orTimeout(3, TimeUnit.SECONDS).thenApply(v -> { ProductInfo info = infoFuture.join(); StockInfo stock = stockFuture.join(); CouponInfo coupon = couponFuture.join(); SellerInfo seller = sellerFuture.join(); List comments = commentFuture.join(); return new ProductDTO(info, stock, coupon, seller, comments); }); } private T remoteCall(String name, long ms) { // 模拟远程调用 sleep(ms); return (T) name + " 数据"; } } ``` 这个例子把前面的 API 串起来了:`supplyAsync` + 自定义线程池(案例八)、`exceptionally` 降级(案例六)、`allOf` 批量并发聚合(案例五)、`orTimeout` 超时快速失败(案例九)、`thenApply` 结果组装(案例一)。 ## 收藏即学会 把速查表截个图存在手机里,遇到"多个异步怎么排"先查它,比背 API 快得多。真正要修炼的,是每写一个回调都问自己:这条链,跑在哪个线程上?会不会饿死公共池?