Java版本：8u261。
对于Java中的线程池，面试问的最多的就是线程池中各个参数的含义，又或者是线程池执行的流程，彷佛这已成为了固定的模式与套路。但是假如我是面试官，现在我想问一些更细致的问题，你还能答得上来吗？比如：
- 线程池是如何实现线程复用的？
- 如果一个线程执行任务的时候抛出异常，那么这个任务是否会被丢弃？
- 当前线程池中有十个线程，其中一个线程正在执行任务，那么剩下的九个线程正在处于一种什么状态呢？
相信如果没有看过线程池的相关源码实现，这些问题是很难回答得完美的。同时这些问题往深了问还会引出Java中阻塞队列以及AQS的实现，你都能接得住吗？
1 简介
因为线程是稀缺资源，如果在高并发的情况下被无限制地创建和销毁，不仅会消耗系统资源，还会降低系统的稳定性。所以线程池的出现就是为了解决这些问题的。线程池通过重用已经存在的线程资源，减少线程创建和销毁的次数，提高了性能。同时还可以进行统一的分配、调优和监控。
在Java中，可以通过Executors类中的newFixedThreadPool、newCachedThreadPool，newScheduledThreadPool或者其他方式来创建各种线程池，它们都会直接或间接地通过ThreadPoolExecutor来进行构建，通过传入不同的参数来实现不同效果的线程池（newScheduledThreadPool比较特殊，它重写了部分ThreadPoolExecutor的逻辑，后续我会写一篇对ScheduledThreadPoolExecutor进行源码分析的文章）。
1.1 线程池参数
在ThreadPoolExecutor中共有七个参数：
-
corePoolSize：核心线程数，核心线程会一直存活，即使没有任务需要执行（除非allowCoreThreadTimeOut参数设置为true，这样的话即使是核心线程也会被超时销毁）；
-
maximumPoolSize：线程池中允许的最大线程数；
-
keepAliveTime：维护工作线程所允许的空闲时间，如果工作线程等待的时间超过了keepAliveTime，则会被销毁；
-
unit：指定keepAliveTime的单位，如TimeUnit.SECONDS；
-
workQueue：用来保存等待被执行任务的阻塞队列。常用的有：ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue和PriorityBlockingQueue等；
-
threadFactory：线程工厂，提供创建新线程的功能。默认的实现是Executors.defaultThreadFactory()，即通过new Thread的方式；
-
handler：如果当前阻塞队列已满，并且当前的线程数量已超过了最大线程数，则会执行相应的拒绝策略。具体有四种（也可以自己实现）：
-
- AbortPolicy：默认实现，会直接抛出RejectedExecutionException；
- CallerRunsPolicy：用调用者所在的线程来执行任务；
- DiscardPolicy：直接抛弃，任务不执行；
- DiscardOldestPolicy：丢弃阻塞队列中最靠前的任务，并执行当前任务。
这四种拒绝策略的实现很简单，这里就不再过多展示说明了，读者可自行查看。
1.2 运行过程
ThreadPoolExecutor的大致运行过程如下：
如果使用的是有界阻塞队列：
有新的任务需要执行，并且当前线程池的线程数小于核心线程数，则创建一个核心线程来执行。如果当前线程数大于核心线程数，则会将除了核心线程处理的任务之外剩下的任务加入到阻塞队列中等待执行。如果队列已满，则在当前线程数不大于最大线程数的前提下，创建新的非核心线程，处理完毕后等到达keepAliveTime空闲时间后会被直接销毁（注意，不一定销毁的就是这些非核心线程，核心线程也可能被销毁，只要减到剩余线程数到达核心线程数就行。核心线程和非核心线程的区别仅在于判断是否到达阈值时有区别：核心线程判断的是核心线程数，而非核心线程判断的是最大线程数。仅此一个区别。后面讲源码时会再强调这一点）。如果当前线程数大于最大线程数，则会执行相应的拒绝策略。
如果使用的是无界阻塞队列：
与有界阻塞队列相比，除非系统资源耗尽，否则无界的阻塞队列不存在任务入队失败的情况。当有新任务到来，系统的线程数小于核心线程数时，则创建一个核心线程来执行。当达到核心线程数后，就不会继续增加。若后续仍有新的任务加入，而没有空闲的线程资源，则任务直接进入阻塞队列中进行等待。如果任务创建和处理任务的速度差异很大，无界阻塞队列会保持快速增长，直到耗尽系统内存。
1.3 线程池状态
在ThreadPoolExecutor中存在五种状态：
- RUNNING：初始状态，在此状态下能够接收新任务，以及对已经添加的任务进行处理；
- SHUTDOWN：通过调用shutdown方法，线程池转成SHUTDOWN状态。此时不再接收新任务，但是能处理已经添加的任务；
- STOP：通过调用shutdownNow方法，线程池转成STOP状态。此时不再接收新任务，不处理已经添加的任务，并且会中断正在处理的任务；
- TIDYING：当线程池中所有的任务已经终止了，任务数量为0并且阻塞队列为空的时候，会进入到TIDYING状态。此时会调用一个钩子方法terminated，它是一个空的实现，可以供调用者覆写；
- TREMINATED：线程池彻底终止的状态。当线程池处于TIDYING状态时，执行完terminated方法后，就会进入到该状态。
在ThreadPoolExecutor中状态是通过ctl属性中的高3位来表示的：
1 //ctl中包含两部分信息：高3位表示运行状态，低29位保存工作线程数量，初始状态是RUNNING
2 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
3 //29
4 private static final int COUNT_BITS = Integer.SIZE - 3;
5 //1左移29位后-1，也就是29个1。用来表示工作线程数量的最大值
6 private static final int CAPACITY = (1 << COUNT_BITS) - 1;
7
8 //ctl高3位为111（低29位都为0）
9 private static final int RUNNING = -1 << COUNT_BITS;
10 //ctl高3位为000（低29位都为0）
11 private static final int SHUTDOWN = 0 << COUNT_BITS;
12 //ctl高3位为001（低29位都为0）
13 private static final int STOP = 1 << COUNT_BITS;
14 //ctl高3位为010（低29位都为0）
15 private static final int TIDYING = 2 << COUNT_BITS;
16 //ctl高3位为011（低29位都为0）
17 private static final int TERMINATED = 3 << COUNT_BITS;
18
19 //获取ctl的高3位（低29位都为0）也就是获取运行状态
20 private static int runStateOf(int c) {
21 return c & ~CAPACITY;
22 }
23
24 //获取ctl的低29位（高3位都为0）也就是获取工作线程数量
25 private static int workerCountOf(int c) {
26 return c & CAPACITY;
27 }
28
29 //用来获取运行状态和工作线程数量拼接起来的值
30 private static int ctlOf(int rs, int wc) {
31 return rs | wc;
32 }
33
34 //判断ctl是否小于s所代表状态的值
35 private static boolean runStateLessThan(int c, int s) {
36 return c < s;
37 }
38
39 //判断ctl是否大于等于s所代表状态的值
40 private static boolean runStateAtLeast(int c, int s) {
41 return c >= s;
42 }
43
44 //判断ctl此时是否是RUNNING状态
45 private static boolean isRunning(int c) {
46 return c < SHUTDOWN;
47 }
1.4 Worker
Worker是ThreadPoolExecutor中的一个内部类，用来封装工作线程：
1 private final class Worker
2 extends AbstractQueuedSynchronizer
3 implements Runnable {
4 //...
5
6 //正在运行Worker的线程
7 final Thread thread;
8 //传入的任务
9 Runnable firstTask;
10 //本Worker已完成的任务数，用于后续的统计与监控
11 volatile long completedTasks;
12
13 Worker(Runnable firstTask) {
14 /*
15 这里设置AQS的state初始为-1是为了将线程发生中断的动作延迟到任务真正开始运行的时候，换句话说就是
16 禁止在执行任务前对线程进行中断。在调用一些像shutdown和shutdownNow等方法中会去中断线程，而在
17 中断前会调用tryLock方法尝试加锁。而这里设置为-1后，tryLock方法就会返回为false，所以就不能中断了
18 */
19 setState(-1);
20 this.firstTask = firstTask;
21 this.thread = getThreadFactory().newThread(this);
22 }
23
24 //因为Worker类实现了Runnable接口，所以当调用thread.start方法时最终会调用到此处运行
25 public void run() {
26 runWorker(this);
27 }
28
29 //...
30 }
由上可以看到Worker继承了AQS（我之前写过对AQS、ReentrantLock和阻塞队列进行源码分析的文章，感兴趣的可以查看AQS源码深入分析之独占模式-ReentrantLock锁特性详解、AQS源码深入分析之条件队列-你知道Java中的阻塞队列是如何实现的吗？），并实现了Runnable接口。之后在分析源码时将会看到：在运行Worker的时候之所以没有用ReentrantLock作为独占锁来使用是因为这里是要求不可重入的，而ReentrantLock是可重入锁。在像一些setCorePoolSize方法去手动更改核心线程数时，如果修改的值比原本的小，那么多余的线程会被中断、会中断正在运行着的的线程。所以使用自己实现的不可重入独占锁而不是使用ReentrantLock就是为了不想让像setCorePoolSize这样的方法来重新获取到锁资源，不想让正在运行的线程发生自我中断。其实上面所说的内容在Worker类的注释中都已经解释了：
2 构造器
1 /**
2 * ThreadPoolExecutor:
3 * 全参数构造器，其他构造器最终都会调用到这里
4 */
5 public ThreadPoolExecutor(int corePoolSize,
6 int maximumPoolSize,
7 long keepAliveTime,
8 TimeUnit unit,
9 BlockingQueue<Runnable> workQueue,
10 ThreadFactory threadFactory,
11 RejectedExecutionHandler handler) {
12 //非法参数校验
13 if (corePoolSize < 0 ||
14 maximumPoolSize <= 0 ||
15 maximumPoolSize < corePoolSize ||
16 keepAliveTime < 0)
17 throw new IllegalArgumentException();
18 //非空校验
19 if (workQueue == null || threadFactory == null || handler == null)
20 throw new NullPointerException();
21 //如果安全管理器不为空，就进行权限访问（本文不展开分析）
22 this.acc = System.getSecurityManager() == null ?
23 null :
24 AccessController.getContext();
25 this.corePoolSize = corePoolSize;
26 this.maximumPoolSize = maximumPoolSize;
27 this.workQueue = workQueue;
28 //将keepAliveTime转换成纳秒
29 this.keepAliveTime = unit.toNanos(keepAliveTime);
30 this.threadFactory = threadFactory;
31 this.handler = handler;
32 }
3 execute方法
1 /**
2 * ThreadPoolExecutor:
3 */
4 public void execute(Runnable command) {
5 //非空校验
6 if (command == null)
7 throw new NullPointerException();
8 int c = ctl.get();
9 //如果当前线程数小于核心线程数的话，就直接创建一个核心线程
10 if (workerCountOf(c) < corePoolSize) {
11 if (addWorker(command, true))
12 return;
13 /*
14 添加失败（可能是线程池状态是SHUTDOWN或以上的状态（SHUTDOWN状态下不再接收
15 新任务），也可能是线程数超过阈值了），就重新获取一下ctl的值，走下面的逻辑
16 */
17 c = ctl.get();
18 }
19 /*
20 走到这里说明当前线程数大于等于核心线程数，又或者是上面添加核心线程失败中解释的情况
21 此时就判断一下当前线程池是否是RUNNING状态，如果是的话就往阻塞队列入队
22 这里offer跟put的区别是如果队列已满，offer不会被阻塞，而是立即返回false
23 */
24 if (isRunning(c) && workQueue.offer(command)) {
25 int recheck = ctl.get();
26 /*
27 这里会再次检查一次当前线程池是否是RUNNING状态，可能此时线程池已经shutdown了
28 如果不是RUNNING状态，就删除上面入队的任务，并执行相应的拒绝策略
29 */
30 if (!isRunning(recheck) && remove(command))
31 reject(command);
32 /*
33 此时还会去判断一下是否当前的工作线程数已经为0了（可能这些线程在上次workerCountOf
34 检查后（第10行代码处）被销毁了（allowCoreThreadTimeOut设置为true）），如果是
35 的话就新创建一个空任务的非核心线程。注意，这里传进addWorker方法的是空任务，因为任务
36 已经在阻塞队列中存在了，所以这个Worker执行的时候，会直接从阻塞队列中取出任务来执行
37 所以说这里的意义也就是要保证线程池在RUNNING状态下必须要有一个线程来执行任务
38 */
39 else if (workerCountOf(recheck) == 0)
40 addWorker(null, false);
41 } else if (!addWorker(command, false))
42 /*
43 走到这里说明线程池不是RUNNING状态，或者阻塞队列已满，此时创建一个非核心线程去执行
44 如果创建失败，说明线程池的状态已经不是RUNNING了，又或者当前线程数已经大于等于最大线程数了
45 那么就执行相应的拒绝策略
46 */
47 reject(command);
48 }
49
50 /**
51 * 第30行代码处：
52 */
53 public boolean remove(Runnable task) {
54 //阻塞队列中删除这个任务
55 boolean removed = workQueue.remove(task);
56 //根据线程池状态来判断是否应该结束线程池
57 tryTerminate();
58 return removed;
59 }
60
61 /**
62 * 第31行和第47行代码处：
63 */
64 final void reject(Runnable command) {
65 //根据是哪种拒绝策略，来具体执行其中的逻辑（具体的四种拒绝策略的代码这里就不再看了，都是很简单的）
66 handler.rejectedExecution(command, this);
67 }
4 addWorker方法
在上面添加任务时会调用到addWorker方法：
1 /**
2 * ThreadPoolExecutor:
3 */
4 private boolean addWorker(Runnable firstTask, boolean core) {
5 retry:
6 for (; ; ) {
7 int c = ctl.get();
8 //获取当前线程池的运行状态
9 int rs = runStateOf(c);
10
11 /*
12 如果当前线程池状态大于SHUTDOWN，就直接返回false，表示不再添加新的Worker
13 如果当前线程池的状态是SHUTDOWN（此时不再接收新的任务，但是还是会继续处理
14 阻塞队列中的任务），但是firstTask不为null（相当于新的任务）或者阻塞队列为空
15 （为空说明也没有必要去创建Worker了）的话，也直接返回false，不再添加新的Worker
16 */
17 if (rs >= SHUTDOWN &&
18 !(rs == SHUTDOWN &&
19 firstTask == null &&
20 !workQueue.isEmpty()))
21 return false;
22
23 for (; ; ) {
24 //重新获取当前线程池的工作线程数
25 int wc = workerCountOf(c);
26 /*
27 <1>如果当前线程数大于等于最大值；
28 <2.1>如果是核心线程，当前线程数大于等于核心线程数；
29 <2.2>如果是非核心线程，当前线程数大于等于最大线程数
30 以上两个条件任意一个满足，就说明当前线程数已经达到阈值了，
31 也直接返回false，不再添加新的任务
32 */
33 if (wc >= CAPACITY ||
34 wc >= (core ? corePoolSize : maximumPoolSize))
35 return false;
36 /*
37 CAS尝试对ctl+1，也就是工作线程数量+1。如果成功了，就跳出死循环，
38 从第58行代码处继续往下执行
39 */
40 if (compareAndIncrementWorkerCount(c))
41 break retry;
42 //如果CAS+1失败了，重新读此时ctl的最新值
43 c = ctl.get();
44 /*
45 如果发现此时的运行状态和之前刚进入该方法时的运行状态不相等，
46 说明在此期间发生了状态的改变，那么就从头开始重试
47 */
48 if (runStateOf(c) != rs)
49 continue retry;
50 /*
51 走到这里说明状态没有发生改变，但是之前ctl+1的CAS操作失败了，那么重新从第25
52 行代码处继续往下执行
53 */
54 }
55 }
56
57 //上面的死循环主要是为了对ctl做+1的操作，而下面是为了创建Worker
58 boolean workerStarted = false;
59 boolean workerAdded = false;
60 Worker w = null;
61 try {
62 //根据firstTask来创建一个Worker（如上面所说，AQS中的state初始值为-1，防止被中断）
63 w = new Worker(firstTask);
64 //每一个Worker都会创建一个Thread
65 final Thread t = w.thread;
66 if (t != null) {
67 final ReentrantLock mainLock = this.mainLock;
68 //上锁
69 mainLock.lock();
70 try {
71 //重新获取当前线程池的运行状态
72 int rs = runStateOf(ctl.get());
73
74 /*
75 如果线程池当前状态是RUNNING状态，或者是SHUTDOWN状态并且firstTask
76 为空（意味着不去处理新任务而是去处理阻塞队列中的任务），才能将创建的
77 新Worker添加到workers集合中
78 */
79 if (rs < SHUTDOWN ||
80 (rs == SHUTDOWN && firstTask == null)) {
81 /*
82 此时线程还没有start，但是isAlive方法返回true，说明这个线程是有问题的，
83 直接抛出异常
84 */
85 if (t.isAlive())
86 throw new IllegalThreadStateException();
87 //在workers集合（HashSet，因为已经加锁了，所以HashSet就行）里面添加本Worker
88 workers.add(w);
89 int s = workers.size();
90 /*
91 如果当前线程池中线程数量超过了largestPoolSize，就更新一下largestPoolSize为
92 当前线程数量，即largestPoolSize中保存着线程池中出现过的最大线程数，用于统计监控
93 */
94 if (s > largestPoolSize)
95 largestPoolSize = s;
96 //创建Worker成功
97 workerAdded = true;
98 }
99 } finally {
100 //释放锁
101 mainLock.unlock();
102 }
103 if (workerAdded) {
104 //如果上面workers集合添加Worker成功，就用Worker中的thread来启动线程
105 t.start();
106 workerStarted = true;
107 }
108 }
109 } finally {
110 if (!workerStarted)
111 //如果没添加成功，就执行失败处理
112 addWorkerFailed(w);
113 }
114 return workerStarted;
115 }
116
117 private void addWorkerFailed(Worker w) {
118 final ReentrantLock mainLock = this.mainLock;
119 //上锁
120 mainLock.lock();
121 try {
122 //如果之前创建Worker成功了，就从workers集合中删除它
123 if (w != null)
124 workers.remove(w);
125 //将ctl-1，里面使用了死循环确保CAS操作一定成功
126 decrementWorkerCount();
127 //根据线程池状态来判断是否应该结束线程池
128 tryTerminate();
129 } finally {
130 //释放锁
131 mainLock.unlock();
132 }
133 }
5 runWorker方法
因为Worker类实现了Runnable接口，所以当调用thread.start方法时最终会调用到Worker的run方法处：
1 /**
2 * ThreadPoolExecutor:
3 * 当调用t.start()方法时最终会调用到此处
4 */
5 public void run() {
6 runWorker(this);
7 }
8
9 final void runWorker(Worker w) {
10 //获取当前线程（当前线程也就是在Worker中的thread）
11 Thread wt = Thread.currentThread();
12 Runnable task = w.firstTask;
13 //把Worker中的firstTask清空，因为下面要执行它了
14 w.firstTask = null;
15 /*
16 因为之前创建Worker的时候将AQS的state初始为-1，是为了防止线程被中断
17 而这里unlock方法是把state重置为0，意思就是已经进入到runWorker方法
18 中，可以允许中断了
19 */
20 w.unlock();
21 boolean completedAbruptly = true;
22 try {
23 //如果task不为空，或者从阻塞队列中拿取到任务了
24 while (task != null || (task = getTask()) != null) {
25 /*
26 上锁（注意，这里是用Worker而不是ReentrantLock来加锁的，为了确保
27 以下的代码不会被同一线程所重入，同时可以做到不同线程可以并发执行）
28 */
29 w.lock();
30 /*
31 如果当前线程池状态大于等于STOP，确保当前线程也是需要中断的（因为这个时候要
32 结束线程池了，不能再添加新的线程）；否则如果在上面这个判断不满足之后调用了shutdownNow
33 方法的时候（注意，shutdownNow方法是ReentrantLock上锁，而代码走到
34 这里是当前Worker上锁，两者上的不是同一个锁，所以可以并发执行），
35 之前的状态要么是RUNNING要么是SHUTDOWN，在走完第一个runStateAtLeast
36 判断条件发现不满足后，现在执行了shutdownNow方法将状态改为了STOP，
37 同时设置Worker中断位。那么此时在该处的第二个判断Thread.interrupted()返回true，
38 同时线程池的状态此时已经改为了STOP，那么也会去中断这个线程（注意，这里说的
39 乃至整个ThreadPoolExecutor中我说的中断线程并不是会去真的中断，
40 wt.interrupt()只是会设置一个中断标志位，需要使用者在run方法中首先
41 通过isInterrupted方法去进行判断，是否应该执行接下来的业务代码）
42 */
43 if ((runStateAtLeast(ctl.get(), STOP) ||
44 (Thread.interrupted() &&
45 runStateAtLeast(ctl.get(), STOP))) &&
46 !wt.isInterrupted())
47 wt.interrupt();
48 try {
49 //钩子方法，空实现
50 beforeExecute(wt, task);
51 Throwable thrown = null;
52 try {
53 //这里就是在具体执行线程的任务了（也就是使用者具体写的任务）
54 task.run();
55 } catch (RuntimeException x) {
56 thrown = x;
57 throw x;
58 } catch (Error x) {
59 thrown = x;
60 throw x;
61 } catch (Throwable x) {
62 thrown = x;
63 throw new Error(x);
64 } finally {
65 //钩子方法，空实现
66 afterExecute(task, thrown);
67 }
68 } finally {
69 //这里将task置为null，下次循环的时候就会在阻塞队列中拿取下一个任务了
70 task = null;
71 //完成的任务数+1
72 w.completedTasks++;
73 //释放锁
74 w.unlock();
75 }
76 }
77 //循环执行上面的while循环来拿取任务，而走到这里说明Worker和阻塞队列中都已经没有了任务
78 completedAbruptly = false;
79 } finally {
80 //最后对Worker做收尾工作
81 processWorkerExit(w, completedAbruptly);
82 }
83 }
84
85 private void processWorkerExit(Worker w, boolean completedAbruptly) {
86 /*
87 completedAbruptly为true表示在runWorker方法中的while循环中抛出了异常，那么此时
88 工作线程是没有-1的，需要-1（正常情况下在while循环最后一次调用getTask方法中会-1）
89 */
90 if (completedAbruptly)
91 decrementWorkerCount();
92
93 final ReentrantLock mainLock = this.mainLock;
94 //上锁
95 mainLock.lock();
96 try {
97 //累加所有Worker已经完成的任务数，用于统计监控
98 completedTaskCount += w.completedTasks;
99 /*
100 把当前Worker（也就是当前线程）剔除出workers集合中，等待GC
101 注意，能走到这里，说明在getTask方法中的timed标志位肯定为true（为false的话就会在getTask方法中的
102 take方法中一直被阻塞，中断唤醒也不可能，因为这种情况下还是会继续在getTask方法中循环）。那么无外乎两种情况，
103 要么是空闲的核心线程超时需要被销毁，要么是空闲的非核心线程超时需要被销毁。不管属于哪一种，当前线程都是
104 要被销毁的
105 */
106 workers.remove(w);
107 } finally {
108 //释放锁
109 mainLock.unlock();
110 }
111
112 //根据线程池状态来判断是否应该结束线程池
113 tryTerminate();
114
115 int c = ctl.get();
116 //如果当前线程池处在RUNNING或SHUTDOWN状态
117 if (runStateLessThan(c, STOP)) {
118 //通过之前的分析，如果completedAbruptly为false，表明此时已经没有任务可以执行了
119 if (!completedAbruptly) {
120 //如果allowCoreThreadTimeOut为true，min就为0，否则为核心线程数
121 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
122 /*
123 如果阻塞队列不为空（可能代码执行到这里阻塞队列中又有数据了），并且allowCoreThreadTimeOut
124 为true，就将min改为1
125 */
126 if (min == 0 && !workQueue.isEmpty())
127 min = 1;
128 /*
129 两种情况：
130 <1>如果阻塞队列不为空，并且allowCoreThreadTimeOut为true，就判断一下当前工作线程数是否大于等于1，
131 如果是的话就直接返回，不是的话说明当前没有工作线程了，就添加一个非核心线程去执行阻塞队列中的任务
132 <2>如果allowCoreThreadTimeOut为false，就判断一下下当前工作线程数是否大于等于核心线程数，如果是
133 的话就直接返回，不是的话说明当前工作线程数小于核心线程数，那么也去添加一个非核心线程
134 */
135 if (workerCountOf(c) >= min)
136 return;
137 }
138 /*
139 上面已经分析了在completedAbruptly为false时的两种情况，下面来分析第三种情况，也就是completedAbruptly为
140 true的时候。completedAbruptly为true表示在runWorker方法中的while循环中抛出了异常，那么也去添加一个
141 非核心线程（虽然之前那个报错的任务是会在finally子句中被清空的，但是在这之前使用者可以覆写afterExecute
142 钩子方法，在其中保存这个执行失败的任务，以此来进行后续的处理。从这个角度上来说，添加一个非核心线程还是
143 有意义的。另外，如之前的分析，在addWorker方法中的第34行代码处，核心线程和非核心线程的区别仅在于阈值的判断上，
144 其他都是一样的。所以这里添加一个非核心线程也是可以的，反正没达到阈值）
145 */
146 addWorker(null, false);
147 }
148 }6 getTask方法
由上所示，在第24行代码处，当本Worker中的task任务为空时，就会从阻塞队列中拿取任务，也就是调用到getTask方法：
1 /**
2 * ThreadPoolExecutor:
3 */
4 private Runnable getTask() {
5 //timedOut标志位用来判断poll方法拿取任务是否超时了
6 boolean timedOut = false;
7
8 for (; ; ) {
9 int c = ctl.get();
10 //重新获取当前线程池的运行状态
11 int rs = runStateOf(c);
12
13 /*
14 如果当前线程池是SHUTDOWN状态，并且阻塞队列为空的时候；或者当前线程池的状态大于等于STOP
15 以上两种情况都会将工作线程-1，直接返回null。因为这两种情况下不需要
16 获取任务了。工作线程-1后，后续会在processWorkerExit方法中从workers集合中剔除掉这个Worker等待GC的
17 */
18 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
19 decrementWorkerCount();
20 return null;
21 }
22
23 /*
24 走到这里说明当前线程池要么是RUNNING状态，要么是SHUTDOWN状态但是阻塞队列不为空（SHUTDOWN状态还是要
25 处理阻塞队列中的任务的）
26
27 重新获取当前线程池的工作线程数
28 */
29 int wc = workerCountOf(c);
30
31 /*
32 timed标志位表示工作线程是否需要超时销毁
33 如果allowCoreThreadTimeOut设置为true（表示空闲的核心线程也是要超时销毁的），或者当前线程数大于
34 核心线程数（这个条件代表的是空闲的非核心线程是要被销毁的，如果allowCoreThreadTimeOut为false，
35 那么线程池中最多保留“传进线程池中的核心线程数”个线程），就将timed置为true
36 */
37 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
38
39 /*
40 如果当前工作线程数大于最大线程数，可能是调用了setMaximumPoolSize方法，把最大线程数改小了（走到这里
41 说明addWorker方法运行成功，而在addWorker方法中的第34行代码处已经判断了大于最大线程数的情况）；
42 timedOut为true说明当前已经不是第一次循环了，在上次循环中已经发生了poll的超时。所以总结来说这个if条件的意思是：
43 <1.1>如果当前工作线程数大于最大线程数
44 <1.2>或者当前线程处于空闲状态并且是需要被销毁的
45 <2.1>并且当前工作线程要有多于一个
46 <2.2>或者当前阻塞队列是空的
47 满足上面两个条件，就将工作线程-1，去掉当前这个多余的线程，然后直接返回
48 */
49 if ((wc > maximumPoolSize || (timed && timedOut))
50 && (wc > 1 || workQueue.isEmpty())) {
51 //这里的方法和decrementWorkerCount方法的区别是不会死循环去一直CAS尝试，如果失败了就直接返回false
52 if (compareAndDecrementWorkerCount(c))
53 return null;
54 //如果CAS-1失败了，就进入到下次循环中继续判断即可
55 continue;
56 }
57
58 try {
59 /*
60 如果timed为true，则通过poll方法进行限时拿取（超过keepAliveTime时间没有拿取到，就直接返回null），
61 否则通过take方法进行拿取（如果阻塞队列为空，take方法在此时就会被阻塞住，也就是本线程会被阻塞住，直到
62 阻塞队列中有数据了。也就是说如果timed为false的话，这些工作线程会一直被阻塞在这里）
63 */
64 Runnable r = timed ?
65 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
66 workQueue.take();
67 if (r != null)
68 //如果拿取到任务了，就直接返回给Worker处理
69 return r;
70 /*
71 走到这里说明发生了poll超时，那么将timedOut标志位置为true，进入到下一次循环中重试
72 （大概率会走到第53行代码处返回null）
73 */
74 timedOut = true;
75 } catch (InterruptedException retry) {
76 //如果在阻塞的过程中发生了中断，那么将timedOut置为false，也进入到下一次循环中重试
77 timedOut = false;
78 }
79 }
80 /*
81 以上的逻辑说明了：核心线程和非核心线程的区别并不是在Worker中有个表示是否是核心线程的属性，Worker是无状态的，
82 每个Worker都是一样的。而区分是通过判断当前工作线程数是否大于核心线程数来进行的（因为只有阻塞队列满了的时候
83 才会去创建新的非核心线程，也就会使工作线程数大于核心线程数）。如果大于，那么不管之前这个线程到底是核心线程
84 还是非核心线程，现在我就认定当前这个线程就是“非核心线程“，那么等这个“非核心线程”空闲时间超过keepAliveTime后，
85 就会被销毁
86 */
87 }7 shutdown方法
关闭线程池时一般调用的是shutdown方法，而不是shutdownNow方法：
1 /**
2 * ThreadPoolExecutor:
3 */
4 public void shutdown() {
5 final ReentrantLock mainLock = this.mainLock;
6 //上锁
7 mainLock.lock();
8 try {
9 //如果有安全管理器，确保调用者有权限关闭线程池（本文不展开分析）
10 checkShutdownAccess();
11 //将线程池状态改为SHUTDOWN，里面使用了死循环确保CAS操作一定成功
12 advanceRunState(SHUTDOWN);
13 interruptIdleWorkers();
14 //钩子方法，空实现
15 onShutdown();
16 } finally {
17 //释放锁
18 mainLock.unlock();
19 }
20 //根据线程池状态来判断是否应该结束线程池
21 tryTerminate();
22 }
23
24 /**
25 * 第13行代码处：
26 */
27 private void interruptIdleWorkers() {
28 //中断所有的空闲线程
29 interruptIdleWorkers(false);
30 }
31
32 private void interruptIdleWorkers(boolean onlyOne) {
33 final ReentrantLock mainLock = this.mainLock;
34 //上锁
35 mainLock.lock();
36 try {
37 for (Worker w : workers) {
38 Thread t = w.thread;
39 if (!t.isInterrupted() && w.tryLock()) {
40 try {
41 /*
42 如果当前Worker中的线程没有被中断过，且尝试加锁成功，就将
43 中断标志位重新置为true，意思就是说要中断这个空闲的Worker
44 */
45 t.interrupt();
46 } catch (SecurityException ignore) {
47 } finally {
48 //将AQS中的state复位为0，恢复为tryLock之前的状态
49 w.unlock();
50 }
51 }
52 if (onlyOne)
53 //如果onlyOne为true，就只尝试中断一次
54 break;
55 }
56 } finally {
57 //释放锁
58 mainLock.unlock();
59 }
60 }8 tryTerminate方法
在上面的实现中可以看到有多处调用到了tryTerminate方法，以此来判断当前线程池是否应该结束：
1 /**
2 * ThreadPoolExecutor:
3 * （注：该方法放在最后再看比较好）
4 */
5 final void tryTerminate() {
6 for (; ; ) {
7 int c = ctl.get();
8 /*
9 <1>如果当前线程池是RUNNING状态，就直接返回，因为这时候不需要结束线程池
10 <2>如果当前线程池是TIDYING或TERMINATED状态，也直接返回，这时候就等着
11 修改状态的那个线程把terminated方法执行完毕就行了
12 <3>如果当前线程池是SHUTDOWN状态并且阻塞队列不为空，也直接返回，因为这时
13 候还是要去执行阻塞队列中的任务的，不能改变线程池状态
14 */
15 if (isRunning(c) ||
16 runStateAtLeast(c, TIDYING) ||
17 (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()))
18 return;
19 /*
20 走到这里说明有两种情况，要么当前线程池是STOP状态，要么当前线程池是SHUTDOWN状态并且阻塞队列为空
21 这个时候是否可以结束线程池还要查看一下当前的工作线程数，如果不为0，说明当前线程不是最后一个执行任务
22 的线程（因为如果当前要销毁的线程是空闲状态，会最终在getTask方法中完成-1的动作（执行时抛出异常会
23 在processWorkerExit方法中完成-1），也就是说每个应该要销毁的空闲线程在最后拿取不到任务时都会-1的，
24 所以如果发现当前工作线程数没有减到0的话，就说明当前线程不是最后一个执行线程），那么就不会结束线程池
25 （结束线程池的任务交给最后一个线程来做）。这里ONLY_ONE永远为true，也就是说如果当前线程不是最后一个
26 执行任务的线程的话，那么就只是中断一个空闲的线程而已（相当于中断自己），然后就直接返回就行了
27 */
28 if (workerCountOf(c) != 0) {
29 interruptIdleWorkers(ONLY_ONE);
30 return;
31 }
32
33 /*
34 走到这里说明当前工作线程数已经为0了，也就是说当前线程是最后一个执行任务的线程，
35 此时需要完成结束线程池的动作
36 */
37 final ReentrantLock mainLock = this.mainLock;
38 //上锁
39 mainLock.lock();
40 try {
41 //CAS将ctl状态改为TIDYING
42 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
43 try {
44 //钩子方法，空实现
45 terminated();
46 } finally {
47 //在执行完terminated方法后，将线程池状态置为TERMINATED
48 ctl.set(ctlOf(TERMINATED, 0));
49 /*
50 可能在此之前某线程调用了awaitTermination方法，一直处在阻塞中，
51 并且没有超时，也没有发生中断。那么在结束线程池的此时就需要唤醒这些线程了
52 */
53 termination.signalAll();
54 }
55 return;
56 }
57 } finally {
58 //释放锁
59 mainLock.unlock();
60 }
61 //走到这里说明之前的CAS将状态改为TIDYING失败了，那么就从头开始重试
62 }
63 }
更多内容请关注微信公众号：奇客时间
