ThreadPoolExecutor · JUC
ThreadPoolExecutor 源码分析
线程池解决两个问题:线程创建销毁开销大,以及无节制开线程会耗尽系统资源。ThreadPoolExecutor 的思路是让固定的一批 worker 线程循环消费同一个阻塞队列,把”提交任务”和”执行任务”解耦;整个池子的生命周期状态和线程数被压进一个 AtomicInteger(ctl),靠位运算拆装。
代码块收起展开
// 基于 JDK 25 (本地 JAVA_Source 仓), java.util.concurrent.ThreadPoolExecutor
public class ThreadPoolExecutor extends AbstractExecutorService {
// 高 3 位存运行状态，低 29 位存线程数。合一的目的：一次 CAS 同时校验并修改两者，杜绝"状态刚查完就变了"的窗口
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; // 29
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1; // 低 29 位掩码，线程数上限约 5.36 亿
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS; // 唯一负数状态：isRunning 退化成一次 c < SHUTDOWN 比较
private static final int SHUTDOWN = 0 << COUNT_BITS; // 不收新任务，但把队列存量跑完（shutdown()）
private static final int STOP = 1 << COUNT_BITS; // 不收新任务，丢弃队列，中断进行中任务（shutdownNow()）
private static final int TIDYING = 2 << COUNT_BITS; // 线程数归零，正在执行 terminated() 钩子
private static final int TERMINATED = 3 << COUNT_BITS; // 彻底结束，awaitTermination 返回
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~COUNT_MASK; }
private static int workerCountOf(int c) { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
private static boolean runStateAtLeast(int c, int s) {
return c >= s; // 状态值刻意排成单调递增，比大小就是比生命周期先后，不用拆位
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
private final BlockingQueue<Runnable> workQueue; // 任务缓冲区。判空必须用 isEmpty()：DelayQueue 的 poll() 返回 null 不代表队列空
private final ReentrantLock mainLock = new ReentrantLock(); // 守护 workers 集合，并串行化中断动作，避免 shutdown 时的中断风暴
private final HashSet<Worker> workers = new HashSet<>(); // 只在持有 mainLock 时访问，所以敢用非线程安全的 HashSet
private final Condition termination = mainLock.newCondition(); // awaitTermination 在这上面等
private volatile ThreadFactory threadFactory;
private volatile RejectedExecutionHandler handler; // 饱和或已关闭时的拒绝策略
private volatile long keepAliveTime; // 空闲线程等任务的超时，构造时已转纳秒
private volatile boolean allowCoreThreadTimeOut; // true 时核心线程也按 keepAliveTime 回收
private volatile int corePoolSize; // volatile 而非加锁：这些参数允许运行时热调，偶尔读到旧值不破坏任何不变式
private volatile int maximumPoolSize;
public void execute(Runnable command) {
Objects.requireNonNull(command, "command");
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) { // 第一步：核心额度没用满就直接开新线程，哪怕现有线程正闲着
if (addWorker(command, true)) // true = 拿 corePoolSize 当上限
return;
c = ctl.get(); // 失败说明并发下额度被抢或池被关，重读 ctl 走后续分支
}
if (isRunning(c) && workQueue.offer(command)) { // 第二步：核心满员，先排队，队列满才继续加线程
int recheck = ctl.get(); // offer 和状态检查两步之间池可能被关，必须复查
if (! isRunning(recheck) && remove(command)) // 入队后被 shutdown：把任务捞出来走拒绝，别让它烂在队列里
reject(command);
else if (workerCountOf(recheck) == 0) // 入队后线程恰好死光（比如 core=0 的池）：补一个空线程去消费
addWorker(null, false);
}
else if (!addWorker(command, false)) // 第三步：队列塞不进，往 maximumPoolSize 冲；还不行只能拒绝
reject(command);
}
}Worker 是理解中断语义的钥匙:它本身就是一把不可重入的 AQS 锁,“持有锁”等价于”正在执行任务”。
代码块收起展开
// 基于 JDK 25 (本地 JAVA_Source 仓), java.util.concurrent.ThreadPoolExecutor
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
final Thread thread; // 真正的线程由 ThreadFactory 造；工厂返回 null 时 addWorker 会整体回滚
Runnable firstTask; // 可为 null：null 表示这是个纯补位线程，直接去队列取活
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker state=-1 表示"还没跑起来"，此时 shutdownNow 也不许中断它
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this); // 线程主体委托回外部类
}
protected boolean isHeldExclusively() {
return getState() != 0; // 0 空闲，1 执行中
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) { // 只许 0->1，故意做成不可重入：任务代码若回调 setCorePoolSize 等
setExclusiveOwnerThread(Thread.currentThread()); // 池控制方法，不能凭重入拿到自己的执行锁再中断自己
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { // state>=0 才动手，呼应构造时的 -1
t.interrupt();
}
}
}
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// Check if queue empty only if necessary.
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null // SHUTDOWN 后拒绝带新任务的线程
|| workQueue.isEmpty())) // 但队列还有存量时，允许加空任务线程去清队列
return false;
for (;;) {
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK)) // core 只决定用哪条上限，线程本身没有核心/非核心身份
return false;
if (compareAndIncrementWorkerCount(c)) // 先 CAS 占名额再建线程，名额是唯一准入凭证
break retry;
c = ctl.get(); // Re-read ctl
if (runStateAtLeast(c, SHUTDOWN))
continue retry; // 状态变了回外层重判；只是计数变了就在内层重试
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int c = ctl.get(); // 拿到锁再重查一次状态：建 Worker 期间池可能被关
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
if (t.getState() != Thread.State.NEW) // 工厂私自 start 过线程，直接判非法
throw new IllegalThreadStateException();
workers.add(w);
workerAdded = true;
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
container.start(t); // 到这一刻线程才真正启动，进入 runWorker
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w); // 任何失败（含 Thread.start 抛 OOM）都要退回名额，否则 ctl 计数永久泄漏
}
return workerStarted;
}worker 的一生:主循环 runWorker、取任务兼裁员的 getTask、善后的 processWorkerExit,加上关闭入口 shutdown。
代码块收起展开
// 基于 JDK 25 (本地 JAVA_Source 仓), java.util.concurrent.ThreadPoolExecutor
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts 把构造时的 state=-1 清成 0，从此可被中断
boolean completedAbruptly = true; // 先假定会异常退出，正常走完循环才翻成 false
try {
while (task != null || (task = getTask()) != null) { // getTask 返回 null 就意味着该退出了
w.lock(); // 持锁 = 正在执行任务，shutdown() 的 interruptIdleWorkers 只中断没持锁的空闲线程
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) && // interrupted() 顺手清中断标记，再查一次状态防 shutdownNow 竞态
!wt.isInterrupted())
wt.interrupt(); // 两头保证：STOP 态任务必须带中断标记跑，非 STOP 态必须干净地跑
try {
beforeExecute(wt, task); // 钩子：可做监控、MDC 传递
try {
task.run(); // 业务异常不吞，抛给 UncaughtExceptionHandler，本线程随之退出
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex); // 异常路径也保证 afterExecute 能拿到异常
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly); // 正常退休和异常战死都从这里善后
}
}
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) { // STOP 立刻走人；SHUTDOWN 要把队列清空才走
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 裁员只看总数是否超编，谁碰上超时谁被裁
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) { // 队列非空时至少留 1 个线程，防止任务无人认领
if (compareAndDecrementWorkerCount(c)) // CAS 减数成功才准退出，输了说明别人先退了，重新评估
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; // 中断只当作"醒来重查状态"的信号，不算超时
}
}
}
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted 正常退出的已在 getTask 里减过计数
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks; // 死前把个人计数并入全局
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate(); // 自己可能是最后一个线程，走一次终止判定，推动池进入 TIDYING
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); // 被业务异常杀死的线程立刻补上：池的容量不因任务抛异常而缩水
}
}
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
advanceRunState(SHUTDOWN); // CAS 推状态，只进不退
interruptIdleWorkers(); // 只叫醒空闲线程，正在干活的不打扰
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) { // tryLock 成功说明该 worker 没在执行任务，才有资格被中断
try {
t.interrupt();
} finally {
w.unlock();
}
}
if (onlyOne)
break; // tryTerminate 走这个分支：每个退出的线程再叫醒一个，中断信号像接力一样传播
}
} finally {
mainLock.unlock();
}
}原理串讲
跟一个任务走完整条链路。调用 execute(task),先读一次 ctl:workerCountOf(c) < corePoolSize 成立就调 addWorker(task, true)。
代码块收起展开
addWorker 里是两段式的:第一段在双层循环里用 `compareAndIncrementWorkerCount` 对 ctl 做 CAS,把线程数加一;第二段才 `new Worker(task)`、持 `mainLock` 加进 `workers` 集合、`container.start(t)` 启动线程。
为什么先 CAS 名额再建线程,而反过来做不行?因为 Thread 创建和启动很贵还可能失败(工厂返回 null、start 抛 OOM),如果先建线程再登记数量,并发提交时会瞬间超编;先抢到名额的一定不超编,失败了由 `addWorkerFailed` 把名额退回去,计数永远守恒。线程启动后进入 runWorker:先 w.unlock() 把构造时的 state=-1 清零(在此之前 interruptIfStarted 看到负数会拒绝中断,保证线程没跑起来就不会挨刀),然后循环 task.run() 加 getTask()。
核心线程数满了以后,后续 execute 走 workQueue.offer,任务躺进队列,由空闲 worker 的 getTask 里那句 workQueue.take() 阻塞取出。
队列也满了 offer 返回 false,才 addWorker(task, false) 用 maximumPoolSize 做上限扩容,再失败就 reject。
为什么顺序是”先排队后扩容”,而非线程越多越快?作者的预期是队列本身就是削峰缓冲:只要缓冲还能兜住,就不值得为瞬时高峰付出建线程和之后回收的成本;真正持续的过载才升级到非核心线程,最后交给拒绝策略把压力显式暴露出来。
反过来这也解释了为什么无界队列(如默认的 LinkedBlockingQueue)会让 maximumPoolSize 形同虚设:offer 永远成功,第三步根本走不到。
回收路径全在 getTask:timed 为 true 时用 poll(keepAliveTime) 限时等,超时后下一轮循环走 compareAndDecrementWorkerCount 退出;为 false 时 take() 无限等。
所谓”核心线程”没有任何身份标记,只是当前线程数不超过 corePoolSize 时大家都用 take 罢了。
关闭时 shutdown() 推状态到 SHUTDOWN 并 interruptIdleWorkers(),用 w.tryLock() 区分闲忙:Worker 把”执行任务”建模成持有一把不可重入锁,tryLock 拿得到就是空闲线程,中断它只会打断 take/poll 的阻塞,绝不会打进用户任务内部。
这就是 Worker 要继承 AQS 自造锁而不用 ReentrantLock 的原因:可重入的话,任务代码回调池的控制方法时能重入拿到自己的锁,把自己误判成”空闲”。
被中断的线程在 getTask 顶部发现 SHUTDOWN 且队列空,减数返回 null,runWorker 退出循环,processWorkerExit 做完统计后调 tryTerminate,最后一个线程负责把状态 CAS 到 TIDYING、执行 terminated()、置 TERMINATED 并 termination.signalAll() 唤醒 awaitTermination 的等待者。
设计取舍
- ctl 合并状态与计数,换来单次 CAS 的原子性,代价是线程数上限从 2^32 缩到 2^29,工程上完全够用。
- 先排队后扩容偏向吞吐平稳;想要”先扩线程再排队”的激进策略,要靠自定义队列的 offer 逻辑(Tomcat 的做法)。
- Worker 用不可重入锁表达闲忙,让 shutdown 的中断天然绕开正在执行的任务;synchronized 或 ReentrantLock 都给不了这个语义。
- execute 里入队后的 recheck 是典型的”检查再检查”模式:offer 与状态判断非原子,宁可多查一次也不让任务滞留在已关闭的池里。
- 任务抛异常会杀死当前 worker 并由 processWorkerExit 补新线程,池子容量自愈,但频繁异常等于反复重建线程,代价可观。