# Java线程池执行与线程复用的原理
# Preview
本文内容来源：
- 《深入浅出 Java 多线程》中 12 线程池原理 (opens new window)
- OpenJDK-16.0.2 源码
在看书的过程中觉得有很多地方讲的不够清晰，对于很多地方也是一笔带过，看源码的过程中产生了很多疑问，解决疑问的过程也花了些时间，所以将这些疑问解决后补充进笔记中
# 构造方法及其参数
ThreadPoolExecutor提供的构造方法：
// 七个参数的构造方法
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
七个参数的意义如下：
int corePoolSize
核心线程的数量
线程池中有两类线程，核心线程和非核心线程
核心线程默认情况下会一直存在于线程池中，即使这个核心线程什么都不干（铁饭碗），而非核心线程如果长时间的闲置，就会被销毁（临时工）
创建线程时并不会记录每个线程具体是核心线程还是非核心线程，核心与否主要是通过线程池中的工作线程workCount与corePoolSize的值来衡量，workCount > corePoolSize就代表线程池中存在非核心线程
int maximumPoolSize
最大线程数，核心线程+非核心线程的总数不能超过这个数
long keepAliveTime
非核心线程数在闲置时的存活时间，如果一个非核心线程的闲置时间超过
keepAliveTime，这个线程就会被销毁如果设置
allowCoreThreadTimeOut(true)，这个数会也作用于核心线程TimeUnit unit
keepAliveTime的单位，是一个枚举值NANOSECONDS： 纳秒MICROSECONDS： 微秒，1微秒 = 1000 纳秒MILLISECONDS：毫秒， 1毫秒 = 1000 微秒SECONDS： 秒，1秒 = 1000 毫秒MINUTES： 分HOURS： 小时DAYS： 天
BlockingQueue<Runnable> workQueue
阻塞队列，维护着等待执行的
Runnable任务对象，常用的阻塞队列如下：LinkedBlockingQueue链式阻塞队列，底层数据结构是链表，默认大小是
Integer.MAX_VALUE，可以指定大小ArrayBlockingQueue数组阻塞队列，底层数据结构是数组，需要指定队列的大小。
SynchronousQueue同步队列，内部容量为0，每个
put操作必须等待一个take操作，反之亦然DelayQueue延迟队列，该队列中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素
ThreadFactory threadFactory（可选）
创建线程的工厂 ，用于批量创建线程，统一在创建线程时设置一些参数，比如是否守护线程、线程的优先级，如果不指定，会新建一个默认的线程工厂
static class DefaultThreadFactory implements ThreadFactory { ... DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } ... }RejectedExecutionHandler handler（可选）
拒绝策略，线程数量大于最大线程数就会采用拒绝处理策略，四种拒绝处理的策略为 ：
ThreadPoolExecutor.AbortPolicy：默认拒绝处理策略，丢弃任务并抛出RejectedExecutionException异常。ThreadPoolExecutor.DiscardPolicy：丢弃新来的任务，但是不抛出异常ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列头部（最旧的）的任务，然后重新尝试执行程序（如果再次失败，重复此过程）ThreadPoolExecutor.CallerRunsPolicy：由调用线程（也就是往线程池中添加任务的线程）处理该任务
# 线程池的状态
线程池本身有一个调度线程，这个线程就是用于管理布控整个线程池里的各种任务和事务，比如创建线程、销毁线程、任务队列管理、线程队列管理
因此，线程池有自己的状态，这个状态记录在AtomicInteger ctl中
// 原子类对象ctl用来记录状态，分成两部分使用
// 32位Integer，高3位是runState（包括补码），低29位为workCount
// runState，运行状态，有五种，在下面有讲
// workCount，有效线程数，已经启动而且还没有停止的线程
// The workerCount is the number of workers that have been permitted to start and not permitted to stop
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;// 高3位是0，低29位是1
// runState is stored in the high-order bits
// 移到最高的3位里，后29位全是0
private static final int RUNNING = -1 << COUNT_BITS;// 111
private static final int SHUTDOWN = 0 << COUNT_BITS;// 000
private static final int STOP = 1 << COUNT_BITS;// 001
private static final int TIDYING = 2 << COUNT_BITS;// 010
private static final int TERMINATED = 3 << COUNT_BITS;// 011
// 获取c的高3位，也就是获取runState，线程池的运行状态
private static int runStateOf(int c) { return c & ~COUNT_MASK; }
// 获取c的低29位,也就是获取workCount，线程池的有效线程数
private static int workerCountOf(int c) { return c & COUNT_MASK; }
// 把runSate和workCount合并
private static int ctlOf(int rs, int wc) { return rs | wc; }
runState表示线程池的状态 ，分别为RUNNING、SHURDOWN、STOP、TIDYING 、TERMINATED
- 线程池创建后处于
RUNNING状态 - 调用
shutdown()方法后处于SHUTDOWN状态，线程池不能接受新的任务，清除一些空闲worker，会等待阻塞队列的任务完成 - 调用
shutdownNow()方法后处于STOP状态，线程池不能接受新的任务，中断所有线程，阻塞队列中没有被执行的任务全部丢弃。此时，poolsize为0，阻塞队列的size也为0 - 当所有的任务已终止，
ctl中的有效线程数workCount为0，线程池会变为TIDYING状态，接着会执行terminated()函数 - 线程池处在
TIDYING状态时，执行完terminated()方法之后，就会由TIDYING转向TERMINATED， 线程池被设置为TERMINATED状态
# 线程池执行任务的流程 - execute
在execute中，需要创建线程时，调用addWorker()完成，addWorker()具体如何创建线程，是下一节的内容
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 获取线程池状态ctl
int c = ctl.get();
// 1. 如果任务数量workCount小于核心线程数，创建核心线程，线程的任务为command
if (workerCountOf(c) < corePoolSize) {
// 如果添加成功，直接结束，addWorker的第二个参数代表是否添加核心线程
if (addWorker(command, true))
return;
// 如果已达核心线程数、线程池状态异常或者其他原因，添加失败，再获取一次线程池状态
c = ctl.get();
}
// 2. 如果线程池状态为RUNNING，尝试往工作队列里放入任务
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 如果任务成功排队，重新检查线程池状态
// 在上一次判断isRunning(c)以后，线程池状态可能发生变化，也可能有线程死亡
// 重新检查一次线程池状态，如果线程池不再是RUNNING，回滚排队，从工作队列中移除任务，并执行抛弃策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 重新检查一次有效线程数，如果原来的线程死亡导致线程数为0，新建一个非核心线程来执行工作队列的任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 没能成功往工作队列里放入任务，证明工作队列已满，尝试新建非核心线程来执行任务
else if (!addWorker(command, false))
// 新建非核心线程失败，证明线程数已经到了maximumPoolSize，执行抛弃策略
reject(command);
}
public boolean remove(Runnable task) {
boolean removed = workQueue.remove(task);
tryTerminate(); // In case SHUTDOWN and now empty
return removed;
}
private volatile RejectedExecutionHandler handler;
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
execute的流程图如下：
# 线程复用的原理
在execute的流程中，创建线程的部分，是使用addWorker()方法来完成的
也就是说，ThreadPoolExecutor在创建线程时，会将线程封装成工作线程worker，并放入工作线程组HashSet<Worker> workers>中
创建Worker时，会传入firstTask作为线程的第一个任务，还会新建一个线程，由Worker维护，这个线程就是实际上执行任务的线程
初次执行任务时，worker会执行firstTask，后面复用线程的时候，worker就会反复从阻塞队列中拿任务去执行
# 回顾一下线程池的几个状态
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;
private static final int COUNT_BITS = Integer.SIZE - 3; // 29
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;// 高3位是0，低29位是1
# 创建worker线程
// 创建worker线程
// firstTask：线程第一次执行的任务，线程执行完firstTask以后会从等待队列中拿任务执行
// core：是否为核心线程，作为workCount判断的条件，并不会真的把线程标记为核心与非核心
private boolean addWorker(Runnable firstTask, boolean core) {
// retry部分主要是 一些参数判断 和 增加workCount
retry:
for (int c = ctl.get();;) {
// 必要时检查队列是否为空
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
// 自旋增加线程数
for (;;) {
// 如果是创建核心线程，判断工作线程数是否大于corePoolSize
// 如果是创建非核心线程，判断工作线程数是否大于maximumPoolSize
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
// CAS增加工作线程数，成功就退出retry，没成功会一直自旋
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// CAS失败时，如果runState不是RUNNING，线程池要关闭，也退出retry
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
}
}
// 下面这部分，真正开始创建worker线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 创建一个Worker, 它的构造方法中会创建一个线程，worker是包装线程的一个壳
w = new Worker(firstTask);
// 这个Thread的Runnable任务就是Worker自己，Worker实现了Runnable
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.
// 在拿锁的时候可能ctl发生变化，重新检查
int c = ctl.get();
// 如果线程池在拿到锁之前变更状态，或者线程工厂出错，if判断不会成立
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
if (t.getState() != Thread.State.NEW)
throw new IllegalThreadStateException();
// 把worker放入HashSet<Worker> workers里
workers.add(w);
workerAdded = true;
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
}
} finally {
// 解锁
mainLock.unlock();
}
if (workerAdded) {
// 启动线程，执行任务
// 因为创建线程的时候，Runnable任务就是Worker，所以JVM会调用Worker的run方法
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
看完上面这段代码以后，有两个关键点
- Worker类到底是什么玩意儿
- 调用Worker中线程的
start()方法以后，线程会做些什么？ - 上面这段代码跟线程复用有什么关系？说到底不还是启动一个线程吗？
带着疑问，继续往下看
# 内部类Worker
Worker类其实就是线程池的一个内部类，他继承了AQS，也实现了Runnable接口，并重写了run()方法，可以作为线程的一个任务
在上面，最最关键的地方，其实就是t.start()，启动线程后，JVM选择一个合适的时机，执行Worker.thread的任务，透过Worker的构造方法可以看到，Worker.thread的任务就是自己这个worker对象
因此，启动线程后，会执行worker对象的run()方法
// 线程池的内部类Worker，继承AQS,实现Runnable
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
// 内部维护了一个线程，而创建线程时的Runnable任务就是Worker自己
// 因此，调用thread.start()以后，JVM会在合适的时机执行run()方法
final Thread thread;
Runnable firstTask;
Worker(Runnable firstTask) {
// state设置为-1的原因稍后再讲
setState(-1);
this.firstTask = firstTask;
// Worker类本身实现了Runnable接口，自己可以作为一个任务
this.thread = getThreadFactory().newThread(this);
}
// 在run方法中调用runWorker
public void run() {
// 内部类Worker调用外部类ThreadPoolExecutor的runWorker方法
// runWorker是实现线程复用真真正正的核心
runWorker(this);
}
}
# 线程复用的核心：runwork()方法
final void runWorker(Worker w) {
// worker.thread.start()会让JVM在合适的时候，由thread调用run()，而run()调用runWorker()
// 所以，执行runWorker()时，当前线程就是Worker里面维护的thread
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // set state = 0，允许中断，为什么设置state就能允许中断？为什么现在才允许中断？在答疑环节有讲
boolean completedAbruptly = true;
try {
// 任务都通过当前线程执行
// 第一次的任务执行完以后，会继续while循环，通过getTask()从等待队列里拿任务
// 直到池关闭、maximumPoolSize改变等原因使getTask()返回null，就退出while循环
// 然后worker的runWorker方法执行完毕，线程死亡
while (task != null || (task = getTask()) != null) {
// CAS把state从0改为1
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 检查线程池状态，如果线程池处于中断状态，当前线程将中断（设置中断标志位为true）
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 默认实现留空 一个hook钩子方法
beforeExecute(wt, task);
try {
// 直接调用run()方法执行任务
task.run();
// 默认实现留空 一个hook钩子方法
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock(); // set state = 0
}
}
completedAbruptly = false;
} finally {
// 完成worker死亡之前的工作，比如从线程池的HashSet<Worker> workers中删除worker
processWorkerExit(w, completedAbruptly);
}
}
看到这里应该明白线程复用是如何实现的了，线程会一直执行while循环，每次循环都会执行一次任务，第一次执行的任务是firstTask，也就是创建Worker时传入的任务，后面执行的任务是从等待队列里面提取的
一直持续到线程池关闭、maximumPoolSize改变等原因使getTask()返回null，就会退出while循环，然后worker的runWorker方法执行完毕，线程死亡
# 从等待队列中提取任务：getTask()方法
// 从等待队列中提取任务
private Runnable getTask() {
// 上次poll是否超时
// 如果超时，证明在指定时间内线程都没有办法拿到任务来执行
// 换句话说，在指定时间内，线程都是空闲的，因为拿不到任务执行
// 可以结合方法底部try-catch的代码理解
boolean timedOut = false;
// 自旋
for (;;) {
int c = ctl.get();
// 如果线程池状态为STOP，或者线程池状态为SHUTDOWN且等待队列为空
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
// 返回null，runWorker会退出while循环
// 当前线程也好，持有当前线程的worker也罢，都会死亡
// 因此调用ctl.addAndGet(-1)提前将工作线程-1
decrementWorkerCount();
return null;
}
// 获取workCount
int wc = workerCountOf(c);
// 如果允许核心线程超时（空闲超过指定时间，默认是非核心线程才淘汰，也就是超过corePoolSize时才淘汰线程）
// 或者workCount大于核心线程数（存在非核心线程，如果非核心线程空闲超过指定时间，会淘汰）
// 则可能有workers要被淘汰，需要计时
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 如果 workCount大于线程池最大容量 或者 要淘汰超时worker(需要计时&&超时)，证明需要淘汰worker
// 具体是否真的淘汰，还要结合线程池的整体状态来判断
// 如果线程池现在有worker(workCount>1)，可以淘汰多余的worker
// 如果队列为空，也可以淘汰多余的worker
// 但如果workCount=1，worker是线程池的独苗
// 而且队列不为空，队列还有任务等待执行，那就不淘汰，留着worker来执行任务
if ( (wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty()) ) {
// CAS使workCount-1，失败就自旋，成功就返回null，让worker死亡
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 获取任务
// 如果线程池中存在非核心线程，就会调用poll方法，如果队列为空会进入阻塞，等待指定时间
// 指定时间内一直拿不到就返回null，不会无期限地等待任务被放入队列中
// 如果线程池中都是核心线程，一般会调用take方法，如果队列为空会一直阻塞，避免浪费CPU时间
// 等到队列有任务放入，线程被唤醒，take成功拿到任务，take是一定能拿到任务的
// 不过，当allowCoreThreadTimeOut=true时，核心线程也会调用poll方法判断是否超时
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
// 能拿到任务，直接返回任务
if (r != null)
return r;
// 如果poll超时导致拿不到任务，设置超时标记，继续for循环自旋
// poll在指定时间内一直阻塞又拿不到任务
// 证明在指定时间内，线程都是空闲的，没有执行任务，因为拿不到任务执行
// 下一次自旋时，可能会使worker被淘汰，注意，只是可能
// 如果wc <= 1 && workQueue.isEmpty()，那么worker就不会被淘汰
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}// 自旋的for
}
# 疑问：为什么Worker的构造方法中要setState(-1)？runWork方法中的允许中断是什么？
设置state的原因：使用shutdownNow()关闭线程池时避免任务丢失
线程池使用shutdownNow()关闭时，会中断所有的worker，并把等待执行的任务拿出来作为List返回
但是，创建Worker时传入的任务，只存储在worker里面，并没有放在等待队列里
如果在创建worker之后、runWorker()执行任务之前，对worker进行中断，那么worker的firstTask就会丢失
为了避免这种丢失，在worker创建之后、runWorker()执行之前的这段时间里，不允许中断worker，而是让worker继续执行任务
返回等待执行的任务列表时，就不需要考虑worker的firstTask，直接返回等待队列的任务即可
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
final Thread thread;
Runnable firstTask;
// Worker还从AQS处继承了state，这个变量一开始是-1，此时不允许worker被中断
// 当runWorker()执行以后，变量置为0，此时worker允许被中断，此后state的值为0或者1
// 加锁后，state为1，解锁后，state为0
Worker(Runnable firstTask) {
// 设置state的原因：关闭线程池时避免任务丢失
// 线程池使用shutdownNow()关闭时，会中断所有的worker，并把等待执行的任务拿出来作为List返回
// 但是，创建Worker时传入的任务，只存储在worker里面，并没有放在等待队列里
// 如果在创建worker之后、runWorker()执行任务之前，对worker进行中断，那么worker的firstTask就会丢失
// 为了避免这种丢失，在worker创建之后、runWorker()执行之前的这段时间里
// 不允许中断worker，而是让worker继续执行任务
// 返回等待执行的任务列表时，就不需要考虑worker的firstTask，直接返回等待队列的任务即可
// state设置为-1的原因：
// 线程池使用shutdownNow()关闭时，会中断所有的worker，对于每一个worker
// 如果 state>=0 && thread不为空 && thread未中断，那么中断worker的线程
// 创建worker时把state设为-1，在runWorker()开始执行时把state设为0
// 那么在这段时间内worker就不会被中断
setState(-1);
this.firstTask = firstTask;
// Worker类本身实现了Runnable接口，自己可以作为一个任务
this.thread = getThreadFactory().newThread(this);
}
// 在run方法中调用runWorker
public void run() {
// 内部类Worker调用外部类ThreadPoolExecutor的runWorker方法
runWorker(this);
}
// Worker实现了AQS
public void lock() { acquire(1); }
public void unlock() { release(1); }
// 如果worker已经启动，中断worker
void interruptIfStarted() {
Thread t;
//
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
...
}
// 关闭线程池
public List<Runnable> shutdownNow() {
...
// 中断所有worker
interruptWorkers();
...
}
// 下面这两个都是线程池的方法
private void interruptWorkers() {
for (Worker w : workers)
w.interruptIfStarted();
}