CompletableFuture：把“做完以后干什么”挂在结果上

• 线程池 + Future：pool.submit() 能并行，但拿结果只能 future.get()，在请求线程里一等就是 300 ms，异步又变回了同步。
• 没法说“完了以后做什么”：想在三条都发完后汇总，只能一个个 get；想在失败时补发，只能 try/catch ExecutionException；想“最多等 500 ms”，只能每个 get 都带超时。
• 自己写回调：任务里调 listener，listener 里再提交下一个任务，几层以后就是回调地狱，异常也不知道该往哪抛。
• 缺的那一块：一个结果格子，可以提前挂上“结果出来以后做什么”，还能把几个格子组合起来。这就是 CompletableFuture（JDK 8）。

不懂这一点时的典型误用：把 supplyAsync 当成“开个线程跑一下”，不传线程池、不处理异常。实测 在 2 核机器上默认池会为每个任务新建一个线程；任务里抛的异常没人 join，就什么都看不到（⑤）。

选择规则：普通 Spring MVC 项目里“并行调几个下游、合并、兜底”，用 CompletableFuture 加自己的线程池；整条链都是响应式才用 Reactor；JDK 21 以上可以考虑虚拟线程的同步写法。本页学 CompletableFuture。

④-1结构：result 和 stack 两个字段

④-2回调在哪个线程跑：先猜再看

• 1thenApplyAsync(fn, pool)：一定交给 pool。
• 2thenApplyAsync(fn)：交给默认池（④-4）。
• 3thenApply(fn)：前一步已经完成，调用 thenApply 的线程当场执行；还没完成，挂到栈上，由完成前一步的那个线程执行。所以同一行代码，跑在哪个线程上取决于时机。

逻辑按 JDK 17 源码的 uniApplyStage（已完成就走 uniApplyNow）、postComplete、ASYNC_POOL 写成，在 node 里把所有组合跑过一遍；⑤ 的 WhichThread 在本机实测了其中六种。“还没完成”时有个竞态：挂回调的同时前一步刚好完成，回调可能由调用线程执行，结论是“说不准”，所以别依赖它。

④-3编排和异常：最常用的几个方法

• 1异常会被包一层：任务里抛的 IllegalStateException，往下传的时候包成 CompletionException（源码 encodeThrowable）。在 exceptionally 里要看 ex.getCause()。
• 2join 和 get：join() 抛不需要声明的 CompletionException，适合写在 lambda 里；get() 抛 ExecutionException 和 InterruptedException，能带超时。
• 3异常沿链往下跳：前一步失败了，后面的 thenApply 都不执行，直到遇到 exceptionally / handle / whenComplete。链的末尾没有它们、又没人 join，异常就悄悄没了。

④-4默认池和超时：两个容易想错的地方

• 1默认池：不传池子时用 ForkJoinPool.commonPool()，并行度默认是“核数 − 1”。并行度不大于 1 时（比如 2 核机器），JDK 换成 ThreadPerTaskExecutor：每个任务 new 一个线程，不复用、没有上限。全景第 10 站的 Thread-552 就是这么来的。commonPool 是整个 JVM 共用的，parallelStream 也用它。
• 2超时不是取消：orTimeout / completeOnTimeout 只是到点由一个叫 CompletableFutureDelayScheduler 的单线程定时器把 future 标成完成。原来那个任务还在跑，占着线程；cancel(true) 也不会中断它（javadoc：参数 mayInterruptIfRunning 在这个类里不起作用）。

④-5为什么这样设计：谁完成，谁顺手执行

决定胜负的约束：链上大多数回调都很短，而 Java 的线程切换不便宜（线程页实测一次交接 3～4 µs）。所以 JDK 默认“就地执行”，把选择权交给你：回调慢、会阻塞，或者不能占用当前线程时，用 Async 并传自己的池。

CompletableFuture<String> sms = CompletableFuture
    .supplyAsync(() -> smsGateway.send(order), notifyPool)        // 一定传自己的池
    .completeOnTimeout("超时，进补发队列", 500, MILLISECONDS)       // JDK 9+
    .exceptionally(ex -> { retryLater(order); return "失败：" + ex.getCause(); });
CompletableFuture<String> inbox  = CompletableFuture.supplyAsync(() -> inbox.send(order), notifyPool);
CompletableFuture<String> points = CompletableFuture.supplyAsync(() -> points.add(order), notifyPool);

