承接的问题
11.1 讲的是静态结构：ctl 怎么编码、五个状态怎么排序、Worker 为什么是一把不可重入的锁。本节讲动态路径：execute 决定要开新线程之后 addWorker 做了什么、线程创建失败时那个已加的计数谁来减、runWorker 的主循环为什么先跑 firstTask、以及 beforeExecute / afterExecute 抛异常到底会发生什么。
addWorker：先抢名额，再起线程
addWorker 的前半段只做一件事：在 ctl 里把 workerCount 加一。
// JDK 25 源码摘录，已删减
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
for (;;) {
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// ... 创建 Worker 并启动线程
}
两层循环的分工很清楚：外层在状态变化时重读 ctl 重来（摘录省略了它的 SHUTDOWN 检查：池已关闭且队列为空、或已 STOP、或有 firstTask 时直接 return false），内层只处理"计数被别人改过"的 CAS 失败。参数用 boolean core 而不是直接传上限值，源码注释解释了原因——“to ensure reads of fresh values after checking other pool state”，corePoolSize 是 volatile，传值会把调用瞬间的旧值带进循环。
先加计数、再起线程，顺序不能反。 计数是逻辑名额，一次 CAS 就能线性化"还能不能再多一个线程"的判断；创建线程可能耗时、可能失败、可能触发 OOM，先创建再登记的话两个线程同时创建就会超额。名额先占到，创建失败再还回去——addWorkerFailed 在 mainLock 里把 worker 从集合移除、decrementWorkerCount()，再调一次 tryTerminate()：这个失败的新 worker 可能正是"最后一个挡住终止的计数"。Worker 的构造函数里也有一个失败点：getThreadFactory().newThread(this) 允许返回 null，addWorker 用 if (t != null) 挡住后走 addWorkerFailed。所以"工厂返回 null“是被支持的行为，后果由调用链决定。本机实测（反射读 ctl，--add-opens java.base/java.util.concurrent=ALL-UNNAMED）：
import java.lang.reflect.Field;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/** 工厂返回 null 时：execute 是否失败、任务去了哪里、计数是否回滚。 */
public final class NullFactoryProbe {
public static void main(String[] args) throws Exception {
ThreadPoolExecutor p = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(8), r -> null);
for (int i = 1; i <= 3; i++) {
p.execute(() -> System.out.println("这行永远不会打印"));
System.out.println("第 " + i + " 次 execute 正常返回");
}
Field f = ThreadPoolExecutor.class.getDeclaredField("ctl");
f.setAccessible(true);
int ctl = ((AtomicInteger) f.get(p)).get();
System.out.println("workerCount=" + (ctl & ((1 << 29) - 1))
+ " poolSize=" + p.getPoolSize() + " 队列长度=" + p.getQueue().size());
}
}
用 javac --release 25 编译，再以 --add-opens java.base/java.util.concurrent=ALL-UNNAMED 运行，本机输出：
第 1 次 execute 正常返回
第 2 次 execute 正常返回
第 3 次 execute 正常返回
workerCount=0 poolSize=0 队列长度=3
工厂返回 null 时 execute 正常返回，三个任务静静躺在队列里，workerCount 是 0，永远不会有线程来取。 回滚是成功的（计数回到 0，没有泄漏），但任务也没有被拒绝。
原因在 execute 的后半段：第一次 addWorker(command, true) 失败后，它继续尝试入队，入队成功后重新检查计数，else if (workerCountOf(recheck) == 0) addWorker(null, false);——这一行的返回值被丢弃了。这是有意为之：任务已经入队，补建线程再失败也没有更好的处理方式，只能等下一次 execute 再试；代价是任务可能无限期滞留。同样是"工厂坏了”，返回 null 是静默丢任务，抛异常则从 addWorker 的 finally 原样上抛、计数回滚、队列保持空——后者才是可运维的行为。
runWorker：先跑 firstTask，再循环 getTask
名额抢到、线程起来之后，Worker.run() 只有一行 runWorker(this)：
// JDK 25 源码摘录，已删减
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
try {
task.run();
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
firstTask立刻置null。 任务对象可能很大，尽早断开引用，避免整个执行期间都被Worker持有。w.unlock()是"允许中断"的开关。 11.1 说过构造时setState(-1)抑制中断，这一行把它清回 0；从这以后interruptIfStarted才会真的interrupt。completedAbruptly初值true，只有循环正常结束才置false。 它决定退出时要不要补位，是整个异常处理的关键；w.lock()在beforeExecute之前，进入临界区就意味着"我在跑任务"，池的中断只找没持锁的线程。- STOP 检查里
ctl.get()读了两次。 第一次判断是否已 STOP；没停就先Thread.interrupted()清掉中断标志，再读一次——清除标志的瞬间池可能刚好被shutdownNow，不重读就会漏掉"任务开始时中断标志被清掉、池却已要求中断"的情况。
取任务在 getTask：
// JDK 25 源码摘录，已删减
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
timed 决定用 poll 还是 take：核心线程默认无限期阻塞在 take()，超过核心数的线程和开了 allowCoreThreadTimeOut 的核心线程用 poll(keepAliveTime) 带超时。timedOut 是"上一轮超时了"的标记，它和 wc > 1 || workQueue.isEmpty() 一起保证不会把最后一个线程也裁掉——队列里还有任务时必须留人。InterruptedException 被吞掉并只重置 timedOut，因为中断在这里是正常信号：shutdown 会中断空闲线程，让它回到循环顶部重新看状态。return null 之前已经做过 compareAndDecrementWorkerCount，所以正常退出路径上计数已经减过；processWorkerExit 只在异常退出时补减。
钩子：beforeExecute 与 afterExecute
两个空方法的调用位置决定了三个容易被忽略的后果。下面用自定义 ThreadFactory 命名线程，用钩子打印完整生命周期：
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/** 自定义 ThreadFactory + 钩子，观察 worker 生命周期与异常路径。 */
public final class LifecycleProbe {
static final AtomicInteger SEQ = new AtomicInteger();
static void log(String s) {
System.out.println("[" + Thread.currentThread().getName() + "] " + s);
}
static ThreadFactory factory() {
return r -> {
Thread t = new Thread(r, "libd-pool-" + SEQ.incrementAndGet());
t.setUncaughtExceptionHandler((th, e) -> System.out.println("[" + th.getName() + "] uncaught " + e));
return t;
};
}
/** failOn 为任务编号，命中时 beforeExecute 抛异常；-1 表示不抛。 */
static ThreadPoolExecutor pool(int failOn) {
return new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(4), factory()) {
@Override
protected void beforeExecute(Thread t, Runnable r) {
int id = ((Task) r).id;
log("beforeExecute task-" + id);
if (id == failOn) throw new IllegalStateException("beforeExecute boom on task-" + id);
}
@Override
protected void afterExecute(Runnable r, Throwable ex) {
log("afterExecute task-" + ((Task) r).id + " ex=" + (ex == null ? "null" : ex.getClass().getSimpleName()));
}
};
}
record Task(int id) implements Runnable {
public void run() {
log("run task-" + id + " 开始");
try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
log("run task-" + id + " 结束");
}
}
public static void main(String[] args) throws Exception {
System.out.println("=== 场景 A：正常执行 ===");
ThreadPoolExecutor a = pool(-1);
for (int i = 1; i <= 3; i++) a.execute(new Task(i));
a.shutdown();
a.awaitTermination(5, TimeUnit.SECONDS);
SEQ.set(0);
System.out.println();
System.out.println("=== 场景 B：task-2 的 beforeExecute 抛异常 ===");
ThreadPoolExecutor b = pool(2);
for (int i = 1; i <= 3; i++) b.execute(new Task(i));
b.shutdown();
b.awaitTermination(5, TimeUnit.SECONDS);
System.out.println("completedTaskCount=" + b.getCompletedTaskCount());
}
}
本机输出：
=== 场景 A：正常执行 ===
[libd-pool-1] beforeExecute task-1
[libd-pool-1] run task-1 开始
[libd-pool-1] run task-1 结束
[libd-pool-1] afterExecute task-1 ex=null
[libd-pool-1] beforeExecute task-2
[libd-pool-1] run task-2 开始
[libd-pool-1] run task-2 结束
[libd-pool-1] afterExecute task-2 ex=null
[libd-pool-1] beforeExecute task-3
[libd-pool-1] run task-3 开始
[libd-pool-1] run task-3 结束
[libd-pool-1] afterExecute task-3 ex=null
=== 场景 B：task-2 的 beforeExecute 抛异常 ===
[libd-pool-1] beforeExecute task-1
[libd-pool-1] run task-1 开始
[libd-pool-1] run task-1 结束
[libd-pool-1] afterExecute task-1 ex=null
[libd-pool-1] beforeExecute task-2
[libd-pool-1] uncaught java.lang.IllegalStateException: beforeExecute boom on task-2
[libd-pool-2] beforeExecute task-3
[libd-pool-2] run task-3 开始
[libd-pool-2] run task-3 结束
[libd-pool-2] afterExecute task-3 ex=null
completedTaskCount=3
场景 A 说明线程确实被复用：三个任务全部跑在 libd-pool-1 上，core=1 的池不会为每个任务新建线程。场景 B 的信息量大得多（uncaught 行与 [libd-pool-2] beforeExecute task-3 行的先后取决于调度，重复运行会互换）：
task-2的run一行都没有——beforeExecute抛异常时任务体根本没执行；它的afterExecute也没被调用，因为它写在beforeExecute之后的内层try里，外层finally只做task = null、completedTasks++、unlock。- 线程
libd-pool-1死了（uncaught打到它的 handler），processWorkerExit(w, true)看到completedAbruptly == true，补了一个新 worker，task-3跑在libd-pool-2上。 completedTaskCount=3，但真正执行完的任务只有 2 个。w.completedTasks++在外层finally里，只要进过循环体就会加一，哪怕任务一行没跑。
这四点合起来是一条实践规则：beforeExecute 里只做记账、打点这类不会抛异常的事。 一旦抛出，池的行为是"杀掉线程、补一个新线程、丢掉当前任务"，而统计数字还会告诉你任务完成了。
afterExecute 有它自己的坑：源码注释引用了 JLS 14.20——任务抛了异常、afterExecute 又抛了一个时，最终生效的是 afterExecute 的那个。任务异常只能作为参数传进 afterExecute，然后被它自己的异常覆盖，UncaughtExceptionHandler 只看得见后者。所以 afterExecute 里记录异常时要自己吞掉（至少 log 一次），绝不能在处理异常的分支里再抛。
processWorkerExit 与 tryTerminate 的收尾
循环结束（getTask 返回 null，或异常抛出）后一定走 processWorkerExit：
// JDK 25 源码摘录，已删减
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
// 持 mainLock 把 w.completedTasks 累加进 completedTaskCount、从 workers 移除
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
补位规则可以写成一句话：异常退出必补；正常退出时，只有"池还没到 STOP"且"当前线程数低于下限"才补。 下限默认是 corePoolSize，开了核心线程超时则为 0；但下限是 0 而队列还有任务时至少要补 1 个，避免任务无人消费。
tryTerminate 的收尾顺序值得单独记：先判断是否"该终止"，该终止但还有 worker 就只中断一个空闲线程去传棒；计数也归零后，在 mainLock 里 CAS 到 TIDYING，执行 terminated() 钩子，最后 ctl.set(ctlOf(TERMINATED, 0))、termination.signalAll()、container.close()。awaitTermination 等的是最后这个 signalAll，所以它一定在 terminated() 之后才返回。
在 libraryd-conc 里落地
PooledLoanService 需要池内执行耗时。只有钩子能低成本拿到，但钩子必须"只记账、不抛异常"：
package com.plume.conc.core;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
/** 借阅池的池内计时。ThreadLocal 必须 remove，否则 worker 复用时会把旧值带给下一个任务。 */
public final class TimedLoanPool extends ThreadPoolExecutor {
private static final ThreadLocal<Long> START = new ThreadLocal<>();
private final LongAdder runNanos = new LongAdder();
private final LongAdder tasks = new LongAdder();
public TimedLoanPool(int core, int max, BlockingQueue<Runnable> q, ThreadFactory f) {
super(core, max, 60, TimeUnit.SECONDS, q, f);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
START.set(System.nanoTime());
}
@Override
protected void afterExecute(Runnable r, Throwable ex) {
Long start = START.get();
START.remove();
if (start != null) { runNanos.add(System.nanoTime() - start); tasks.increment(); }
if (ex != null) System.err.println("task failed: " + ex);
}
public long averageRunMillis() {
long n = tasks.sum();
return n == 0 ? 0 : runNanos.sum() / n / 1_000_000;
}
}
三个约束写进注释：钩子跑在 worker 线程上，任何阻塞都会直接拖慢任务；afterExecute 不要重抛，否则按 JLS 14.20 会覆盖任务异常并再搭上一个线程；ThreadLocal 用完必须 remove，worker 线程是复用的，不清理会把上一个任务的状态带给下一个（卷一 7.1 讲过的泄漏形态在这里同样成立）。线程工厂仍用卷一第 10 章那套命名加 UncaughtExceptionHandler，这样"线程悄悄死掉又被替换"才有日志可查。
这个设计的代价
第一，钩子异常等于丢任务。 beforeExecute 抛出后任务不执行、不拒绝、不重试，只有 completedTaskCount 默默加一。把鉴权、参数校验、事务开启放进 beforeExecute 是在给自己埋雷。
第二，异常退出会引发线程抖动。 每次任务抛异常，worker 就死一个、补一个，ThreadFactory 被再调一次。任务持续失败时池会不停创建销毁线程，日志里满屏 uncaught。要么在任务内部兜住异常，要么接受这份抖动并把它当告警信号。
第三，addWorker 的返回值在入队路径上被丢弃。 这不是 bug，但它意味着"入队成功"不等于"有人会执行"。工厂坏了、线程创建持续失败时任务会安静堆积，直到队列满才表现为拒绝；监控要同时看队列长度和 poolSize，只看拒绝率会发现得太晚。
第四，completedTaskCount 不是"成功任务数"。 它统计的是"进入过循环体的次数"，包含失败、包含 beforeExecute 抛出的那些，也不包含还在队列里的，拿它算成功率一定是错的。也不要为了拿指标去重写 execute：重写会绕过 addWorker 内部"先加计数再建线程"的原子性，也可能在 shutdown 之后仍然接受任务。要统计就用钩子，要限流就用有界队列加拒绝策略——这两条卷一第 10 章都已经铺好了。
小结
addWorker 先用一次 CAS 抢到 workerCount 名额，再创建 Worker 和线程；创建失败由 addWorkerFailed 回滚计数并重新检查终止条件。runWorker 先跑 firstTask 再循环 getTask，completedAbruptly 决定退出后是否补位。钩子的调用位置带来两个硬约束：beforeExecute 抛异常等于丢任务且统计失真，afterExecute 抛异常会覆盖任务异常。
下一节转向定时任务的容器：DelayedWorkQueue 如何用一个堆数组加 leader 字段，让多个 worker 等到期任务时只有一个真正带超时等待。
阅读导航：上一节 11.1 ctl 与 Worker 生命周期 ｜ 下一节 11.3 DelayedWorkQueue
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。