一文让你领悟线程池的原理和机制设计—洞虚篇

书接上文，一文加深你对Java线程池的了解与使用—筑基篇，本文将从线程池内部的最最核心类 ThreadPoolExecutor 源码中的重要方法入手，也是本文分析的对象，从状态/任务/线程这三个模块剖析线程池的机制，掌握背后的核心设计。

一、线程池如何管理自身的状态/生命周期

在ThreadPoolExecutor 类中，有以下的定义：

//Integer的范围为[-2^31,2^31 -1], Integer.SIZE-3 =32-3= 29，用来辅助左移位运算
private static final int COUNT_BITS = Integer.SIZE - 3;
//(1 << 29) - 1=000011111111111111111111111111111,前三位是0，后29为1。
//常量值，被用以辅助与运算求出线程池运行状态or线程池线程数量，见runStateOf与workerCountOf方法
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
//线程池状态以常量值被存储在高位中(前三位)
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// 使用32位的变量ctl同时容纳两个重要概念的值，前3位存储线程池自身状态，后29位存储线程数量
//runStateOf负责取出状态值，workerCountOf负责取出线程数量值，ctlOf负责将两个值合成到一个32位的值
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

//自增型Integer,初始化时会将 RUNNING | 0 合成到一起
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

从以上源码可以看出：线程池的自身状态，共有5种，通过常量值方式定义下来，线程池被启动后，线程池状态存储在32位的自增型Integer变量

ctl

runStateOf(ctl)

那

ctl

workerCountOf(ctl)

通过

ctlOf(RUNNING, 0)

Integer

通过一个变量，巧妙地包含两部分的信息：线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，两者互不干扰，可避免在操作前做相关判断时为了维护两者的一致而占用锁和资源。

线程池状态含义

线程池如何切换状态

在官方给出的说明中，可以清晰看出线程池各个状态转变的触发条件：

RUNNING -> SHUTDOWN：On invocation of shutdown(), perhaps implicitly in finalize()
(RUNNING or SHUTDOWN) -> STOP: On invocation of shutdownNow()
SHUTDOWN -> TIDYING: When both queue and pool a
3ff8
re empty
STOP -> TIDYING:  When pool is empty
TIDYING -> TERMINATED: When the terminated() hook method has completed

线程池状态的生命周期

二、线程池如何管理任务

任务的调度机制

不论是哪一种类的线程池，调用execute往线程添加任务后，最后都会进入

ThreadPoolExecutor.execute(runnable)

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//获取工作线程数量 与核心线程数对比
if (workerCountOf(c) < corePoolSize) {
//传入true表示创建新线程时与核心线程数做比较，执行command
if (addWorker(command, true))
return;
c = ctl.get();
}
//来到此处，说明 工作线程数 已经大于 核心线程数
//短路原则，先判断线程池状态是否Running，处于Running则再判断阻塞队列是否可以存储新任务
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//短路原则，如果重检查线程池状态不在Running了，则尝试remove(command)阻塞队列移除此任务
if (! isRunning(recheck) && remove(command))
reject(command); //若队列成功移除任务后，就拒绝掉此任务
else if (workerCountOf(recheck) == 0)//运行状态，存储在阻塞队列，但工作线程目前为0
//null表示单纯创建工作线程，传入false表示创建新线程时与最大线程数做比较
addWorker(null, false);
}
//阻塞队列不可以存储任务，尝试增加工作线程执行command
else if (!addWorker(command, false))
reject(command); //若增加线程操作也失败了-->拒绝掉任务
}

在这短短的二十行代码里，出现了多个 if else，说明任务调度还是有点复杂的，下面来逐步理清它。

此处的

corePoolSize

this.corePoolSize = corePoolSize;

execute方法是外界往线程池添加任务的入口，也是线程池内部首先接触到外界任务的地方，它需要对任务的去向进行管理，对任务的管理有以下三个选项：

• 缓冲到队列中等待执行——

  workQueue.offer(command)

• 创建新的线程直接执行——

  addWorker(command, false)

• 拒绝掉该任务，执行线程池当前的拒绝策略——

  reject(command)

addWorker(Runnable firstTask, boolean core)

根据源码，可以总结下以下的判断条件，需要综合阻塞队列状态，当前工作线程数量，核心线程数，最大线程数这些线程池核心属性进行判断。

任务进队

在

execute()

从构造函数可以看出不同种类的阻塞队列都实现了

