一文读懂线程池的实现原理

2023-04-17
广东
本文字数：7778 字
阅读完需：约 26 分钟

一、前言

上个月底群里的一个好朋友向老周提出啥时候分享 ThreadPoolExecutor 解析大全，我说后面会提上日程；然后前些天有读者也反馈说在面试中有被问到线程池，问我啥时候出一篇线程池相关的文章。今天老周就来安排一波线程池，现在很多公司都喜欢问线程池相关的面试题，为什么面试官这么热衷于问线程池相关的面试题呢？因为这是多线程的基础，ThreadPoolExecutor 的几个重要参数你必须会知道设置以及什么场景选择哪种 Executor 、线程池队列的选择以及相应的拒绝策略。

下面老周收集了几个朋友提供的大厂关于线程池的面试题：

线程池的使用场景
线程池各个参数的含义，你平时用的什么队列以及拒绝策略？
程序中哪些地方用到了线程池，用线程池的好处有哪些？
如何自己实现一个线程池
JDK 提供了哪些线程池的默认实现
阿里巴巴 Java 开发手册为啥不允许默认实现的线程池
线程池里的参数你是怎么得出来的，根据什么算出来的？
说说你自定义线程池里的工作流程
...

这里老周就不带大家一个个对面试题进行分析了，这里对只讲核心原理再结合动态调整线程池参数的实践来帮助你对线程池有个清晰的认识，知道了原理再结合自己的实践，那面试线程池也是得心应手了。那你有可能问，老周啊，我平时也没用到线程池啊，用的也都是定义类继承 Thread 类或者定义类实现 Runnable 接口来实现多线程的啊。额，如果你是面的 Java 中高级开发，那你千万不要这样说，这会让面试官一下觉得你不值中高级。如果你面的中高级还不知道线程池的话也没关系，幸好你看到了老周这篇文章，还不算晚；如果你是已经用过线程池相关，那这篇文章也会让你对线程池的原理更加清楚，在项目中应用也会得心应手。

二、线程池的概念

2.1 线程池是什么

线程池是一种线程使用模式。线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。

2.2 使用线程池的好处

降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
提高响应速度：任务到达时，无需等待线程创建即可立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池 ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。

2.3、ThreadPoolExecutor 的核心参数

网上说的天花乱坠的，也不如直接看 Doug Lea 大佬源码的注释来的更加贴切些。

corePoolSize：the number of threads to keep in the pool, even if they are idle, unless {@code allowCoreThreadTimeOut} is set 核心线程数：线程池中保留的线程数，即使它们是空闲的，除非设置 allowCoreThreadTimeOut。
maximumPoolSize：the maximum number of threads to allow in the pool 最大线程数：线程池中允许的最大线程数
keepAliveTime：when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.线程空闲时间：如果经过 keepAliveTime 时间后，超过核心线程数的线程还没有接受到新的任务，那就回收。
unit：the time unit for the {@code keepAliveTime} argument 单位：keepAliveTime 的时间单位
workQueue：the queue to use for holding tasks before they are executed. This queue will hold only the {@code Runnable} tasks submitted by the {@code execute} method.存放待执行任务的队列：当提交的任务数超过核心线程数后，再提交的任务就存放在这里。它仅仅用来存放被 execute 方法提交的 Runnable 任务。(这里不要再翻译成工作队列了好吗)
threadFactory：the factory to use when the executor creates a new thread 线程工厂：执行程序创建新线程时使用的工厂。比如我们项目中自定义的线程工厂，排查问题的时候，根据线程工厂的名称就知道这个线程来自哪里，很快的定位出问题，
handler ：the handler to use when execution is blocked because the thread bounds and queue capacities are reached 拒绝策略：当队列里面放满了任务、最大线程数的线程都在工作时，这时继续提交的任务线程池就处理不了，应该执行怎么样的拒绝策略。

三、线程池的实现原理

本文描述线程池是 JDK 8 中提供的 ThreadPoolExecutor 类，那我们就从 ThreadPoolExecutor 类来看下它的 UML 依赖关系。

3.1 总体设计

蓝色实线：继承关系
绿色虚线：接口实现关系
绿色实线：接口继承关系

ThreadPoolExecutor 实现的顶层接口是 Executor，顶层接口只提供了void execute(Runnable command); 这么一个方法，Executor 提供的是一种思想：将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，用户只需提供 Runnable 对象，将任务的运行逻辑提交到执行器(Executor)中，由 Executor 框架完成线程的调配和任务的执行部分。

