承接的问题
13.1 看到 thenApply 的回调在调用 complete 的线程上跑完了。这留下三个必须回答的问题，因为它们直接决定线上会不会被一条慢链拖死：
thenApplyAsync不传执行器时，任务到底进了哪个池？defaultExecutor()和ForkJoinPool.commonPool()是同一个东西吗？thenApply（没有 executor）是不是就一定在完成者线程上跑？
第三个问题的答案会让你意外：不一定。先看源码里的三档模式，再看两个实验。
三档模式：谁来执行
Completion.tryFire(int mode) 的 mode 只有三个取值，符号是有意义的：
// JDK 25 源码摘录，已删减
private static final ForkJoinPool ASYNC_POOL =
ForkJoinPool.asyncCommonPool(); // ensures minimal parallelism
// Modes for Completion.tryFire. Signedness matters.
static final int SYNC = 0;
static final int ASYNC = 1;
static final int NESTED = -1;
public Executor defaultExecutor() {
return ASYNC_POOL;
}
| 模式 | 值 | 谁调用 | 行为 |
|---|---|---|---|
SYNC | 0 | unipush / bipush / orpush 发现源已完成 | 就地执行，但先 claim() |
ASYNC | 1 | Completion.run() / exec() | 跳过 claim()，直接执行 |
NESTED | -1 | postComplete() | 就地执行并 claim()，结果交回外层循环 |
三个模式在 UniApply.tryFire 里汇成一句判断：
// JDK 25 源码摘录，已删减
try {
if (mode <= 0 && !claim())
return null;
else {
@SuppressWarnings("unchecked") T t = (T) r;
d.completeValue(f.apply(t));
}
} catch (Throwable ex) {
d.completeThrowable(ex);
}
mode <= 0 覆盖 SYNC 与 NESTED，两者都必须先 claim()；mode > 0（ASYNC）跳过认领，因为它已经在执行器线程上了。而 claim() 在 executor != null 时并不执行，而是把节点 e.execute(this) 扔出去并返回 false——所以带 executor 的节点无论从哪条路径被触发，最终都落在那个执行器上，这是 *Async 唯一的可靠保证。
*Async 不传执行器时落到哪
thenApply 与 thenApplyAsync 的差别只有第一个参数：
// JDK 25 源码摘录，已删减
public <U> CompletableFuture<U> thenApply(
Function<? super T,? extends U> fn) {
return uniApplyStage(null, fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn) {
return uniApplyStage(defaultExecutor(), fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn, Executor executor) {
return uniApplyStage(Objects.requireNonNull(executor), fn);
}
defaultExecutor() 返回 ASYNC_POOL，而 ASYNC_POOL 来自 ForkJoinPool.asyncCommonPool()：
// JDK 25 源码摘录，已删减
static ForkJoinPool asyncCommonPool() {
ForkJoinPool cp; int p;
if ((p = (cp = common).parallelism) == 0)
U.compareAndSetInt(cp, PARALLELISM, 0, 2);
return cp;
}
它返回的就是 ForkJoinPool.common 本身——不是另建一个池。唯一的分支是：如果 common pool 被配置成并行度 0（-Djava.util.concurrent.ForkJoinPool.common.parallelism=0，意味着没有 worker），它用 CAS 把内部字段从 0 抬到 2，保证 CF 的默认执行器至少能跑。
本机实测（-Djava.util.concurrent.ForkJoinPool.common.parallelism=0）：
触到 CF 之前 getCommonPoolParallelism = 1
supplyAsync 跑在 = ForkJoinPool.commonPool-worker-1
触到 CF 之后 getCommonPoolParallelism = 2
第一行的 1 不是笔误：getParallelism() 的实现是 Math.max(getParallelismOpaque(), 1)，把 0 报成 1；而 asyncCommonPool() 改的是原始字段，所以触到 CF 之后读到的是真实的 2。
defaultExecutor 存在的意义
既然默认就是 common pool，为什么还要暴露 defaultExecutor()？因为它是一个可覆盖点。源码注释给的示例是子类改写 newIncompleteFuture() 与 defaultExecutor()，让整条链默认走自己的池：
class MyFuture<T> extends CompletableFuture<T> {
private final Executor pool;
MyFuture(Executor pool) { this.pool = pool; }
@Override public <U> CompletableFuture<U> newIncompleteFuture() {
return new MyFuture<U>(pool);
}
@Override public Executor defaultExecutor() {
return pool;
}
}
newIncompleteFuture() 决定 thenApply 返回的下游对象是什么类型，defaultExecutor() 决定它的异步阶段去哪个池。两者一起改写，才能让链上每一跳都留在自己的类型和池里；只改 defaultExecutor()，下游仍然退回普通 CompletableFuture。这条路径在框架里有用，业务代码里通常直接用 thenApplyAsync(fn, executor) 更清楚。
commonPool 的并行度与阻塞调用
common pool 的默认并行度是 availableProcessors() - 1（源码：pc = Math.max(1, Runtime.getRuntime().availableProcessors() - 1)）。本机 10 核，所以是 9。
它不是 CF 专用的：并行流、Arrays.parallelSort、ForkJoinTask 都用同一个池。把阻塞任务丢进去，占的是全 JVM 共享的 9 个位置。本机实验：提交 9 个阻塞任务占满后，第 10 个 runAsync 在 800ms 内没有开始；释放后立刻开始。
availableProcessors=10 getCommonPoolParallelism=9
占满 9 个 worker 后新任务 800ms 内开始 = false
有一个容易被误传的细节：ForkJoinPool 确实支持阻塞补偿，但只对声明过的阻塞生效。CompletableFuture.join() 走的是 Signaller implements ForkJoinPool.ManagedBlocker，waitingGet 里调用 ForkJoinPool.managedBlock(q)，所以 worker 在 join 上阻塞时池子知道，可以补线程。但 Thread.sleep、JDBC、HTTP 客户端、文件 IO 都不是 ManagedBlocker，池子不知道你在阻塞，只会干等。这就是"CF 里调阻塞 API 必须显式传自己的执行器"的根本原因——不是因为 common pool 小，而是因为它无法识别你的阻塞。
自定义执行器与"帮忙倒栈"
先看线程归属。thenApply 用 null executor，thenApplyAsync 用默认池，thenApplyAsync(fn, notify) 用自定义的两线程池：
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/** 观察 thenApply / thenApplyAsync 的线程归属，以及 commonPool 被占满时的行为。 */
public final class ExecProbe {
static String who() { return Thread.currentThread().getName(); }
public static void main(String[] args) throws Exception {
System.out.printf("availableProcessors=%d getCommonPoolParallelism=%d%n",
Runtime.getRuntime().availableProcessors(), ForkJoinPool.getCommonPoolParallelism());
System.out.println("defaultExecutor() == ForkJoinPool.commonPool() -> "
+ (new CompletableFuture<Void>().defaultExecutor() == ForkJoinPool.commonPool()));
AtomicInteger seq = new AtomicInteger();
ExecutorService notify = Executors.newFixedThreadPool(2,
r -> new Thread(r, "notify-" + seq.incrementAndGet()));
CompletableFuture<String> src = new CompletableFuture<>();
CompletableFuture<String> mail = src.thenApply(s -> "mail@" + who());
CompletableFuture<String> sms = src.thenApplyAsync(s -> "sms@" + who());
CompletableFuture<String> push = src.thenApplyAsync(s -> "push@" + who(), notify);
Thread producer = new Thread(() -> src.complete("v"), "producer");
producer.start();
producer.join();
System.out.println("thenApply -> " + mail.join());
System.out.println("thenApplyAsync -> " + sms.join());
System.out.println("thenApplyAsync(io) -> " + push.join());
int p = ForkJoinPool.getCommonPoolParallelism();
CountDownLatch occupied = new CountDownLatch(p);
CountDownLatch release = new CountDownLatch(1);
CountDownLatch late = new CountDownLatch(1);
for (int i = 0; i < p; i++) {
CompletableFuture.runAsync(() -> {
occupied.countDown();
try { release.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
});
}
occupied.await(10, TimeUnit.SECONDS);
CompletableFuture<Void> pending = CompletableFuture.runAsync(late::countDown);
System.out.println("占满 " + p + " 个 worker 后新任务 800ms 内开始 = "
+ late.await(800, TimeUnit.MILLISECONDS));
release.countDown();
pending.join();
notify.shutdown();
}
}
javac --release 25 -d out ExecProbe.java && java -cp out ExecProbe
本机输出（连跑 10 次，thenApply 那一行 10 次都是 mail@producer）：
availableProcessors=10 getCommonPoolParallelism=9
defaultExecutor() == ForkJoinPool.commonPool() -> true
thenApply -> mail@producer
thenApplyAsync -> sms@ForkJoinPool.commonPool-worker-1
thenApplyAsync(io) -> push@notify-1
占满 9 个 worker 后新任务 800ms 内开始 = false
看起来很干净：同步的在 producer，默认异步的在 commonPool-worker-1，自定义的在 notify-1。但把节点数放大，这个"干净"就碎了。
先完成的异步阶段会帮忙倒源栈
同一个源 future 上挂 200 个 thenApply 和 1 个 thenApplyAsync(fn, io)，其中 io 是单线程池：
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/** 验证 postFire 让先完成的异步阶段帮忙倒源 future 的栈。 */
public final class HelpProbe {
static String who() { return Thread.currentThread().getName(); }
public static void main(String[] args) throws Exception {
ExecutorService io = Executors.newSingleThreadExecutor(r -> new Thread(r, "io-1"));
CompletableFuture<String> src = new CompletableFuture<>();
List<CompletableFuture<String>> plain = new ArrayList<>();
for (int i = 0; i < 200; i++) plain.add(src.thenApply(s -> "plain@" + who()));
CompletableFuture<String> once = src.thenApplyAsync(s -> "async@" + who(), io);
Thread producer = new Thread(() -> src.complete("v"), "producer");
producer.start();
producer.join();
once.join();
Map<String, Integer> tally = new TreeMap<>();
for (CompletableFuture<String> f : plain) tally.merge(f.join(), 1, Integer::sum);
System.out.println("200 个 thenApply 的归属: " + tally);
System.out.println("thenApplyAsync 的归属: " + once.join());
io.shutdown();
}
}
本机输出（连跑 6 次，每次比例不同）：
200 个 thenApply 的归属: {plain@io-1=83, plain@producer=117}
thenApplyAsync 的归属: async@io-1
200 个 thenApply 的归属: {plain@io-1=112, plain@producer=88}
thenApplyAsync 的归属: async@io-1
200 个 thenApply 的归属: {plain@io-1=102, plain@producer=98}
thenApplyAsync 的归属: async@io-1
200 个没有指定执行器的 thenApply，大约一半跑在了 io-1 上。 这个池子是那个 thenApplyAsync 的执行器，和这 200 个节点毫无关系。原因在 postFire 的开头：
// JDK 25 源码摘录，已删减
final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
if (a != null && a.stack != null) {
Object r;
if ((r = a.result) == null)
a.cleanStack();
if (mode >= 0 && (r != null || a.result != null))
a.postComplete();
}
...
}
a 是源 future。异步阶段在 io-1 上跑完时 mode == ASYNC(1) >= 0，于是它在 io-1 上调用 a.postComplete()——顺手帮源 future 把剩下的依赖栈也倒了。producer 此刻也在倒同一个栈，两边用 STACK 的 CAS 争抢节点，谁抢到谁执行。
这件事有三个直接后果：
thenApply（不带 executor）的执行线程不保证是完成者。 它只保证"在某个正在推进这条链的线程上"，可能是 producer，也可能是任何先完成的异步兄弟的执行器线程。- 借来的线程可能被长时间占用。 那 200 个回调如果在
io-1上各跑 10ms，这个单线程池就被别人的段耽误了 1 秒——而io-1的所有者以为它只跑一个任务。 - 反过来说，这也是吞吐优化。 完成者不必独自把整棵依赖树走完，先空出来的线程可以分担。对小扇出是好事，对大扇出是麻烦。
结论很实际：不要依赖"哪个线程执行我的非异步阶段"。要固定线程归属，就显式 thenApplyAsync(fn, executor)；要不关心线程，就别在回调里放阻塞代码。
在 libraryd-conc 里落地
NotifyOrchestrator 的三个下游是网络调用，绝不能落在 common pool 上。执行器作为构造参数传进来，每个下游都显式指定：
package com.plume.conc.core;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
/** 通知扇出：所有下游都走同一个有界 IO 池，不碰 commonPool。 */
public final class NotifyFanout {
public record Delivery(String channel, boolean ok, String detail) {}
public interface Notifier {
Delivery send(Loan loan);
}
private final Executor io;
private final List<String> channels = List.of("mail", "sms", "push");
private final java.util.Map<String, Notifier> notifiers;
public NotifyFanout(Executor io, java.util.Map<String, Notifier> notifiers) {
this.io = io;
this.notifiers = notifiers;
}
public CompletableFuture<List<Delivery>> fanOut(Loan loan) {
List<CompletableFuture<Delivery>> parts = channels.stream()
.map(ch -> CompletableFuture
.supplyAsync(() -> notifiers.get(ch).send(loan), io)
.exceptionally(ex -> new Delivery(ch, false, "failed")))
.toList();
return CompletableFuture.allOf(parts.toArray(CompletableFuture[]::new))
.thenApply(ignored -> parts.stream().map(CompletableFuture::join).toList());
}
}
装配侧的关键不是这段代码，而是 io 从哪来：
int cores = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor io = new ThreadPoolExecutor(
cores, cores * 2, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(256),
Thread.ofPlatform().name("notify-", 0).factory(),
new ThreadPoolExecutor.CallerRunsPolicy());
四个决定：核心线程数与 CPU 数挂钩（下游是 IO 等待，可以略多于核数，但不能无限）；队列有界 256（无界队列会把超时压力变成内存压力）；线程名带 notify- 前缀（线程转储里一眼能认出来）；拒绝策略用 CallerRunsPolicy（打满时让提交者自己跑，形成天然背压，而不是无声丢弃）。
这里还有一个细节值得写进注释：thenApply(ignored -> ...) 这一段没有指定执行器，会由"最后完成的那一个下游的执行器线程"来跑——也就是 io 里的某条线程。因为它只做三次 join（此时必然已完成）和一次 toList，耗时可以忽略，不需要再切一次线程。判断要不要 *Async 的标准不是"它可能跑在别的线程"，而是"这段代码会不会占用那条线程很久"。
什么时候不要用
第一，不要在 common pool 上做任何可能阻塞的事。 包括 thenApplyAsync 不传执行器、supplyAsync 不传执行器、runAsync 不传执行器。它不是 CF 私有的池，占满它等于同时拖慢全 JVM 的并行流。
第二，不要用 Executors.newCachedThreadPool() 当 CF 的执行器。 它无上限创建线程，下游一慢就是线程数暴涨；CF 的超时、取消都不会回收已经提交的任务，池子只会越来越大。要有界，且要显式拒绝策略。
第三，不要在一条链上反复换执行器。 每换一次就是一次跨线程交接，缓存局部性、ThreadLocal、MDC 日志上下文全部丢失。一条链固定一个执行器，需要隔离的边界（比如"写库"和"发通知"）才分开。
第四，简单的一次并行调用不必用 CF。 三个独立调用等全部返回，用 CF 要建至少 5 个对象、维护一棵 allOf 归并树；如果只是"并发发起、全部等待"，普通 invokeAll 或虚拟线程加 join 更直接。CF 的价值在组合——链、分支、任一、异常恢复——而不是"并发跑几个任务"。
小结
thenApplyAsync 不传执行器时用的是 ASYNC_POOL，它就是 ForkJoinPool.commonPool()，只在并行度为 0 时被 asyncCommonPool() 抬到 2。defaultExecutor() 是给子类覆盖的默认值入口，改它要连同 newIncompleteFuture() 一起改。common pool 默认 CPU-1 个 worker，与并行流共享，阻塞补偿只对 ManagedBlocker 生效，sleep/JDBC/HTTP 都不算。SYNC/ASYNC/NESTED 三档模式决定"是否先认领、能否就地跑"，而 postFire 会让先完成的异步阶段调用源 future 的 postComplete()，于是不带执行器的阶段可能被借到任意一条正在推进这条链的线程上——实验中 200 个 thenApply 有一半跑到兄弟阶段的 io-1。要确定的线程归属，就显式传执行器。
阅读导航：上一节 13.1 CompletableFuture 的依赖栈 ｜ 下一节 13.3 异常传播与取消
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。