一文搞懂ThreadPoolExecutor原理
前言
都说经典的就是好的，这句话放在Java的ThreadPoolExecutor上那是一点都没错，像现在数据库连接的池化实现，或者像Tomcat这种WEB服务器的线程管理，处处都有着ThreadPoolExecutor的影子，本篇文章将结合源码实现，对ThreadPoolExecutor的原理进行一个深入学习。
正文
一. Executor框架简介
Executor框架提供了组件来管理Java中的线程，Executor框架将其分为任务，线程执行任务，任务执行结果三部分。下面以表格形式对这三部分进行说明。
项说明任务Executor框架提供了Runnable接口和Callable接口，任务需要实现这两个接口才能被线程执行线程执行任务Executor框架提供了接口Executor和继承于Executor的ExecutorService接口来定义任务执行机制。Executor框架中的线程池类ThreadPoolExecutor和ScheduledThreadPoolExecutor均实现了ExecutorService接口任务执行结果Executor框架提供了Future接口和实现了Future接口的FutureTask类来定义任务执行结果。
组件之间的类图关系如下所示。
Executor接口是线程池的顶层接口，通常说到的线程池指的是ThreadPoolExecutor，同时ThreadPoolExecutor还有一个子类叫做ScheduledThreadPoolExecutor，其扩展实现了延时执行任务和定时执行任务的功能。
Executor框架指的是任务，执行任务的线程池和任务执行结果这三部分，切不可将Executor框架与Executor接口相混淆。
本篇文章就将对Executor框架中的ThreadPoolExecutor的源码实现进行学习。
二. 认识ThreadPoolExecutor状态
在学习ThreadPoolExecutor如何执行任务前，先认识一下ThreadPoolExecutor的状态。
ThreadPoolExecutor继承于AbstractExecutorService，并实现了ExecutorService接口，是Executor框架的核心类，用于管理线程。
ThreadPoolExecutor使用了原子整型ctl来表示线程池状态和Worker数量。ctl是一个原子整型，前3位表示线程池状态，后29位表示Worker数量。ThreadPoolExecutor中这部分的源码如下所示。
java复制代码public class ThreadPoolExecutor extends AbstractExecutorService { private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY = (1 << COUNT_BITS) - 1; 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; // 取整型前3位，即获取线程池状态 private static int runStateOf(int c) { return c & ~CAPACITY; } // 取整型后29位，即获取Worker数量 private static int workerCountOf(int c) { return c & CAPACITY; } // 根据线程池状态和Worker数量拼装ctl private static int ctlOf(int rs, int wc) { return rs | wc; } // 线程池状态判断 private static boolean runStateLessThan(int c, int s) { return c < s; } // 线程池状态判断 private static boolean runStateAtLeast(int c, int s) { return c >= s; } // 判断线程池状态是否为RUNNING private static boolean isRunning(int c) { return c < SHUTDOWN; } ...... }
可知ThreadPoolExecutor有如下五种线程池状态。
RUNNING，线程池接受新任务，会执行任务阻塞队列中的任务，ctl前三位表示为111；
SHUTDOWN，线程池拒绝新任务，会执行任务阻塞队列中的任务，ctl前三位表示为000；
STOP，线程池拒绝新任务，不会执行任务阻塞队列中的任务，尝试中断正在执行的任务，ctl前三位表示为001；
TIDYING，所有任务被关闭，Worker数量为0，ctl前三位表示为010；
TERMINATED，terminated() 执行完毕，ctl前三位表示为011。
得益于ctl的结构，所以无论Worker数量是多少，ThreadPoolExecutor中线程池状态存在如下关系。
RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
因此runStateLessThan()，runStateAtLeast() 和isRunning() 方法可以方便的对线程池状态进行判断。
三. 执行任务源码分析
作为线程池，ThreadPoolExecutor最重要也最经典的地方，当然就是执行任务了。本节对ThreadPoolExecutor执行任务的流程进行一个学习。
ThreadPoolExecutor中执行任务的入口方法为execute()，其实现如下。
java复制代码public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 如果Worker数量小于核心线程数，则创建Worker并执行任务 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 如果Worker数量大于等于核心线程数，则将任务添加到任务阻塞队列 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 如果线程池状态突然不再是RUNNING，则尝试将任务从任务阻塞队列中删除，删除成功则为该任务执行拒绝策略 if (! isRunning(recheck) && remove(command)) reject(command); // 如果线程池中Worker数量突然为0，则创建一个Worker来执行任务 else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 执行到这里表示线程池状态已经不是RUNNING或者任务阻塞队列已满 // 此时尝试新建一个Worker来执行任务 // 如果新建一个Worker来执行任务失败，表明线程池状态不再是RUNNING或者Worker数量已经达到最大线程数，此时执行拒绝策略 else if (!addWorker(command, false)) reject(command); }
在execute() 中会根据Worker数量和线程池状态来决定是新建Worker来执行任务还是将任务添加到任务阻塞队列。新建Worker来执行任务的实现如下所示。
java复制代码private boolean addWorker(Runnable firstTask, boolean core) { // 标记外层for循环 retry: for (;;) { int c = ctl.get(); // 获取线程池状态 int rs = runStateOf(c); // 线程池状态为RUNNING时，可以创建Worker // 线程池状态为SHUTDOWN，且任务阻塞队列不为空时，可以创建初始任务为null的Worker if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { // 获取Worker数量 int wc = workerCountOf(c); // 如果Worker数量大于CAPACITY，拒绝创建Worker // core为true表示创建核心线程Worker，如果Worker数量此时已经大于等于核心线程数，则拒绝创建Worker，转而应该将任务添加到任务阻塞队列 // core为false表示创建非核心线程Worker，如果Worker数量此时已经大于等于最大线程数，则拒绝创建Worker，转而应该执行拒绝策略 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 以CAS方式将Worker数量加1 // 加1成功表明无竞争发生，从外层for循环跳出 if (compareAndIncrementWorkerCount(c)) break retry; // 加1失败表明有竞争发生，此时需要重新获取ctl的值 c = ctl.get(); // 重新获取ctl后如果发现线程池状态发生了改变，此时重新执行外层for循环，即需要基于新的线程池状态判断是否允许创建Worker // 重新获取ctl后如果线程池状态未发生改变，则继续执行内层for循环，即尝试再一次以CAS方式将Worker数量加1 if (runStateOf(c) != rs) continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 创建一个Worker w = new Worker(firstTask); // 获取Worker的线程 final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; // 由于线程池中存储Worker的集合为HashSet，因此将Worker添加到Worker集合时需要获取全局锁保证线程安全 mainLock.lock(); try { // 再一次获取线程池状态 int rs = runStateOf(ctl.get()); // 如果线程池状态还是为RUNNING或者线程池状态为SHUTDOWN但创建的Worker的初始任务为null，则允许将创建出来的Worker添加到集合 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // 检查一下Worker的线程是否可以启动（处于活动状态的线程无法再启动） if (t.isAlive()) throw new IllegalThreadStateException(); // 将Worker添加到Worker集合 workers.add(w); int s = workers.size(); // largestPoolSize用于记录线程池最多存在过的Worker数 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 启动Worker线程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) // Worker线程没有成功启动起来，此时需要对该Worker的创建执行回滚操作 addWorkerFailed(w); } return workerStarted; }
addWorker() 方法中只允许两种情况可以创建Worker。
线程池状态为RUNNING，可以创建Worker；
线程池状态为SHUTDOWN，且任务阻塞队列不为空，可以创建初始任务为null的Worker。
一旦Worker创建成功，就会将Worker的线程启动，如果Worker创建失败或者Worker的线程启动失败，则会调用addWorkerFailed() 方法执行回滚操作，其实现如下所示。
java复制代码private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 如果Worker添加到了Worker集合中，则将Worker从Worker集合中删除 if (w != null) workers.remove(w); // 以CAS方式将Worker数量减1 decrementWorkerCount(); // 尝试终止线程池 tryTerminate(); } finally { mainLock.unlock(); } }
由于Worker自身实现了Runnable，因此Worker自身就是一个任务，实际上Worker的线程执行的任务就是Worker本身，因此addWorker() 中将Worker的线程启动时，会调用Worker的run() 方法，其实现如下。
java复制代码public void run() { runWorker(this); }
在Worker的run() 方法中调用了ThreadPoolExecutor的runWorker() 方法，其实现如下所示。
java复制代码final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); boolean completedAbruptly = true; try { // 如果task为null，则从任务阻塞队列中获取任务 // 通常Worker启动时会先执行初始任务，然后再去任务阻塞队列中获取任务 while (task != null || (task = getTask()) != null) { w.lock(); // 线程池正在停止时，需要确保当前Worker的线程是被中断的 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; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { // Worker执行任务发生异常或者从getTask()中获取任务为空时会执行这里的逻辑 // processWorkerExit()会将Worker从Worker集合中删除，并尝试终止线程池 processWorkerExit(w, completedAbruptly); } }
runWorker() 方法就是先让Worker将初始任务（如果有的话）执行完，然后循环从任务阻塞队列中获取任务来执行，如果Worker执行任务发生异常或者从任务阻塞队列获取任务失败（获取到的任务为null），则调用processWorkerExit() 方法来将自身从Worker集合中删除。下面先看一下getTask() 方法的实现。
java复制代码private Runnable getTask() { boolean timedOut = false; for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 如果线程池状态为SHUTDOWN，且任务阻塞队列为空，则不再允许从任务阻塞队列中获取任务 // 如果线程池状态为STOP，则不再允许从任务阻塞队列中获取任务 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // 如果allowCoreThreadTimeOut为true，或者当前线程数大于核心线程数，此时timed为true，表明从任务阻塞队列以超时退出的方式获取任务 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 如果当前线程数大于最大线程数，则当前Worker应该被删除 // 如果当前Worker上一次从任务阻塞队列中获取任务时超时退出，且任务阻塞队列现在还是为空，则当前Worker应该被删除 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表明Worker上一次从任务阻塞队列中获取任务时超时退出 timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
getTask() 方法在如下情况不允许Worker从任务阻塞队列中获取任务。
线程池状态为SHUTDOWN，且任务阻塞队列为空；
线程池状态为STOP。
如果Worker有资格从任务阻塞队列获取任务，那么当allowCoreThreadTimeOut为true，或者当前线程数大于核心线程数时，Worker以超时退出的方式获取任务，否则Worker以一直阻塞的方式获取任务。
当Worker在getTask() 方法中获取任务失败时，getTask() 方法会返回null，从而导致Worker会执行processWorkerExit() 方法来删除自身，其实现如下所示。
java复制代码private void processWorkerExit(Worker w, boolean completedAbruptly) { // completedAbruptly为true表明是执行任务时发生异常导致Worker需要被删除 if (completedAbruptly) // 修正Worker数量 decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; // 将Worker从Worker集合中删除 workers.remove(w); } finally { mainLock.unlock(); } // 尝试终止线程池 tryTerminate(); int c = ctl.get(); if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; if (workerCountOf(c) >= min) return; } addWorker(null, false); } }
Worker在processWorkerExit() 方法中删除自身之后，还会调用tryTerminate() 尝试终止线程池，tryTerminate() 方法很精髓，后面会对其进行详细分析，这里暂且不谈。至此，Worker的创建，执行任务，获取任务和删除的整个流程已经大体分析完毕。
对于执行任务，现在简单进行一个小结。
ThreadPoolExecutor执行任务，第一步是根据Worker数量来决定是新建Worker来执行任务还是将任务添加到任务阻塞队列，这里的判断规则如下。
如果Worker数量小于核心线程数，则创建Worker来执行任务；
如果Worker数量大于等于核心线程数，则将任务添加到任务阻塞队列；
如果任务阻塞队列已满，则创建Worker来执行任务；
如果Worker数量已经达到最大线程数，此时执行任务拒绝策略。
当要新建Worker来执行任务时，只有两种情况可以新建Worker，如下所示。
线程池状态为RUNNING，可以创建Worker；
线程池状态为SHUTDOWN，且任务阻塞队列不为空，可以创建初始任务为null的Worker。
Worker自身实现了Runnable，且Worker持有一个线程，当Worker启动时，就是启动Worker持有的线程，而这个线程执行的任务就是Worker自身。
Worker启动后，会首先执行自己的初始任务，然后再去任务阻塞队列中获取任务。
四. 关闭线程池源码分析
不再使用的线程池，可以进行关闭。关闭ThreadPoolExecutor的方法有shutdown() 和shutdownNow()，本节将对ThreadPoolExecutor的关闭进行分析。
1. shutdown()
首先分析shutdown() 方法，其实现如下。
java复制代码public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); // 循环通过CAS方式将线程池状态置为SHUTDOWN advanceRunState(SHUTDOWN); // 中断空闲Worker interruptIdleWorkers(); onShutdown(); } finally { mainLock.unlock(); } // 尝试终止线程池 tryTerminate(); }
在shutdown() 方法中首先会将线程池状态置为SHUTDOWN，然后调用interruptIdleWorkers() 方法中断空闲Worker，最后调用tryTerminate() 方法来尝试终止线程池。那么这里要解释一下什么是空闲Worker，先看一下interruptIdleWorkers() 的实现。
java复制代码private void interruptIdleWorkers() { interruptIdleWorkers(false); } private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; // 中断线程前需要先尝试获取Worker的锁 // 只能获取到空闲Worker的锁，所以shutdown()方法只会中断空闲Worker if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } }
调用interruptIdleWorkers() 方法中断Worker前首先需要尝试获取Worker的锁，已知Worker除了实现Runnable接口外，还继承于AbstractQueuedSynchronizer，因此Worker本身是一把锁，然后在runWorker() 中Worker执行任务前都会先获取Worker的锁，这里看一下Worker的lock() 方法的实现。
java复制代码public void lock() { acquire(1); } protected boolean tryAcquire(int unused) { // 以CAS方式将state从0设置为1 if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; }
可以发现，Worker在lock() 中调用了acquire() 方法，该方法由AbstractQueuedSynchronizer抽象类提供，在acquire() 中会调用其子类实现的tryAcquire() 方法，tryAcquire() 方法会以CAS方式将state从0设置为1，因此这样的设计让Worker是一把不可重入锁。
回到interruptIdleWorkers() 方法，前面提到该方法中断Worker前会尝试获取Worker的锁，能够获取到锁才会中断Worker，而因为Worker是不可重入锁，所以正在执行任务的Worker是无法获取到锁的，只有那些没有执行任务的Worker的锁才能够被获取，因此所谓的中断空闲Worker，实际就是中断没有执行任务的Worker，那些执行任务的Worker在shutdown() 方法被调用时不会被中断，这些Worker执行完任务后会继续从任务阻塞队列中获取任务来执行，直到任务阻塞队列为空，此时没有被中断过的Worker也会被删除掉，等到线程池中没有Worker以及任务阻塞队列没有任务后，线程池才会被终止掉。
对于shutdown() 方法，一句话总结就是：将线程池状态置为SHUTDOWN并拒绝接受新任务，等到线程池Worker数量为0，任务阻塞队列为空时，关闭线程池。
2. shutdownNow()
现在再来分析shutdownNow() 方法。
java复制代码public List<Runnable> shutdownNow() { List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); // 循环通过CAS方式将线程池状态置为STOP advanceRunState(STOP); // 中断所有Worker interruptWorkers(); // 将任务阻塞队列中的任务获取出来并返回 tasks = drainQueue(); } finally { mainLock.unlock(); } // 尝试终止线程池 tryTerminate(); return tasks; } private void interruptWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 中断线程池中所有Worker for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); } }
在shutdownNow() 方法中首先会将线程池状态置为STOP，然后调用interruptWorkers() 方法中断线程池中的所有Worker，接着调用tryTerminate() 方法来尝试终止线程池，最后shutdownNow() 方法会将任务阻塞队列中还未被执行的任务返回。
shutdownNow() 方法调用之后，线程池中的所有Worker都会被中断，包括正在执行任务的Worker，等到所有Worker都被删除之后，线程池即被终止，也就是说，shutdownNow() 不会保证当前时刻正在执行的任务会被安全的执行完，并且会放弃执行任务阻塞队列中的所有任务。
3. tryTerminate()
关于线程池的关闭，还有一个重要的方法，那就是前面多次提到的tryTerminate() 方法，该方法能确保线程池可以被正确的关闭，其实现如下所示。
java复制代码final void tryTerminate() { for (;;) { int c = ctl.get(); // 如果线程池状态为RUNNING，则没有资格终止线程池 // 如果线程池状态大于等于TIDYING，则没有资格终止线程池 // 如果线程池状态为SHUTDOWN但任务阻塞队列不为空，则没有资格终止线程池 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; // 线程池状态为SHUTDOWN且任务阻塞队列为空会执行到这里 // 线程池状态为STOP会执行到这里 // Worker数量不为0，表明当前还有正在执行任务的Worker或者空闲的Worker，此时中断一个空闲的Worker // 在这里被中断的空闲Worker会在getTask()方法中返回null，从而执行processWorkerExit()，最终该Worker会被删除 // processWorkerExit()方法中又会调用tryTerminate()，因此将shutdown信号在空闲Worker之间进行了传播 if (workerCountOf(c) != 0) { interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 将线程池状态置为TIDYING if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { // 终止线程池 terminated(); } finally { // 将线程池状态最终置为TERMINATED ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } } }
在tryTerminate() 方法的官方注释中给出了两种线程池会被终止的情况：
线程池的状态为SHUTDOWN，Worker数量为0，任务阻塞队列为空；
线程池的状态为STOP，Worker数量为0。
官方注释中还说明在所有可能导致线程池终止的操作中都应该调用tryTerminate() 方法来尝试终止线程池，因此线程池中Worker被删除时和任务阻塞队列中任务被删除时会调用tryTerminate()，以达到在线程池符合终止条件时及时终止线程池。
4. 小结
对于关闭线程池，简单小结如下。
关闭ThreadPoolExecutor有两种方式，如下所示。
shutdown()。调用shutdown() 方法会首先将线程池状态置为SHUTDOWN并拒绝接受新任务，然后中断空闲Worker，等到线程池中Worker数量为0，任务阻塞队列为空时，线程池被真正关闭；
shutdownNow()。调用shutdownNow() 方法会首先将线程池状态置为STOP，然后中断所有Worker（包括正在执行任务的Worker），并将任务阻塞队列中还未被执行的任务返回，当线程池Worker数量为0时，线程池被真正关闭。
还有一点需要说明，Worker除了实现Runnable接口外，还继承于AbstractQueuedSynchronizer，因此Worker本身是一把锁，Worker执行任务前都会先获取Worker的锁，所以正在执行任务的Worker的锁是无法被获取的，换言之，只有没有执行任务的Worker的锁才能被获取，这些Worker就称为空闲Worker。