ExecutorService 接口增加了一些能力：

扩充执行任务的能力，补充可以为一个或一批异步任务生成 Future 的方法；
提供了管控线程池的方法，比如停止线程池的运行。

AbstractExecutorService 则是上层的抽象类，将执行任务的流程串联了起来，保证下层的实现只需关注一个执行任务的方法即可。最下层的实现类 ThreadPoolExecutor 实现最复杂的运行部分，ThreadPoolExecutor 将会一方面维护自身的生命周期，另一方面同时管理线程和任务，使两者良好的结合从而执行并行任务。

我们来看下 ThreadPoolExecutor 的运行流程：

线程池在内部实际上构建了一个生产者消费者模型，将线程和任务两者解耦，并不直接关联，从而良好的缓冲任务，复用线程。线程池的运行主要分成两部分：任务管理、线程管理。

任务管理部分充当生产者的角色，当任务提交后，线程池会判断该任务后续的流转：

直接申请线程执行该任务
缓冲到队列中等待线程执行
拒绝该任务

线程管理部分充当消费者的角色，它们被统一维护在线程池内，根据任务请求进行线程的分配，当线程执行完任务后则会继续获取新的任务去执行，最终当线程获取不到任务的时候，线程就会被回收。

下面就从以下三个核心机制来详细讲解线程池运行机制：

线程池如何维护自身状态
线程池如何管理任务
线程池如何管理线程

3.2 线程池如何维护自身状态

线程池运行的状态，并不是用户显式设置的，而是伴随着线程池的运行，由内部来维护。线程池内部使用一个变量维护两个值：运行状态(runState)和线程数量(workerCount)。

ctl 这个 AtomicInteger 类型，是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段，它同时包含两部分的信息：线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，高 3 位保存 runState，低 29 位保存 workerCount，两个变量之间互不干扰。用一个变量去存储两个值，可避免在做相关决策时，出现不一致的情况，不必为了维护两者的一致，而占用锁资源。通过阅读线程池源代码也可以发现，经常出现要同时判断线程池运行状态和线程数量的情况。线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数。这里都使用的是位运算的方式，相比于基本运算，速度也会快很多。

关于内部封装的获取生命周期状态、获取线程池线程数量的计算方法如下代码：

哇，Doug Lea 大佬简直了，设计的真好。老周等等我，这里怎么设计的就好了？CAPACITY 这里是多少呀？

不着急，老周这就带你来分析分析为什么一个整型变量既可以保存运行状态，又可以保存线程数量？

首先，我们知道 Java 中 1 个整型占 4 个字节，也就是 32 位，所以 1 个整型有 32 位。

所以整型 1 用二进制表示就是：0000 0000 0000 0000 0000 0000 0000 0001

整型 -1 用二进制表示就是：1111 1111 1111 1111 1111 1111 1111 1111 (这个是补码，这个忘了的话那得去复习下原码、反码、补码等计算机基础知识了。)

在 ThreadPoolExecutor，整型中 32 位的前 3 位用来表示线程池状态，后 29 位表示线程池中有效的线程数。

这里你有可能问了，老周啊，CAPACITY = (1 << 29) - 1 怎么就得到 0001 1111 1111 1111 1111 1111 1111 1111。

好吧，老周就带你分析下 CAPACITY 怎么来的，下面的那些状态大家也可以自己去分析下哈。

我们先来看 1 << 29，首先看 1 的二进制代表 0000 0000 0000 0000 0000 0000 0000 0001。

然后 0000 0000 0000 0000 0000 0000 0000 0001 向左移 29 位，得到 0010 0000 0000 0000 0000 0000 0000 0000。

最后将 0010 0000 0000 0000 0000 0000 0000 0000 减 1 得到 0001 1111 1111 1111 1111 1111 1111 1111。

我们下面再来了解下 ThreadPoolExecutor 所定义的状态，这些状态都和线程的执行密切相关：

RUNNING：能接受新提交的任务，并且也能处理阻塞队列中的任务。
SHUTDOWN：指调用了 shutdown() 方法，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。
STOP：指调用了 shutdownNow() 方法，不再接受新提交的任务，同时抛弃阻塞队列里的所有任务并中断所有正在执行任务。
TIDYING：所有任务都执行完毕，workerCount 有效线程数为 0。
TERMINATED：终止状态，当执行 terminated() 后会更新为这个状态。

3.3 线程池如何管理任务

3.3.1 任务调度

