【JAVA并发编程】ThreadPoolExecutor的源码解读 java

一、简介 ThreadPoolExecutor是Java并发编程中使用最广泛的类之一，由于线程的创建和销毁需要消耗系统资源，所以通过使用线程池来有效管理线程。
二、工作流程当向线程池提交一个任务后，线程池是如何来处理的？

文章图片

如上图：

判断核心线程池是否已满。如果不是，则创建一个新的工作线程来执行任务。如果已满，进入下个步骤
判断工作队列是否已满。如果工作队列没有满，则将新提交的任务存储在工作队列里。如果队列满了，进入下个步骤
判断线程池是否已满。如果没有，则创建一个新的工作线程来执行任务
如果线程池已满，则交给饱和策略来处理这个任务

三、源码解读 3.1 线程池参数
ThreadPoolExecutor的构造函数参数含义：

corePoolSize：核心线程数大小，当线程数
maximumPoolSize：最大线程数，当线程数 >= corePoolSize的时候，会把任务放入工作队列中，当工作队列也满了之后，会再创建新的线程，直到达到最大线程数
keepAliveTime ：保持存活时间，当线程数大于corePoolSize的空闲线程能保持存活的最大时间。
unit：时间单位
workQueue：保存任务的阻塞队列
threadFactory：创建线程的工厂
handler：拒绝策略（包括4种：直接抛出异常 - 默认、使用调用者所在线程来运行任务、丢弃队列里最近一个任务来执行当前任务、不处理直接丢弃）

3.2 线程池运行状态、工作线程数量
ThreadPoolExecutor很巧妙的使用了一个AtomicInteger类型的ctl变量，来表示线程池运行状态和工作线程数量，高3位用来表示线程池运行状态，低29位用于表示工作线程数量。线程池运行状态有5种，状态和其对应的高3位分别是RUNNING(111)、SHUTDOWN(000)、STOP(001)、TIDYING(010)和TERMINATED(011)。runStateOf()和workerCountOf()方法分别可以计算出线程池运行状态和工作线程数量。源码如下：

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; // runState is stored in the high-order bits 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; // Packing and unpacking ctl private static int runStateOf(int c){ return c & ~CAPACITY; } private static int workerCountOf(int c){ return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; }

3.3 工作线程Worker
ThreadPoolExecutor内部有一个工作线程Worker类，它并不是继承Thread类，而是内部包含了一个Thread对象，而且在初始化该Thread对象时，还将当前Worker对象作为构造函数参数传入，形成了一种相互引用的关系。
工作线程Worker类也是设计的很巧妙，这个类继承了AbstractQueuedSynchronizer且实现了Runnable接口，实现Runnable接口是比较容易理解的，它的run()方法就是该工作线程的需要做的工作。而为什么要继承AbstractQueuedSynchronizer呢？这是用于后续判断该工作线程是否已经在执行中，如果正在执行中，那么该工作线程就不能被中断。代码如下

private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** Thread this worker is running in.Null if factory fails. */ final Thread thread; /** Initial task to run.Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); }/** Delegates main run loop to outer runWorker*/ public void run() { runWorker(this); } // 下面是一些实现AQS需要实现的方法，不一一展开了 }

3.4 提交任务
我们通常使用execute()或submit()来提交任务，那么submit()内部也是使用execute()方法来执行的，所以这里我们看下execute()方法的源码：

public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * 分为3个步骤 *1. 如果当前线程数小于核心线程池大小，会调用addWorker方法，addWorker方法会根据运行状态和线程数量来判断是否需要新增工作线程 *2. 如果当前线程池运行状态是RUNNING，尝试添加任务到任务队列workQueue中 *3. 如果无法添加到任务队列中，继续调用addWorker方法，判断是否新增工作线程 *4. 最后如果还是无法新增工作线程，则采取拒绝策略 */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }

3.5 新增工作线程
通过上面的代码，很明显我们接下来就需要看下addWorker()是如何根据运行状态和线程数量来判断是否需要新增工作线程，并且这个方法也是ThreadPoolExecutor中最重要的方法。首先根据运行状态和线程数量来判断是否新增工作线程，如果判断不新增，直接返回false，否则新增工作线程。源码如下：