BlockingQueue<Runnable>

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
//···
}

因为BlockingQueue接口提供的操作可以自定义，所以也有了能适应各种不同场景的阻塞队列。不同的阻塞队列对 offer 方法的重写也是各不相同。

下面以

CachedThreadPool

SynchronousQueue

offer

public boolean offer(E e) {
if (e == null) throw new NullPointerException();
//transfer()传入的e若为null则作为消费者，若非null则作为生产者
//运用transferer.transfer达到put前需要take，不存储任务的目的
return transferer.transfer(e, true, 0) != null;
}

任务出队

任务的出队是任务管理模块与线程管理模块的联系，简单来说任务从阻塞队列中被取出说明有工作线程需要执行任务了。

任务被执行会有两种可能：第一种是任务直接由新创建的线程执行。另一种是线程从任务队列中获取任务然后执行，执行完任务的空闲线程会再次去从队列中取出任务再执行。

那么任务是什么时候又是怎样地会被取出呢？从

addWorker

private boolean addWorker(Runnable firstTask, boolean core) {
//... 省略

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask); //创建工作线程
final Thread t = w.thread; //获取工作线程持有的线程实例
if (t != null) {
//省略获取锁，加入阻塞队列，再次判断线程池运行状态部分的代码
if (workerAdded) {
t.start();
workerStarted = true;
}
}
...
return workerStarted;
}

接下来看下，这个持有线程实例的

Worker

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
/** 具体执行任务的线程实例， 若线程工厂提供不了可能为null */
final Thread thread;
/** 初始运行的任务.  可能为 null  */
Runnable firstTask;
/** 线程已完成任务计数器r */
volatile long completedTasks;
/**
* 初始化给定的firstTask和从线程工厂获取线程实例
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}

/** Delegates main run loop to outer runWorker  */
public void run() {
runWorker(this);
}
//省略以下关于锁和取消的方法
}

再来看下

runWorker(this);

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) {
//省略大量关于任务执行部分的代码
}
}
}

可以看出，

runWorker(worker)

下面再来看下这个从阻塞队列中返回任务的

getTask()

private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 检查队列是否为空，检查当前线程池的状态
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
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())) {
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;
}
}

由代码可以看到，里面的逻辑主要在死循环 for( ; 😉 ，任务返回null的条件有两个：第一个是线程池自身状态为

SHUTDOWN

STOP

TIDYING

TERMINATED

ctl

任务的拒绝策略

根据线程池当前设定的拒绝策略

RejectedExecutionHandler

可选的拒绝策略

JDK提供的四种已有拒绝策略，其特点如下：

自定义拒绝策略

拒绝策略是一个接口，其设计如下：

public interface RejectedExecutionHandler {
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

用户可以通过实现这个接口去定制拒绝策略，在手动配置线程池时的构造函数传入或者通过方法

setRejectedExecutionHandler

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
//省略其他代码
this.handler = handler;
}

public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
if (handler == null)
throw new NullPointerException();
this.handler = handler;
}

拒绝策略的执行

从调用

execute()

reject(command)

RejectedExecutionHandler

rejectedExecution(r,executor)

final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}

三、线程池如何管理工作线程

工作线程的增加

经历哪些判断，才能/才需要增加工作线程，通过什么方式增加

增加线程是通过线程池中的

addWorker(firstTask , core)

addWorker(firstTask , core)

firstTask

core

操作结果返回false的原因

情况一：不满足在线程池运行状态处于

SHUTDOWN

情况二：线程池运行状态处于

STOP

TIDYING

TERMINATED

情况三：当前活动线程数大于corePoolSize / 当前活动线程数大于maximumPoolSize

在将新线程

w

private boolean addWorker(Runnable firstTas
16c0
k, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 检查线程池状态和 阻塞队列
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();  // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}

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 {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());

if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 预检查线程是否能启动
throw new IllegalThreadStateException();
workers.add(w); //往阻塞队列中添加任务
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); //开启线程执行任务
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

工作线程的获取任务

工作线程获取任务的过程其实与阻塞队列的任务出队其实是相通的关系。这里的获取任务，换个容易理解的角度，就是让线程自身不处于空闲状态。

一个普普通通的Worker被创建的时候，它是分携带有 firstTask 和 不携带 firstTask两种情况的。如果这个Worker已经有了

firstTask

getTask()

一直获取任务失败会怎样

