摘要：联合ThreadPoolExecutor类的源码深度剖析线程池执行工作的整体流程。
本文分享自华为云社区《【高并发】通过ThreadPoolExecutor类的源码深度解析线程池执行工作的外围流程》，作者： 冰 河。
ThreadPoolExecutor类中存在一个workers工作线程汇合，用户能够向线程池中增加须要执行的工作，workers汇合中的工作线程能够间接执行工作，或者从工作队列中获取工作后执行。ThreadPoolExecutor类中提供了整个线程池从创立到执行工作，再到沦亡的整个流程办法。本文，就联合ThreadPoolExecutor类的源码深度剖析线程池执行工作的整体流程。
在ThreadPoolExecutor类中，线程池的逻辑次要体现在execute(Runnable)办法，addWorker(Runnable, boolean)办法，addWorkerFailed(Worker)办法和回绝策略上，接下来，咱们就深入分析这几个外围办法。
execute(Runnable)办法
execute(Runnable)办法的作用是提交Runnable类型的工作到线程池中。咱们先看下execute(Runnable)办法的源码，如下所示。
public void execute(Runnable command) { //如果提交的工作为空，则抛出空指针异样 if (command == null) throw new NullPointerException(); //获取线程池的状态和线程池中线程的数量 int c = ctl.get(); //线程池中的线程数量小于corePoolSize的值 if (workerCountOf(c) < corePoolSize) { //从新开启线程执行工作 if (addWorker(command, true)) return; c = ctl.get(); } //如果线程池处于RUNNING状态，则将工作增加到阻塞队列中 if (isRunning(c) && workQueue.offer(command)) { //再次获取线程池的状态和线程池中线程的数量，用于二次查看 int recheck = ctl.get(); //如果线程池没有未处于RUNNING状态，从队列中删除工作 if (! isRunning(recheck) && remove(command)) //执行回绝策略 reject(command); //如果线程池为空，则向线程池中增加一个线程 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //工作队列已满，则新增worker线程，如果新增线程失败，则执行回绝策略 else if (!addWorker(command, false)) reject(command);}整个工作的执行流程，咱们能够简化成下图所示。
接下来，咱们拆解execute(Runnable)办法，具体分析execute(Runnable)办法的执行逻辑。
（1）线程池中的线程数是否小于corePoolSize外围线程数，如果小于corePoolSize外围线程数，则向workers工作线程汇合中增加一个外围线程执行工作。代码如下所示。
//线程池中的线程数量小于corePoolSize的值if (workerCountOf(c) < corePoolSize) { //从新开启线程执行工作 if (addWorker(command, true)) return; c = ctl.get();}（2）如果线程池中的线程数量大于corePoolSize外围线程数，则判断以后线程池是否处于RUNNING状态，如果处于RUNNING状态，则增加工作到待执行的工作队列中。留神：这里向工作队列增加工作时，须要判断线程池是否处于RUNNING状态，只有线程池处于RUNNING状态时，能力向工作队列增加新工作。否则，会执行回绝策略。代码如下所示。
if (isRunning(c) && workQueue.offer(command)) （3）向工作队列中增加工作胜利，因为其余线程可能会批改线程池的状态，所以这里须要对线程池进行二次查看，如果以后线程池的状态不再是RUNNING状态，则须要将增加的工作从工作队列中移除，执行后续的回绝策略。如果以后线程池依然处于RUNNING状态，则判断线程池是否为空，如果线程池中不存在任何线程，则新建一个线程增加到线程池中，如下所示。
//再次获取线程池的状态和线程池中线程的数量，用于二次查看int recheck = ctl.get();//如果线程池没有未处于RUNNING状态，从队列中删除工作if (! isRunning(recheck) && remove(command)) //执行回绝策略 reject(command);//如果线程池为空，则向线程池中增加一个线程else if (workerCountOf(recheck) == 0) addWorker(null, false);（4）如果在步骤（3）中向工作队列中增加工作失败，则尝试开启新的线程执行工作。此时，如果线程池中的线程数量曾经大于线程池中的最大线程数maximumPoolSize，则不能再启动新线程。此时，示意线程池中的工作队列已满，并且线程池中的线程已满，须要执行回绝策略，代码如下所示。
//工作队列已满，则新增worker线程，如果新增线程失败，则执行回绝策略else if (!addWorker(command, false)) reject(command);这里，咱们将execute(Runnable)办法拆解，联合流程图来了解线程池中工作的执行流程就比较简单了。能够这么说，execute(Runnable)办法的逻辑基本上就是个别线程池的执行逻辑，了解了execute(Runnable)办法，就根本了解了线程池的执行逻辑。
留神：无关ScheduledThreadPoolExecutor类和ForkJoinPool类执行线程池的逻辑，在【高并发专题】系列文章中的后文中会具体阐明，了解了这些类的执行逻辑，就根本全面把握了线程池的执行流程。
在剖析execute(Runnable)办法的源码时，咱们发现execute(Runnable)办法中多处调用了addWorker(Runnable, boolean)办法，接下来，咱们就一起剖析下addWorker(Runnable, boolean)办法的逻辑。
addWorker(Runnable, boolean)办法
总体上，addWorker(Runnable, boolean)办法能够分为三局部，第一局部是应用CAS平安的向线程池中增加工作线程；第二局部是创立新的工作线程；第三局部则是将工作通过平安的并发形式增加到workers中，并启动工作线程执行工作。
接下来，咱们看下addWorker(Runnable, boolean)办法的源码，如下所示。
private boolean addWorker(Runnable firstTask, boolean core) { //标记重试的标识 retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 查看队列是否在某些特定的条件下为空 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; //上面循环的次要作用为通过CAS形式减少线程的个数 for (;;) { //获取线程池中的线程数量 int wc = workerCountOf(c); //如果线程池中的线程数量超出限度，间接返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //通过CAS形式向线程池新增线程数量 if (compareAndIncrementWorkerCount(c)) //通过CAS形式保障只有一个线程执行胜利，跳出最外层循环 break retry; //从新获取ctl的值 c = ctl.get(); //如果CAS操作失败了，则须要在内循环中从新尝试通过CAS新增线程数量 if (runStateOf(c) != rs) continue retry; } } //跳出最外层for循环，阐明通过CAS新增线程数量胜利 //此时创立新的工作线程 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //将执行的工作封装成worker w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { //独占锁，保障操作workers时的同步 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //此处须要从新查看线程池状态 //起因是在取得锁之前可能其余的线程扭转了线程池的状态 int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) throw new IllegalThreadStateException(); //向worker中增加新工作 workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; //将是否增加了新工作的标识设置为true workerAdded = true; } } finally { //开释独占锁 mainLock.unlock(); } //增加新任胜利，则启动线程执行工作 if (workerAdded) { t.start(); //将工作是否曾经启动的标识设置为true workerStarted = true; } } } finally { //如果工作未启动或启动失败，则调用addWorkerFailed(Worker)办法 if (! workerStarted) addWorkerFailed(w); } //返回是否启动工作的标识 return workerStarted;}乍一看，addWorker(Runnable, boolean)办法还蛮长的，这里，咱们还是将addWorker(Runnable, boolean)办法进行拆解。
（1）查看工作队列是否在某些特定的条件下为空，代码如下所示。
// 查看队列是否在某些特定的条件下为空if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false;（2）在通过步骤（1）的校验后，则进入内层for循环，在内层for循环中通过CAS来减少线程池中的线程数量，如果CAS操作胜利，则间接退出双重for循环。如果CAS操作失败，则查看以后线程池的状态是否产生了变动，如果线程池的状态产生了变动，则通过continue关键字从新通过外层for循环校验工作队列，测验通过再次执行内层for循环的CAS操作。如果线程池的状态没有发生变化，此时上一次CAS操作失败了，则持续尝试CAS操作。代码如下所示。
for (;;) { //获取线程池中的线程数量 int wc = workerCountOf(c); //如果线程池中的线程数量超出限度，间接返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //通过CAS形式向线程池新增线程数量 if (compareAndIncrementWorkerCount(c)) //通过CAS形式保障只有一个线程执行胜利，跳出最外层循环 break retry; //从新获取ctl的值 c = ctl.get(); //如果CAS操作失败了，则须要在内循环中从新尝试通过CAS新增线程数量 if (runStateOf(c) != rs) continue retry;}（3）CAS操作胜利后，示意向线程池中胜利增加了工作线程，此时，还没有线程去执行工作。应用全局的独占锁mainLock来将新增的工作线程Worker对象平安的增加到workers中。
总体逻辑就是：创立新的Worker对象，并获取Worker对象中的执行线程，如果线程不为空，则获取独占锁，获取锁胜利后，再次查看线线程的状态，这是防止在获取独占锁之前其余线程批改了线程池的状态，或者敞开了线程池。如果线程池敞开，则须要开释锁。否则将新减少的线程增加到工作汇合中，开释锁并启动线程执行工作。将是否启动线程的标识设置为true。最初，判断线程是否启动，如果没有启动，则调用addWorkerFailed(Worker)办法。最终返回线程是否起送的标识。
//跳出最外层for循环，阐明通过CAS新增线程数量胜利//此时创立新的工作线程boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try { //将执行的工作封装成worker w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { //独占锁，保障操作workers时的同步 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //此处须要从新查看线程池状态 //起因是在取得锁之前可能其余的线程扭转了线程池的状态 int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) throw new IllegalThreadStateException(); //向worker中增加新工作 workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; //将是否增加了新工作的标识设置为true workerAdded = true; } } finally { //开释独占锁 mainLock.unlock(); } //增加新任胜利，则启动线程执行工作 if (workerAdded) { t.start(); //将工作是否曾经启动的标识设置为true workerStarted = true; } }} finally { //如果工作未启动或启动失败，则调用addWorkerFailed(Worker)办法 if (! workerStarted) addWorkerFailed(w);}//返回是否启动工作的标识return workerStarted;addWorkerFailed(Worker)办法
在addWorker(Runnable, boolean)办法中，如果增加工作线程失败或者工作线程启动失败时，则会调用addWorkerFailed(Worker)办法，上面咱们就来看看addWorkerFailed(Worker)办法的实现，如下所示。
private void addWorkerFailed(Worker w) { //获取独占锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //如果Worker工作不为空 if (w != null) //将工作从workers汇合中移除 workers.remove(w); //通过CAS将工作数量减1 decrementWorkerCount(); tryTerminate(); } finally { //开释锁 mainLock.unlock(); }}addWorkerFailed(Worker)办法的逻辑就比较简单了，获取独占锁，将工作从workers中移除，并且通过CAS将工作的数量减1，最初开释锁。
回绝策略
咱们在剖析execute(Runnable)办法时，线程池会在适当的时候调用reject(Runnable)办法来执行相应的回绝策略，咱们看下reject(Runnable)办法的实现，如下所示。
final void reject(Runnable command) { handler.rejectedExecution(command, this);}通过代码，咱们发现调用的是handler的rejectedExecution办法，handler又是个什么鬼，咱们持续跟进代码，如下所示。
private volatile RejectedExecutionHandler handler;再看看RejectedExecutionHandler是个啥类型，如下所示。
package java.util.concurrent;public interface RejectedExecutionHandler { void rejectedExecution(Runnable r, ThreadPoolExecutor executor);}能够发现RejectedExecutionHandler是个接口，定义了一个rejectedExecution(Runnable, ThreadPoolExecutor)办法。既然RejectedExecutionHandler是个接口，那咱们就看看有哪些类实现了RejectedExecutionHandler接口。
看到这里，咱们发现RejectedExecutionHandler接口的实现类正是线程池默认提供的四种回绝策略的实现类。
至于reject(Runnable)办法中具体会执行哪个类的回绝策略，是依据创立线程池时传递的参数决定的。如果没有传递回绝策略，则默认会执行AbortPolicy类的回绝策略。否则会执行传递的类的回绝策略。
在创立线程池时，除了可能传递JDK默认提供的回绝策略外，还能够传递自定义的回绝策略。如果想应用自定义的回绝策略，则只须要实现RejectedExecutionHandler接口，并重写rejectedExecution(Runnable, ThreadPoolExecutor)办法即可。例如，上面的代码。
public class CustomPolicy implements RejectedExecutionHandler { public CustomPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { System.out.println("应用调用者所在的线程来执行工作") r.run(); } }}应用如下形式创立线程池。
new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), Executors.defaultThreadFactory(), new CustomPolicy());至此，线程池执行工作的整体外围逻辑剖析完结。
点击关注，第一工夫理解华为云陈腐技术~