结合美团技术篇详述Java线程池实现原理
一、写在前面
1.1 线程池是什么
线程池（Thread Pool） 是一种池化思想管理线程的工具，经常出现在多线程服务器中，如MySQL。
线程过多会带来额外的开销，其中包括创建销毁线程的开销，操作系统调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务是创建销毁线程开销代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对操作系统内核的充分利用。
本文描述的线程池是JDK提供的ThreadPoolExecutor类使用线程池带来的好处
- 降低资源消耗：通过赤化技术重复利用已创建的线程，降低想成创建和 销毁造成的消耗
- 提高响应速度：任务到达时，无需等待线程创建即可立即执行
- 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分配导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控
- 提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加风多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行
1.2 线程池解决的问题是什么
线程池解决的问题就是资源管理的问题。在并发环境下，系统不能够确定在任意时刻有多少任务需要执行，有多少资源需要投入。在这种不确定性下将会带来以下若干的问题
- 频繁申请/销毁资源和调度资源，将带来额外的开销，可能是非常巨大的
- 对资源无限申请缺少抑制手段，容易引发系统资源耗尽问题的风险
- 系统无法合理管理内部的资源分布，会减低系统的稳定性
为了解决资源分配的问题，线程池采用“池化”（Pooling）思想。池化，顾名思义，是为了做大化收益并最小化风险，而将资源统一在一起管理的一种思想。
在计算机领域池化技术表现为：统一管理IT资源，包括服务器资源、存储、网络资源等。通过共享资源，使用户在第投入中获益。
除去线程池其他比较典型的几种使用策略包括
- 内存池（Memory Pooling）：预先申请内存，提升申请内存的速度，减少内存碎片
- 连接池（Connection Pooling）：预先申请数据库连接，提升申请连接的速度，降低系统开销
- 实例池（Object Pooling）：循环使用对象，减资源在初始化和释放时昂贵的损耗
二、线程池和核心设计与实现
2.1 总体设计
Java中线程池核心实现类是ThreadPoolExecutor，本章基于JDK1.8的源码来分析Java线程池的核心设计与实现。首先看一下ThreadPoolExecutor的UML图，了解ThreadPoolExecutor的继承关系
ThreadPoolExecutor实现的顶层接口是Executor，顶层接口Executor提供了一种思想：将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，用户只需提供Runnable对象，将任务的运行逻辑提交到执行器Executor中，由Executor框架完成线程的调配和任务的执行部分。
ExecutorService
- 扩充执行任务的能力，补充可以为一个或者一批异步任务生成Future的方法
- 提供了管理线程池的方法，比如停止线程池的运行
AbstractExecutorService
- 串联任务流程，保证下层的实现只需要关注一个执行任务的方法
ThreadPoolExecutor
- 维护自身的生命周期
- 管理线程和任务，使两者良好的结合从而执行并行任务
ThreadPoolExecutor是如何运行，如何同时维护线程和执行任务的呢？其运行机制如下图所示
ThreadPoolExecutor运行流程
线程池在内部实际上构造了一个生产者消费者模型，将线程和任务两者解耦，并不直接关联，从而良好的管理缓冲任务，复用线程。线程池的运行主要分成两部分:：任务管理、线程管理。任务管理充当生产者角色，当任务提交后，线程池会判断该任务后续流转
- 任务申请线程执行该任务
- 缓冲到队列中等待线程执行
- 拒绝该任务
线程管理部分是消费者，它们被统一维护在线程池内，根据任务请求进行线程的分配，当线程执行完任务后会继续获取新的任务执行，最终获取不到任务的时候，线程会被回收。
接下来按照如下三个方面讲解线程池的运行机制：
- 线程池如何维护自身状态
- 线程池如何管理任务
- 线程池如何管理线程
2.2 生命周期管理
线程池运行的状态，并不是用户显式设置的，而是伴随着线程池的运行，由内部来维 护。线程池内部使用一个变量维护两个值：运行状态 (runState) 和线程数量 (workerCount)。在具体实现中，线程池将运行状态 (runState)、线程数量 (workerCount)
两个关键参数的维护放在了一起，如下代码所示：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));ctl 这个 AtomicInteger 类型，是对线程池的运行状态和线程池中有效线程的数量 进行控制的一个字段，它同时包含两部分的信息：线程池的运行状态 (runState) 和 线程池内有效线程的数量 (workerCount)，高 3 位保存 runState，低 29 位保存 workerCount，两个变量之间互不干扰。用一个变量去存储两个值，可避免在做相关 决策时，出现不一致的情况，不必为了维护两者的一致，而占用锁资源。通过阅读线 程池源代码也可以发现，经常出现要同时判断线程池运行状态和线程数量的情况。线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数。这里都使用的是位运算的方式，相比于基本运算，速度也会快很多。
关于内部封装的获取生命周期状态、获取线程池线程数量的计算方法如以下代码 所示：
// Packing and unpacking ctl
// 计算当前运行状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 计算当前线程数据
private static int workerCountOf(int c) { return c & CAPACITY; }
// 通过状态和线程数生成ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }ThreadPoolExecutor 的运行状态有 5 种，分别为：
// 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;| 运行状态 |
状态描述 |
| RUNNING |
能接受新提交的任务，并且也能处理阻塞队列中的任务 |
| SHUTDOWN |
状态关闭，不在接受新提交的任务，但是能继续处理阻塞队列已保存的让任务 |
| STOP |
不接受新任务，也不处理队列中的任务，会中断正在处理任务的线程 |
| TIDYING |
所有让任务都已终止，workerCount（有效处理让任务线程）状态为0 |
| TERMINATED |
在terminated()方法执行结束后进入该状态 |
其生命周期转换如下图所示
线程池生命周期
2.3 任务调度机制
2.3.1 任务调度
任务调度是线程池的主要入口，当用户提交了一个任务，接下来这个任务将如何执行 都是由这个阶段决定的。了解这部分就相当于了解了线程池的核心运行机制。 首先，所有任务的调度都是由 execute 方法完成的，这部分完成的工作是：检查现在线程池的运行状态、运行线程数、运行策略，决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。其执行过程如下：
- 首先检测线程池运行状态，如果不是 RUNNING，则直接拒绝，线程池要保证在 RUNNING 的状态下执行任务
- 如果 workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务
- 如果 workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
- 如 果 workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
- 如果 workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满 , 则根据拒绝策略来处理该任务 , 默认的处理方式是直接抛异常。
其执行流程如下
任务调度流程图
2.3.2 任务缓冲
任务缓冲模块是线程池能够管理任务的核心部分。线程池的本质是对任务和线程的管 理，而做到这一点最关键的思想就是将任务和线程两者解耦，不让两者直接关联，才 可以做后续的分配工作。线程池中是以生产者消费者模式，通过一个阻塞队列来实现 的。阻塞队列缓存任务，工作线程从阻塞队列中获取任务。
阻塞队列 (BlockingQueue) 是一个支持两个附加操作的队列。这两个附加的操作是： 在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程 会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元 素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。
阻塞队列
使用不同的队列可以实现不一样的任务存取策略。在这里，我们可以再介绍下阻塞队列的成员：
| 名称 |
描述 |
| ArrayBlockingQueue |
一个用数组实现的有界阻塞队列，此队列按照先进先出（FIFO）的原则对元素进行排序。支持公平锁和非公平锁 |
| LinkedBlockingDeque |
一个由链表结构组成的有界队列，此队列按照先进先出（FIFO）的原则对元素进行排序。此队列的默认长度为Integer.MAX_VALUE，所以默认创建此队列有容量危险 |
| PriorityBlockingQueue |
一个支持线程优先级排序的无界队列，默认自然进行排序，也可以自定义实现compareTo()方法指定排序故障，不能保证同优先级元素的顺序。 |
| DelayQueue |
一个实现PriorityBlockingQueue实现延迟获取的无界队列，在创建元素时，可以指定多久才能从队列中获取当前元素。只有延迟期满后才能从队列中获取元素。 |
| SynchronousQueue |
一个不存储元素的阻塞队列，每个put操作必须等待take操作，否则不能添加元素。支持公平锁和非公平锁。SynchronousQueue的一个使用场景是在线程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue，这个线程池根据需要（新任务来）创建新的线程，如果有空闲的线程就使用空闲线程，线程空闲60秒会被回收。 |
| LinkedTransferQueue |
一个由链表结构组成的无界阻塞队列，相当于其他队列，LinkedTransferQueue多了transfer和tryTransfer方法 |
| LinkedBlockingQueue |
一个由链表结构组成的双向阻塞队列，队列的头部和尾部都可以插入和删除元素，多线程并发时，可以将锁的竞争最多降到一半 |
2.3.3 任务申请
由上文的任务分配部分可知，任务的执行有两种可能：一种是任务直接由新创建的线 程执行。另一种是线程从任务队列中获取任务然后执行，执行完任务的空闲线程会再 次去从队列中申请任务再去执行。第一种情况仅出现在线程初始创建的时候，第二种 是线程获取任务绝大多数的情况。
线程需要从任务缓存模块中不断地取任务执行，帮助线程从阻塞队列中获取任务，实现线程管理模块和任务管理模块之间的通信。这部分策略由 getTask 方法实现，其 执行流程如下图所示：
线程获取任务的流程
getTask 这部分进行了多次判断，为的是控制线程的数量，使其符合线程池的状 态。如果线程池现在不应该持有那么多线程，则会返回 null 值。工作线程 Worker 会不断接收新任务去执行，而当工作线程 Worker 接收不到任务的时候，就会开始 被回收。
源码分析
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 判断线程池是否已停止运行
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;
}
}
}2.3.4 任务拒绝
任务拒绝模块是线程池的保护部分，线程池有一个最大的容量，当线程池的任务缓存 队列已满，并且线程池中的线程数目达到 maximumPoolSize 时，就需要拒绝掉该任务，采取任务拒绝策略，保护线程池。
拒绝策略是一个接口，其设计如下：
public interface RejectedExecutionHandler {
/**
* Method that may be invoked by a {@link ThreadPoolExecutor} when
* {@link ThreadPoolExecutor#execute execute} cannot accept a
* task. This may occur when no more threads or queue slots are
* available because their bounds would be exceeded, or upon
* shutdown of the Executor.
*
* <p>In the absence of other alternatives, the method may throw
* an unchecked {@link RejectedExecutionException}, which will be
* propagated to the caller of {@code execute}.
*
* @param r the runnable task requested to be executed
* @param executor the executor attempting to execute this task
* @throws RejectedExecutionException if there is no remedy
*/
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}用户可以通过实现这个接口去定制拒绝策略，也可以选择 JDK 提供的四种已有拒绝策略，其特点如下
| 名称 |
描述 |
| ThreadPoolExecutor.AbortPolicy |
丢弃任务并抛出RejectedExecutionException异常。这是线程池默认的拒绝策略，在任务不能在提交的时候，抛出异常，及时反馈程序运行状态。如果是比较关键的业务，推荐使用该策略，这样子在系统不能承载更大并发的时候，能过及时的通过异常发现。 |
| ThreadPoolExecutor.DiscardPolicy |
丢弃任务，但是不抛出异常。使用该策略，可能会使我们无法发现系统的异常状态。建议一些无关紧要的业务采用此策略。 |
| ThreadPoolExecutor.DiscardOldestPolicy |
丢弃队列最前面的任务，然后重新提交比拒接的任务。是否要采用此种策略，需要根据实际业务是否允许丢弃老任务来认真衡量 |
| ThreadPoolExecutor.CallerRunsPolicy |
由调用线程（提交任务的线程）来处理任务。这种情况是需要让所有的任务都执行完毕，那么就适合大量计算的任务类型去执行，多线程仅仅是增加大吞吐量的手段，最终必须要让每个任务都执行 |
/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
/**
* A handler for rejected tasks that throws a
* {@code RejectedExecutionException}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}2.4 Worker线程管理
2.4.1 Worker线程
线程池为了掌握线程的状态并维护线程的生命周期，设计了线程池内的工作线程Worker。
Java Worker源码部分
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/** Thread this worker is running in. Null if factory fails. */
// worker持有的线程
final Thread thread;
/** Initial task to run. Possibly null. */
// 初始化的任务，可以为null
Runnable firstTask;
...
}Worker 这个工作线程，实现了 Runnable 接口，并持有一个线程 thread，一个初始化的任务 firstTask。thread 是在调用构造方法时通过 ThreadFactory 来创建的线程，可以用来执行任务；firstTask 用它来保存传入的第一个任务，这个任务可以有也可以为 null。如果这个值是非空的，那么线程就会在启动初期立即执行这个任务，也就对应核心线程创建时的情况；如果这个值是 null，那么就需要创建一个线程去执行任务列表（workQueue）中的任务，也就是非核心线程的创建
/**
* The queue used for holding tasks and handing off to worker
* threads. We do not require that workQueue.poll() returning
* null necessarily means that workQueue.isEmpty(), so rely
* solely on isEmpty to see if the queue is empty (which we must
* do for example when deciding whether to transition from
* SHUTDOWN to TIDYING). This accommodates special-purpose
* queues such as DelayQueues for which poll() is allowed to
* return null even if it may later return non-null when delays
* expire.
*/
# workerQueue 源码定义
private final BlockingQueue<Runnable> workQueue;worker执行任务
线程池需要管理线程的生命周期，需要在线程长时间不运行的时候进行回收。线程池 使用一张 Hash 表去持有线程的引用，这样可以通过添加引用、移除引用这样的操作 来控制线程的生命周期。这个时候重要的就是如何判断线程是否在运行。
/**
* Set containing all worker threads in pool. Accessed only when
* holding mainLock.
*/
private final HashSet<Worker> workers = new HashSet<Worker>();Worker 是通过继承 AQS，使用 AQS 来实现独占锁这个功能。没有使用可重入锁ReentrantLock，而是使用 AQS，为的就是实现不可重入的特性去反应线程现在的执行状态。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable- lock方法一旦获取了独占锁，表示当前线程正在执行任务中
- 如果正在执行任务，则不应该中断线程
- 如果该线程现在不是独占锁状态，也就是空闲状态，说明它没有正在处理任务，这时可以对该线程进行中断
- 线程池在执行shutdown方法或tryTeriminate方法是或调用interruptIdleWorkers方法来中断空闲线程，interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态，如果是空闲状态则可以安全回收
shutdown方法源码
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
// 执行interruptIdleWorkers方法
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}tryTerminate方法源码
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
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
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
}
}interruptIdleWorkers方法源码
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()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}在线程回收过程中就使用到了这种特性，回收过程如下图所示：
线程池回收过程
2.4.2 worker线程增加
增加线程是通过线程池中的 addWorker 方法，该方法的功能就是增加一个线程， 该方法不考虑线程池是在哪个阶段增加的该线程，这个分配线程的策略是在上个步 骤完成的，该步骤仅仅完成增加线程，并使它运行，最后返回是否成功这个结果。 addWorker 方法有两个参数：firstTask、core。firstTask 参数用于指定新增的线程执行的第一个任务，该参数可以为空；core 参数为 true 表示在新增线程时会判断当前活动线程数是否少于 corePoolSize，false 表示新增线程前需要判断当前活动线程数是否少于 maximumPoolSize，其执行流程如下图所示：
申请线程执行流程图
源码分析
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 判断线程是否已经停止
// 判断线程是否正在停止 如果是则判断线程是否用于执行剩余任务firstTask
// workQueue是否为空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 获取线程数量
int wc = workerCountOf(c);
// 判断线程是否超过容量
// 判断线程是否超过对应核心数 上面讲了core 传true/false区别
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()) // precheck that t is startable
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;
}2.4.3 worker线程回收
线程池中线程的销毁依赖 JVM 自动的回收，线程池做的工作是根据当前线程池的状态维护一定数量的线程引用，防止这部分线程被 JVM 回收，当线程池决定哪些线 程需要回收时，只需要将其引用消除即可。Worker 被创建出来后，就会不断地进行轮询，然后获取任务去执行，核心线程可以无限等待获取任务，非核心线程要限时获取任务。当 Worker 无法获取到任务，也就是获取的任务为空时，循环会结束，Worker 会主动消除自身在线程池内的引用。
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) {
// 执行任务
} finally {
// 获取不到任务，主动回收自己
processWorkerExit(w, completedAbruptly);
}
}
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);
} finally {
mainLock.unlock();
}
...
}事实上，在这个方法中，将线程引用移出线程池就已经结束了线程销毁的部分。但由于引起线程销毁的可能性有很多，线程池还要判断是什么引发了这次销毁，是否要改变线程池的现阶段状态，是否要根据新状态，重新分配线程。
线程销毁流程
2.4.4 worker线程执行任务
在 Worker 类中的 run 方法调用了 runWorker 方法来执行任务，runWorker 方法的执行过程如下：
- while循环不断获取getTask()方法获取任务
- getTask()方法从阻塞队列获取任务
- 如果线程池正在停止，那么保证当前线程是中断状态，否则要保证当前线程不是中断状态
- 执行任务
- 如果getTask结果为null则调出循环，执行processWorkerExit()，销毁线程
执行任务流程
2.4.5 worker如何保证核心线程不被回收
源码分析
我们通常都是通过执行execute(Runnable command)方法来向线程池提交一个不需要返回结果的任务的如果你需要返回结果那么就是 <T> Future<T> submit(Callable<T> task)方法）
- 第一步：execute方法分析
public void execute(Runnable command) {
// 提交任务为null 抛出异常
if (command == null)
throw new NullPointerException();
// 获取线程池状态\线程池线程数据
int c = ctl.get();
// 小于核心线程数 addWorker()
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 大于核心线程数，当前线程池是运行状态，向阻塞队列中添加任务
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 队列添加失败 拒绝策略处理
else if (!addWorker(command, false))
reject(command);
}- 第二步：addWorker()方法分析
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 死循环
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果当前线程状态是SHUTDOWN STOP TIDYING TERMINATED 并且SHUTDOWN状态时任务队列为空 返回false
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 死循环
for (;;) {
int wc = workerCountOf(c);
// core参数 true corePoolSize核心线程数 false maximumPoolSize最大线程数
// CAPACITY integer最大值 (1 << COUNT_BITS) - 1;
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()) // precheck that t is startable
throw new IllegalThreadStateException();
// 向正在被执行的任务队列workers中添加worker
// 注意区分
// HashSet<Worker> workers = new HashSet<Worker>() 线程池中线程
// private final BlockingQueue<Runnable> workQueue 等待被执行的任务
workers.add(w);
int s = workers.size();
// 记录任务最大数
if (s > largestPoolSize)
largestPoolSize = s;
// 添加任务成功
workerAdded = true;
}
} finally {
// 释放锁
mainLock.unlock();
}
// 添加任务成功，那么开始执行任务
if (workerAdded) {
// 重点代码 -- 我们需要查看worker中的run()
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}- 第三步：查看worker中的run()
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}- 第四步：查看runWorker()
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 获取worker对象中的任务 可以为null
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 死循环
// 判断任务是否为空，如果为空则getTask()获取任务
while (task != null || (task = getTask()) != null) {
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
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();
} 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 则会从getTask()重新获取
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 回收worker
processWorkerExit(w, completedAbruptly);
}
}我们可以看到beforeExecute(Thread t, Runnable r)方法和afterExecute(Runnable r, Throwable t)会在任务的执行前后执行，我们可以通过继承线程池的方式来重写这两个方法，这样就能够对任务的执行进行监控啦。
- 第五步：查看getTask()
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
// 死循环
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 判断线程池状态
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
// 统计worker
int wc = workerCountOf(c);
// 如果设置了allowCoreThreadTimeOut(true) 或者当前运行的统计worker数大于设置的核心线程数，那么timed =true
// 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 {
// 看完这里就明白了
// 阻塞队列获取
// workQueue.poll() 规定时间获取任务
// workQueue.take() 会一直等待，知道阻塞队列中任务不为空
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
// 获取任务返回
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}参考资料
- JDK 1.8 源码
- 美团2020后台技术篇
- 深入理解 Java 线程池：ThreadPoolExecutor
- 点赞
- 收藏
- 关注作者
评论（0）