非核心线程要在一定时间内获取任务。因此某非核心线程在一定时间内无法获取到任务，空闲状态下的循环会结束，进入回收过程。

工作线程的执行任务

核心是

runWorker(worker)

final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取工作线程中用来执行任务的线程实例
w.firstTask = null;
w.unlock(); // 释放锁，允许线程被中断
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {//循环获取任务
w.lock(); //上锁
//如果池处于 STOP，请确保线程被中断;
//如果没有，请确保线程不被中断。
//在第二种情况下需要再次检查才能执行shutdownNow，然后执行中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() && runStateAtLeast(ctl.get(), STOP) ) )
&& ! wt.isInterrupted())
wt.interrupt();//中断当前线程
try {
beforeExecute(wt, task); //执行任务前回调操作
Throwable thrown = null;
try {
task.run(); //运行任务task
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown); //任务执行后回调操作
}
} finally {
task = null; //清除任务引用
w.completedTasks++; //工作线程任务执行数+1
w.unlock(); //释放锁
}
}
completedAbruptly = false; //任务执行完毕，将 未完成标志设为 false
} finally {
processWorkerExit(w, completedAbruptly);//退出获取任务循环后进入回收
}
}

引用大佬画的Worker执行任务流程图：

工作线程的回收

线程池中线程的销毁依赖JVM自动的回收，线 2958 程池做的工作是根据当前线程池的运行状态和线程池种类维护对应数量的线程引用，并防止这部分线程被JVM回收，当线程池只需要将待回收的线程引用消除则视为回收。

Worker被创建出来后，会被调用runWorker()，就会不断地进行轮询从阻塞队列中获取任务去执行，核心线程可以无限等待获取任务，非核心线程要限时获取任务。当Worker在一定时间内无法获取到任务，也就是获取的任务为空时，空闲状态下的循环会结束，Worker会执行processWorkerExit()，主动消除自身在线程池内的引用。

try {
while (task != null || (task = getTask()) != null) {
//省略执行任务代码
}
} finally {
processWorkerExit(w, completedAbruptly);//当一定时间内获取不到任务时，主动回收自身
}

跟进去看看

processWorkerExit

private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount(); //自减工作线程数

final ReentrantLock mainLock = this.mainLock;
mainLock.lock(); //可重入锁，上锁
try {
completedTaskCount += w.completedTasks; //记录增加该工作线程的已完成的任务数
workers.remove(w); //将阻塞队列中的w移除
} finally {
mainLock.unlock();
}

tryTerminate();//终止线程

int c = ctl.get();
if (runStateLessThan(c, STOP)) {//若运行状态在STOP之下
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; //工作线程数大于所需线程最小值，不需要取代
}
addWorker(null, false);//创建新的工作线程
}
}

由于线程池中线程的销毁依赖JVM自动地回收，当线程引用被移出线程池管辖范围时就已经结束了

再来看看这个

tryTerminate()

• 线程池状态为RUNNING
• 在TIDYING及之上
• SHUTDOWN且阻塞队列仍有任务
• 当前活跃工作线程数不为0

final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {//通过CAS自旋判断直到当前线程池运行状态为TIDYING以及活跃线程数要为0
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));//更新线程池状态和线程数
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}

四、总结

再次借用大佬的图，感谢感谢~

线程池机制设计的高度概括

线程池的设计分成三个管理模块：核心状态管理、任务管理、线程管理，管理操作都有对线程池运行状态、活跃线程数、阻塞队列这些线程池的核心状态来进行判断，在不同的核心状态下线程池会有不同的反应。

线程池通过维护一个32位的二进制变量，前3位的高位存储五种不同的线程池运行状态，后29位低位存储活跃线程数，为其他操作提供方便获取状态。

当外界通过execute()提交任务进线程池后，线程池会根据线程池的状态与线程数，判断该任务的命运：

• 创建新的工作线程执行该任务
• 缓冲到队列中，等待工作线程申请执行任务
• 拒绝执行任务，执行拒绝策略

线程池使用了一个内部类——工作线程，会维护指定数量的工作线程引用，任务进入线程池内部后线程池会根据核心属性对线程数量进行管理，当线程执行完获取到的任务后则会继续获取新的任务去执行直到队列已空。当某个工作线程在一定时间内都处于获取不到任务的空闲状态，线程池将它回收。

五、参考资料

• [1] JDK 1.8源码
• [2] Java线程池实现原理及其在美团业务中的实践