任务调度是线程池的主要入口，当用户提交了一个任务，接下来这个任务将如何执行都是由这个阶段决定的。了解这部分就相当于了解了线程池的核心运行机制。

首先，所有任务的调度都是由 execute 方法完成的，比如我们业务代码中 threadPool.execute(new Job());。

这部分完成的工作是：检查现在线程池的运行状态、运行线程数、运行策略，决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。其执行过程如下：

首先检测线程池运行状态，如果不是 RUNNING，则直接拒绝，线程池要保证在 RUNNING 的状态下执行任务。
如果 workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。
如果 workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
如果 workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
如果 workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满，则根据拒绝策略来处理该任务，默认的处理方式是直接抛异常。

执行流程图如下：

3.3.2 待执行任务的队列

待执行任务的队列是线程池能够管理任务的核心部分。线程池的本质是对任务和线程的管理，而做到这一点最关键的思想就是将任务和线程两者解耦，不让两者直接关联，才可以做后续的分配工作。线程池中是以生产者消费者模式，通过一个阻塞队列来实现的。阻塞队列缓存任务，工作线程从阻塞队列中获取任务。

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。

这两个附加的操作是：

在队列为空时，获取元素的线程会等待队列变为非空。
当队列满时，存储元素的线程会等待队列可用。

阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

下图中展示了 Thread1 往阻塞队列中添加元素，而线程 Thread2 从阻塞队列中移除元素：

使用不同的队列可以实现不一样的任务存取策略。我们下面来看下阻塞队列的成员：

3.3.3 任务申请

从上文可知，任务的执行有两种可能：

一种是任务直接由新创建的线程执行
另一种是线程从任务队列中获取任务然后执行，执行完任务的空闲线程会再次去从队列中申请任务再去执行。

第一种情况仅出现在线程初始创建的时候，第二种是线程获取任务绝大多数的情况。

线程需要从待执行任务的队列中不断地取任务执行，帮助线程从阻塞队列中获取任务，实现线程管理模块和任务管理模块之间的通信。

这部分策略由 getTask 方法实现，我们来看下 getTask 方法的代码。

getTask 方法在阻塞队列中有待执行的任务时会从队列中弹出一个任务并返回，如果阻塞队列为空，那么就会阻塞等待新的任务提交到队列中直到超时（在一些配置下会一直等待而不超时），如果在超时之前获取到了新的任务，那么就会将这个任务作为返回值返回。所以一般 getTask 方法是不会返回 null 的，只会阻塞等待下一个任务并在之后将这个新任务作为返回值返回。

当 getTask 方法返回 null 时会导致当前 Worker 退出，当前线程被销毁。在以下情况下 getTask 方法才会返回 null：

当前线程池中的线程数超过了最大线程数。这是因为运行时通过调用 setMaximumPoolSize 修改了最大线程数而导致的结果；
线程池处于 STOP 状态。这种情况下所有线程都应该被立即回收销毁；
线程池处于 SHUTDOWN 状态，且阻塞队列为空。这种情况下已经不会有新的任务被提交到阻塞队列中了，所以线程应该被销毁；
线程可以被超时回收的情况下等待新任务超时。线程被超时回收一般有以下两种情况：
允许核心线程超时（线程池配置）的情况下线程等待任务超时
超出核心线程数部分的线程等待任务超时

3.3.4 任务拒绝

任务拒绝模块是线程池的保护部分，线程池有一个最大的容量，当线程池的任务缓存队列已满，并且线程池中的线程数目达到 maximumPoolSize 时，就需要拒绝掉该任务，采取任务拒绝策略，保护线程池。

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

用户可以通过实现这个接口去定制拒绝策略，也可以选择 JDK 提供的四种已有拒绝策略，其特点如下：

3.4 线程池如何管理线程

3.4.1 Worker 线程

线程池为了掌握线程的状态并维护线程的生命周期，设计了线程池内的工作线程 Worker。我们来看一下它的代码：

Worker 这个工作线程，实现了 Runnable 接口，并持有一个线程 thread，一个初始化的任务 firstTask。thread 是在调用构造方法时通过 ThreadFactory 来创建的线程，可以用来执行任务；

firstTask 用它来保存传入的第一个任务，这个任务可以有也可以为 null。如果这个值是非空的，那么线程就会在启动初期立即执行这个任务，也就对应核心线程创建时的情况；如果这个值是空的，那么就需要创建一个线程去执行任务列表（workQueue）中的任务，也就是非核心线程的创建。

