Java线程池深度揭秘
作为 JAVA 程序员，无论是技术面试、项目研发或者是学习框架源码，不彻底掌握 Java 多线程的知识，做不到心中有数，干啥都没底气，尤其是技术深究时往往略显发憷。
坐稳扶好，通过今天的分享，能让你轻松 get 如下几点。
1. Executor 框架家族简介；
2. 源码解读：线程池状态以及状态流转；
3. 源码解读：部分成员变量及方法；
4. 源码解读：任务提交submit方法背后；
5. 源码揭秘之后的反思；
6. 寄语。
Executor 家族简介
一图胜千言，脑中有图心不慌。
executor 家族简图
（一）Executor 接口。
public interface Executor {
void execute(Runnable command);
}
Executor 是一个接口（主要用于定义规范），定义了 execute 方法，用于接收 Runnable 对象。
（二）ExecutorService 接口。
public interface ExecutorService extends Executor {
// ... ...
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
// ... ...
}
ExecutorService 也是一个接口，继承了 Executor 接口，增加了更多方法，相当于扩展了 Executor 接口的功能，例如定义了 submit() 系列方法，支持任务执行后得到返回结果。
（三）AbstractExecutorService 抽象类。
public abstract class AbstractExecutorService implements ExecutorService {
// ... ...
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
// ... ...
}
AbstractExecutorService 是一个抽象类，实现了 ExecutorService 接口中的部分方法，例如提供了任务提交的 submit 方法的默认实现，而 submit 方法最终会调用 execute 方法。
不过 AbstractExecutorService 并没有实现 execute 方法，相当于为子类留了个口子，让子类去灵活扩展（钩子函数）。
（四）ScheduledExecutorService 接口。
public interface ScheduledExecutorService extends ExecutorService {
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
}
ScheduledExecutorService 接口继承了 ExecutorService，增加定时调度的方法，使其成为一个可定时调度任务的接口，相当于扩展了 ExecutorService 的功能。
（五）ScheduledThreadPoolExecutor 类。
public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
// ... ...
}
ScheduledThreadPoolExecutor 类继承自 ThreadPoolExecutor 类，并且实现了 ScheduledExecutorService 接口，变成一个可定时调度任务的线程池。
（六）ThreadPoolExecutor 类。
public class ThreadPoolExecutor extends AbstractExecutorService {
// ... ...
}
ThreadPoolExecutor 继承 AbstractExecutorService 抽象类，并实现了 execute 等一系列方法。
（七）Executors 类。
public class Executors {
// ... ...
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
// ... ...
}
研发人员可以通过 Executors 工厂类来创建线程池并返回一个ExecutorService 对象，而内部几乎全是对 ThreadPoolExecutor 的封装。
通过 Executor 的家族简单认识，应该能感觉到 ThreadPoolExecutor 类的重要性，所以接下来要重点对 ThreadPoolExecutor 类的源码进行剖析。
源码解读：线程池状态以及状态流转
上面注释截图来源于 ThreadPoolExecutor 的源码，别懵圈，仔细看差不多都能懂，能够看出线程池的五种状态以及对应的状态流转。
不知道你能看懂多少，看不懂也没关系，接下来把上面的注释用图呈现给大家。通过源码中的注释，能够勾勒出如下线程池的状态流转图（好的注释是多么的重要啊，感叹号！）。
源码解读：部分成员变量及方法
/**
* ctl 是一个 AtomicInteger 类型的原子对象。
* 其实设计很有意思：ctl 共包括 32 位（高 3 位表示"线程池状态"，低 29 位表示"线程池中的线程数量"）。
* 个人感觉：线程池状态与线程数量合二为一，用一个变量来表示，来减少锁竞争，提高并发效率。
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/** 表示线程池线程数的位数：32 - 3 = 29 位 */
private static final int COUNT_BITS = Integer.SIZE - 3;
/** 表示最大线程容量（000,11111111111111111111111111111）*/
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits（运行状态保存在 int 值的高 3 位）
/** 111,00000000000000000000000000000 */
private static final int RUNNING = -1 << COUNT_BITS;
/** 000,00000000000000000000000000000 */
private static final int SHUTDOWN = 0 << COUNT_BITS;
/** 001,00000000000000000000000000000 */
private static final int STOP = 1 << COUNT_BITS;
/** 010,00000000000000000000000000000 */
private static final int TIDYING = 2 << COUNT_BITS;
/** 011,00000000000000000000000000000 */
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
/** 获取线程池的运行状态 */
private static int runStateOf(int c) { return c & ~CAPACITY; }
/** 线程池内有效线程的数量 (workerCount) */
private static int workerCountOf(int c) { return c & CAPACITY; }
/** 线程池的状态和线程的数量组装，成为 ctl */
private static int ctlOf(int rs, int wc) { return rs | wc; }
仔细去看上面的代码，注释已经很清晰啦。重点关注 ctl 变量，这个变量将线程池自身状态和线程数量，融合在这一个变量中，其中高 3 位表示线程池状态，低 29 位表示线程池中的线程数量，这样在多线程环境下更易保证线程池自身状态和线程数量的统一，不得不佩服源代码作者 Doug Lea，可谓是设计甚妙！
源码解读：任务提交 submit 方法背后
疑问？当调用 submit() 方法，把一个任务提交给线程池去处理的时候，线程池的处理过程是什么样的呢？
通过开篇对 Executor 的家族简介，能够看到 submit() 方法最终会调用 ThreadPoolExecutor 的 execute 方法，走进源码好好看看 execute 方法都做了啥？
重点关注源码中的注释（红框圈住部分），若看懂注释，提交任务时线程池对应的处理，也就懂了一半啦（感触：好的编码规范真的好重要，业务开发时，核心代码一定要有注释）。
若依然很懵逼，一图胜千言，那就继续上图吧。
了解上图的整体流程，再去看看源码就彻悟啦。
public void execute(Runnable command) {
//【Step 0. 如果任务为空则抛出 NPE 异常】
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//【Step 1. 判断核心线程是否已满】
// 1.1. 判断当前线程数是否已经达到核心线程数的限制
if (workerCountOf(c) < corePoolSize) {
// 1.2. 如果未达到核心线程数的限制，则会直接添加一个核心线程，并指定首次执行的任务，进行任务处理
if (addWorker(command, true))
return;
// 1.3. 如果添加失败，则刷新线程池的状态和线程的数量对应的变量 ctl
c = ctl.get();
}
//【Step 2. 判断阻塞队列是否已满】
// 2.1. 检查线程池是否是运行状态，然后将任务添加到等待队列
if (isRunning(c) && workQueue.offer(command)) {
// 2.2. 任务成功添加到等待队列，再次刷新 ctl
int recheck = ctl.get();
// 2.3. 添加任务到等待队列成功后，如果线程池不是运行状态，则将刚添加的任务从队列移除并执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 2.4. 判断当前线程数量，如果线程数量为 0，则添加一个非核心线程，并且不指定首次执行任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//【Step 3. 判断最大线程数量是否已经达到】
// 3.1. 添加非核心线程，指定首次执行任务，如果添加失败，执行异常策略
else if (!addWorker(command, false))
reject(command);
}
结合注释去读代码，应该都能搞懂。很显然 execute 方法中，多处都调用了 addWorker 方法，接下来简单剖析一下 addWorker 方法。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
// ... ...
for (;;) {
// ... ...
// 通过 CAS 自旋，增加线程数 +1，增加成功跳出双层循环，继续往下执行
if (compareAndIncrementWorkerCount(c))
break retry;
// ... ...
}
}
// 到这儿，说明已经成功的将线程数 +1 了，但是真正的线程还没有被添加
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 添加线程，Worker 是继承了 AQS，实现了 Runnable 接口的包装类
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// ... ...
// 添加新增的 Worker
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
// ... ...
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 启动 Worker
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
为了简明扼要，方法酌情进行了删减。addWorker 方法主要是通过双重 for 循环进行线程数 +1，然后创建 Worker，并进行添加到 HashSet<Worker> workers 列表中，然后调用 t.start() 启动 Worker。
那么接下来再一起看看 Worker 里面都做了啥？
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable {
// ... ...
final Thread thread;
Runnable firstTask;
/**
* 通过指定的 firstTask 任务创建 Worker 对象
*/
Worker(Runnable firstTask) {
setState(-1);
this.firstTask = firstTask;
// 通过当前 Worker 对象创建对应的线程对象 t，
// 所以调用 t.start() 时最终会调用 Worker 的 run 方法
this.thread = getThreadFactory().newThread(this);
}
public void run() {
// run 方法最终会调用 ThreadPoolExecutor 的 runWorker 方法
runWorker(this);
}
// ... ...
}
通过 Worker 的构造函数能够了解到，会通过创建的 Worker 对象去构建线程对象，当线程对象启动时最终会调用 runWorker 方法。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 取出需要执行的任务
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 如果 task 不是 null 或者去 workQueue 队列中取到待执行的任务不为 null
while (task != null || (task = getTask()) != null) {
// ... ...
try {
// 开始执行任务前的钩子方法
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
// ... ...
} finally {
// 任务执行后的钩子方法
afterExecute(task, thrown);
}
} finally {
// ... ...
}
}
completedAbruptly = false;
} finally {
// Worker 退出
processWorkerExit(w, completedAbruptly);
}
}
runWorker 方法，首先会取出要执行的任务 task，如果为空则会调用 getTask 方法从任务队列中获取，然后调用任务对应的 run 方法进行执行，另外预置了 beforeExecute、afterExecute 两个钩子函数，让研发人员监控线程执行成为可能。
另外，线程池中的线程如何从队列中获取待执行的任务的呢？走进 getTask 方法看一看。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
// 这块体现了：线程池的线程是复用的，通过循环去获取队列中的任务去执行。
for (;;) {
int c = ctl.get();
// ... ...
int wc = workerCountOf(c);
// allowCoreThreadTimeOut: 是否允许核心线程超时.
// 如果设置为 false，那么线程池在达到 corePoolSize 个工作线程之前，不会让闲置的工作线程退出。
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// ... ...
try {
// 从 workQueue 队列中取待执行的任务，根据 timed 来选择等待时间
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
为了便于理解，源码做了部分删减。重点关注从任务队列中获取待执行任务的对象的方法调用：workQueue.poll()、workQueue.take() ，前者是移除并返回队列中的头部元素，如果队列为空，则返回 null，而后者是移除并返回队列中的头部元素，如果队列为空，则阻塞。
烟未灭，酒过半 ... ... 源码探讨就谈到这儿... ...
源码揭秘之后的反思
（一）钩子函数的使用场景
场景一：
如上面自定义的 MyThreadPoolExecutor，可以让日志打印线程及线程数等等信息。意味着研发人员可以扩展 ThreadPoolExecutor，对钩子函数 beforeExecute、afterExecute 进行实现，进而可以知晓线程池内部的调度细节，可以有效进行监控，针对故障排查应该很有帮助。
场景二：
AbstractExecutorService 并没有实现 execute 方法，而是为子类 ThreadPoolExecutor 留了个口子，让子类去灵活扩展（钩子函数）。
仔细想想业务开发时，诸多的使用场景，何尝不是如此呢？
（二）线程池的 submit 方法与 execute 方法啥区别呢？
execute 方法，适用于不需要关注返回值的场景，只需要将线程丢到线程池中去执行就可以了。
而 submit() 方法，适用于需要关注返回值的场景，不过最终会调用 execute() 方法。
考虑到性能提升，如果不需要关注返回值，则建议直接调用 execute() 方法，因为那样会屏蔽很多中间调度。
（三）线程池状态与线程数量用一个 ctl 变量表示的好处？
线程池状态和线程数量合二为一，用一个原子变量来表示，来减少锁竞争，提高并发效率。
（四）清晰的注释是否有必要？
通过探秘源码，很多图都是根据源码注释勾勒出来的。可以看出清晰的注释，对于核心流程而言真的很重要，一定要养成良好的编码习惯，关键业务逻辑、核心流程，建议一定要写好注释，利人又利己，何乐而不为之。
（五）Executor 家族框架，若写基础框架时，是否有借鉴意义呢？
个人感觉很有借鉴意义，因为无论业务开发还是基础服务，总会看到类似模式框架的身影，总会有大牛模仿着造轮子，所以闲暇之余可以抽象一下。
寄语写最后
本次，主要对 Executor 家族进行了简单介绍，并着重对线程池背后的 ThreadPoolExecutor 类进行深度剖析，知其然知其所以然，希望对大家有帮助。