private boolean addWorker(Runnable firstTask, boolean core) { retry: for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); /* * 1. 如果运行状态是STOP、TIDYING和TERMINATED，那么直接返回false，新增工作线程失败 * 2. 如果运行状态是RUNNING，直接进行下一步 * 3. 如果运行状态是SHUTDOWN，当firstTask不为空（代表新来的任务），也直接返回false，新增工作线程失败。 *firstTask为空的话，如果任务队列为空，也不新增工作线程，反之新增工作线程。 */ if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; /* * 这里主要是判断工作线程数量是否大于核心线程数（core为true）或大于最大线程数（core为false），如果大于，则添加工作线程失败 * 否则将ctl的值加1，也就是将工作线程数量加1，并且跳出两层循环 */ for (; ; ) { int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } }boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try {// new一个工作线程 w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 获取线程池的锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int rs = runStateOf(ctl.get()); // 再次检查线程池运行状态 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); // 如果当前工作线程数大于工作线程数的历史最大值，则更新历史最大值 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 启动工作线程 t.start(); workerStarted = true; } } } finally { // 如果新增工作线程失败，将ctl的值减1，也就是将工作线程数量减1 if (! workerStarted) addWorkerFailed(w); } return workerStarted; }

3.6 工作线程的工作流程
工作线程Worker的run()方法中的runWroker(this)就是其工作流程，如果firstTask不为空，直接执行firstTask，否则从任务队列中获取任务（如果需要回收工作线程，则使用超时获取任务机制，一旦获取任务超时，则回收工作线程，如果不需要回收工作线程，工作线程将会一直阻塞，直到获取任务），代码如下

final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 如果firstTask不为空，直接执行，否则从任务队列中获取任务（超时获取或阻塞获取） while (task != null || (task = getTask()) != null) { // 获取工作线程的锁，代表工作线程不空闲 w.lock(); // 如果运行状态大于等于STOP，中断当前工作线程 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { // 钩子函数，可以继承ThreadPoolExecutor，并重写该方法，通常用来监控任务执行 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 { // 钩子函数，同beforeExecute() afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } // 任务是否意外终止 completedAbruptly = false; } finally { // 退出工作线程 processWorkerExit(w, completedAbruptly); } }

退出工作线程可分为正常退出或任务异常导致，当completedAbruptly为true时，代表是任务异常导致退出工作线程，completedAbruptly为false，代表任务队列为空，且正常回收线程（当allowCoreThreadTimeOut为false时，线程数量大于核心线程数，或当allowCoreThreadTimeOut为true时，线程数量大于0），代码如下：

private void processWorkerExit(Worker w, boolean completedAbruptly) { // 如果任务异常导致工作线程退出，工作线程数减1，如果工作线程正常退出时，在getTask()方法中已经将工作线程数减1 if (completedAbruptly) decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; // 移除工作线程 workers.remove(w); } finally { mainLock.unlock(); }tryTerminate(); int c = ctl.get(); // 如果运行状态小于STOP，判断是否需要新增工作线程来处理 if (runStateLessThan(c, STOP)) { // 如果工作线程正常退出 if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; // 如果工作线程已经大于等于min值，那么不需要执行下面的addWork()，直接返回 if (workerCountOf(c) >= min) return; // replacement not needed } // 执行addWorker()来新增工作线程 addWorker(null, false); } }

3.7 关闭线程池
我们通常会使用shutdown()或shutdownNow()方法来关闭线程池。

shutdown是将线程池的状态设置为SHUTDOWN状态，不接受新的任务，正在执行的任务和队列中的任务会继续执行完，如果任务队列为空，则中断并退出空闲的工作线程。
shutdownNow将线程池的状态设置成STOP，不接受新的任务，且尝试中断所有正在执行的任务，不执行并返回任务队列中的任务列表。
shutdown()和shutdownNow()方法类似，这里只看shutdown()方法，代码如下：

public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); // 更新运行状态为SHUTDOWN advanceRunState(SHUTDOWN); // 中断空闲的工作线程，因为任务队列为空，所以该工作线程会退出 interruptIdleWorkers(); // 钩子函数 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } /** * 尝试终止线程池 * 当运行状态为SHUTDOWN且任务队列为空且工作线程数为0，或者运行状态为STOP且工作线程数为0 * 代表线程池关闭，执行terminated函数，这也是个钩子函数 */ tryTerminate(); }

四、总结 【【JAVA并发编程】ThreadPoolExecutor的源码解读】以上解读的就是ThreadPoolExecutor中最核心的代码，包括提交任务、新增工作线程、工作线程的工作流程以及线程池的关闭等，理解了这些代码，我们就可以知道ThreadPoolExecutor线程池具体是如何工作的，也能在工作中更加得心应手地去使用它。文中若有错误，欢迎指正。