3.4.1.1 AQS 作用

Worker 继承了 AbstractQueuedSynchronizer，主要目的有两个：

将锁的粒度细化到每个 Worker 如果多个 Worker 使用同一个锁，那么一个 Worker Running 持有锁的时候，其他 Worker 就无法执行，这显然是不合理的。
直接使用 CAS 获取，避免阻塞。如果这个锁使用阻塞获取，那么在多 Worker 的情况下执行 shutDown。如果这个 Worker 此时正在 Running 无法获取到锁，那么执行 shutDown() 线程就会阻塞住了，显然是不合理的。

3.4.1.2 Runnable 作用

Worker 还实现了 Runnable，它有两个属性 thead、firstTask。

firstTask 用它来保存传入的第一个任务，这个任务可以有也可以为 null。

如果这个值是非空的，那么线程就会在启动初期立即执行这个任务，也就对应核心线程创建时的情况。
如果这个值是 null，那么就需要创建一个线程去执行任务列表（workQueue）中的任务，也就是非核心线程的创建。

根据整体流程：

线程池调用 execute —> 创建 Worker（设置属性 thead、firstTask）—> worker.thread.start() —> 实际上调用的是 worker.run() —> 线程池的 runWorker(worker) —> worker.firstTask.run() (如果 firstTask 为 null 就从等待队列中拉取一个)。

Worker 执行任务的模型如下图所示：

3.4.2 Worker 线程增加

增加线程是通过线程池中的 addWorker 方法，该方法的功能就是增加一个线程，该方法不考虑线程池是在哪个阶段增加的该线程，这个分配线程的策略是在上个步骤完成的，该步骤仅仅完成增加线程，并使它运行，最后返回是否成功这个结果。

addWorker 方法有两个参数：firstTask、core。

firstTask 参数用于指定新增的线程执行的第一个任务，该参数可以为空；
core 参数为 true 表示在新增线程时会判断当前活动线程数是否少于 corePoolSize，false 表示新增线程前需要判断当前活动线程数是否少于 maximumPoolSize。

我们来看一下 addWorker 的源码：

源码看着是不是挺费劲的？没关系，再看一张执行流程图加深下映象。

3.4.3 Worker 线程执行任务

Worker 中的线程 start 的时候，调用 Worker 本身 run 方法，这个 run 方法调用外部类 ThreadPoolExecutor 的 runWorker 方法，直接看 runWorker 方法的源码：

执行流程如下：

while 循环不断地通过 getTask() 方法获取任务
getTask() 方法从阻塞队列中取任务
如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态。
执行任务
如果 getTask 结果为 null 则跳出循环，执行 processWorkerExit() 方法，销毁线程。

3.4.4 Worker 线程回收

线程池中线程的销毁依赖 JVM 自动的回收，线程池做的工作是根据当前线程池的状态维护一定数量的线程引用，防止这部分线程被 JVM 回收，当线程池决定哪些线程需要回收时，只需要将其引用消除即可。Worker 被创建出来后，就会不断地进行轮询，然后获取任务去执行，核心线程可以无限等待获取任务，非核心线程要限时获取任务。当 Worker 无法获取到任务，也就是获取的任务为空时，循环会结束，Worker 会主动消除自身在线程池内的引用。

线程回收的工作是在 processWorkerExit 方法完成的。

在回收 Worker 的时候线程池会尝试结束自己的运行，tryTerminate 方法：

3.4.4 Worker 线程关闭

说到线程关闭，我们就不得不来说说 shutdown 方法和 shutdownNow 方法。

3.4.4.1 shutdown

interruptIdleWorkers 方法，注意，这个方法打断的是闲置 Worker，打断闲置 Worker 之后，getTask 方法会返回 null，然后 Worker 会被回收。那什么是闲置 Worker 呢？

闲置 Worker 是这样解释的：Worker 运行的时候会去阻塞队列拿数据(getTask 方法)，拿的时候如果没有设置超时时间，那么会一直阻塞等待阻塞队列进数据，这样的 Worker 就被称为闲置 Worker。由于 Worker 也是一个 AQS，在 runWorker 方法里会有一对 lock 和 unlock 操作，这对 lock 操作是为了确保 Worker 不是一个闲置 Worker。

所以 Worker 被设计成一个 AQS 是为了根据 Worker 的锁来判断是否是闲置线程，是否可以被强制中断。

下面我们看下 interruptIdleWorkers 方法：