CompletableFuture.allOf(sms, inbox, points)
    .thenRun(() -> audit(order, sms.join(), inbox.join(), points.join()))   // 都完成了，join 不会等
    .whenComplete((v, ex) -> { if (ex != null) log.error("通知汇总失败", ex); });
// 请求线程到这里就返回响应了，不 join

ThreadLocal 怎么带、JDK 8 没有 orTimeout 怎么办、Spring 的 @Async、虚拟线程，放到第⑥步。

实验 1：同一个 thenApply，跑在哪个线程

$ java -cp … WhichThread                      # 本机 2 核
CPU 核数 2，commonPool 并行度 1
1 supplyAsync 默认池              → Thread-0
2 thenApply，前一步还没完成        → Thread-1（完成前一步的线程）
3 thenApply，前一步已经完成        → main（调用 thenApply 的线程）
4 thenApplyAsync                   → Thread-2
5 thenApplyAsync(自己的池)         → notify-1
6 orTimeout 超时后的 exceptionally → CompletableFutureDelayScheduler（TimeoutException）

$ java -Djava.util.concurrent.ForkJoinPool.common.parallelism=4 … WhichThread
1 supplyAsync 默认池              → ForkJoinPool.commonPool-worker-1
4 thenApplyAsync                   → ForkJoinPool.commonPool-worker-1
（3、5、6 不变）

$ java … WhichThread block                  # 8 个阻塞 200 ms 的任务丢进默认池
并行度 1（2 核默认）：204 ms，用了 8 个线程（每个任务 new 一个）
并行度 2：            806 ms，用了 2 个线程
并行度 4：            405 ms，用了 4 个线程

• 12 核机器上默认池根本不是池：Thread-0、Thread-1、Thread-2 每次都是新线程。到了 8 核的生产机，就变成 7 个共享的 commonPool 线程。
• 2超时以后，非 Async 的回调跑在 JDK 内部那个单线程定时器上。在这里做慢操作，全 JVM 所有 future 的超时都会被拖晚。
• 3阻塞任务丢进 commonPool：线程数就是并行度，8 个 200 ms 的任务，2 个线程要 806 ms。生产上就是一个慢下游把整个 JVM 的 parallelStream 和异步任务都拖慢。

实验 2：第 10 站的通知，五个场景

$ java -cp … Notify
1 串行：短信✓ 站内信✓ 积分✓，603 ms
2 并行 allOf：短信✓ 站内信✓ 积分✓，320 ms
3 runAsync 里抛了异常，没人 join：上面什么都没打印
  加 exceptionally：短信✗ 记下来补发（CompletionException: 短信网关 503）
  join 抛的是 CompletionException，cause 才是 IllegalStateException
4 completeOnTimeout：第 503 ms 拿到 "短信⏳ 超时，进补发队列"
  …第 1501 ms：卡住的短信任务其实还在跑，现在才结束（短信✓）
5 主线程 USER=1024，异步线程 notify-4 里 USER=null
  包一层 withUser：异步线程 notify-2 里 USER=1024
  包装用完后同一个池里 USER=null（finally 清掉了，没串号）

• 1并行：总耗时约等于最慢的短信 300 ms，再加上线程调度的二十来毫秒。
• 2异常：exceptionally 拿到的 ex 是 CompletionException，真正的原因在 getCause() 里。
• 3超时：500 ms 拿到了兜底值，但短信任务跑满了 1500 ms，而且最后短信其实发出去了。所以补发前要先查状态，否则用户会收到两条。
• 4上下文：withUser 在提交时抓取 USER，执行前 set、finally 里恢复，就是上一页说的 TaskDecorator 思路。

• CompletableFuture 源码：tryFire 的 SYNC / ASYNC / NESTED 三种模式，postComplete 怎么避免长链把栈压爆
• StructuredTaskScope：虚拟线程下的“并行、统一取消”（第 9 个知识点）
• 异步调用链的 trace：traceId 怎么跨线程池、跨服务传下去（J13）
• ForkJoinPool 的工作窃取，和 ManagedBlocker 为什么能让 commonPool 在阻塞时临时多开线程
• 和 Reactor 对照：同一个“并行三个下游”在 Mono.zip 里怎么写，背压从哪来

自测