1.写在前面

本文整理自组内的技术分享，主要参考了以下文档。

JDK版本1.8.0_202
美团点评的文章《Java线程池实现原理及其在美团业务中的实践》https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html
我们在代码中增加了详细的注释供大家理解，有不正确的地方欢迎大家多多指正~

2. 线程池是什么

线程池（Thread Pool）是一种基于池化思想管理线程的工具，经常出现在多线程服务器中，如MySQL。线程池解决的核心问题就是资源管理问题。在并发环境下，系统不能确定在任意时刻有多少任务需要执行，有多少资源需要投入。线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。本文中我们描述线程池是JDK中提供的ThreadPoolExecutor类。

当然，使用线程池可以带来一系列好处：

降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
提高响应速度：任务到达时，无需等待线程创建即可立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。

这里也给出查看进程中线程的数量的几种方法：

cat /proc/$pid/status | grep Threads
pstree -p $pid
top -Hp $pid （该命令可以看到占用cpu最高的线程，然后将pid转成16进制在jstack中搜索，例如pid是209，转成16进制是0xd1，在jstack中搜nid=0xd1即可找到占用cpu最高线程的调用栈）

3. JDK中线程池的实现（ThreadPoolExecutor）

3.1 总体设计

Java中的线程池核心实现类是ThreadPoolExecutor，我们基于JDK 1.8的源码来分析Java线程池的核心设计与实现。ThreadPoolExecutor的UML类图
如图1所示。

Executor接口：ThreadPoolExecutor实现的顶层接口是Executor，顶层接口Executor提供了一种思想：将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，用户只需提供Runnable对象，将任务的运行逻辑提交到执行器(Executor)中，由Executor框架完成线程的调配和任务的执行部分。
ExecutorService接口：接口增加了一些能力：（1）扩充执行任务的能力，补充可以为一个或一批异步任务生成Future的方法；（2）提供了管控线程池的方法，比如停止线程池的运行。

void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task)
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException;

抽象类AbstractExecutorService：将执行任务的流程串联了起来，保证下层的实现只需关注一个执行任务的方法即可。线程池部分功能的基本实现，例如submit方法，invokeAll/invokeAny等等，例如submit方法底层也是依赖Executor#execute() 方法实现的执行，其内部包装了一个FutureTask来获取执行结果。从这里也可以看出，AbstractExecutorService中实现的这几个方法是一个模板方法，或者理解为标准，其他的线程池想使用这些方法只需要实现Executor#execute() 方法即可。

public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;
}

实现类ThreadPoolExecutor：最基础的线程池实现版本，我们常用的线程池的核心实现就在这个类中。实现最复杂的运行部分，ThreadPoolExecutor将会一方面维护自身的生命周期，另一方面同时管理线程和任务，使两者良好的结合从而执行并行任务。ThreadPoolExecutor是如何运行，如何同时维护线程和执行任务的呢？

3.2 线程池的构造方法与基础属性

构造方法

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;
}

corePoolSize: 核心线程数
maximumPoolSize: 最大线程数
keepAliveTime: 表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0
unit：上个参数的时间单位
workerQueue：任务队列
threadFactory：线程工厂，可以定义线程创建的方法
handler：拒绝策略，可以决定如果线程池溢出了这个任务要怎么做

状态管理

线程池运行的状态，并不是用户显式设置的，而是伴随着线程池的运行，由内部来维护。线程池内部使用一个变量ctl维护两个值：运行状态(runState)和线程数量 (workerCount)。在具体实现中，线程池将运行状态(runState)、线程数量 (workerCount)两个关键参数的维护放在int中，高3位保存runState，低29位保存workerCount，两个变量之间互不干扰。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; // 00000000000000000000000000011101
private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // 00011111111111111111111111111111// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS; // 11100000000000000000000000000000 , -1 * 2^29 = -536870912
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 00000000000000000000000000000000 , 0 * 2 ^ 29 = 0
private static final int STOP       =  1 << COUNT_BITS; // 00100000000000000000000000000000 , 1 * 2 ^ 29 = 536870912
private static final int TIDYING    =  2 << COUNT_BITS; // 01000000000000000000000000000000 , 2 * 2 ^ 29 = 1073741824
private static final int TERMINATED =  3 << COUNT_BITS; // 01100000000000000000000000000000 , 3 * 2 ^ 29 = 1610612736
private static int runStateOf(int c)     { return c & ~CAPACITY; } // 得到高3位，即线程池状态
private static int workerCountOf(int c)  { return c & CAPACITY; } // 得到低29位，即当前线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; } // 通过状态和线程数生成ctl

为什么这样设计？

ctl这个变量使用AtomicInteger实现能够保证数值在各个线程的可见性，也就能保证线程状态和数量在线程池全局可见性。
用一个变量去存储两个值，可避免在做相关决策时，出现不一致的情况，不必为了维护两者的一致，而占用锁资源。
位运算的速度比基本运算会快很多。
状态设计的参考

启示

设计状态时，状态编号的大小也可以表示含义，例如线程池的runState>=0时表示线程池正在关闭。Kwaiflow的Ti以及DR状态也可以参考这样的设计。采用这样的设计后，我们也可以用比较运算来判断ti或者dr是否处于终态。

ThreadPoolExecutor的运行状态有5种，分别为：

线程池生命周期转换如图3所示：

3.3 任务执行机制

任务调度

任务调度流程如图3所示。

首先检测线程池运行状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING的状态下执行任务。
如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。
如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task.  The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread.  If it fails, we know we are shut down or saturated* and so reject the task.*/int c = ctl.get();// 当前线程数量小于corePoolSize，尝试新增线程if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// offer是非阻塞的，队列满直接返回falseif (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);}// 线程池状态不是RUNNING或者向队列中添加任务失败，则尝试增加线程（例如corePoolSize是3，队列长度是5，maxPoolSize是10。当队列满了之后，会走这里）else if (!addWorker(command, false))// 线程池状态不是RUNNING或者线程数大于等于最大线程数(即CAPACITY，536870912)reject(command);
}

任务缓冲

线程池中是以生产者消费者模式，通过一个阻塞队列来实现的。阻塞队列缓存任务，工作线程从阻塞队列中获取任务。

使用不同的队列可以实现不一样的任务存取策略。在这里，我们再介绍下阻塞队列的成员：

ThreadPoolExecutor中使用了BlockingQueue的offer、remove、poll、take方法。可以看到，ThreadPoolExecutor向队列中放元素的时候并没有使用阻塞方法，从队列取元素的时候用到了阻塞和非阻塞的方法，在下面解析getTask方法时我们会再进行详细的说明。

public interface BlockingQueue<E> extends Queue<E> {// 队列满时抛异常boolean add(E e);// 队列满时返回falseboolean offer(E e);// 队列满时阻塞void put(E e) throws InterruptedException;// 队列空时阻塞E take() throws InterruptedException;// 等待超时返回nullE poll(long timeout, TimeUnit unit)throws InterruptedException;// 不包含o则返回falseboolean remove(Object o);
}

任务申请

任务的执行有两种可能：一种是任务直接由新创建的线程执行。另一种是线程从任务队列中获取任务然后执行，执行完任务的空闲线程会再次去从队列中申请任务再去执行。第一种情况仅出现在线程初始创建的时候，第二种是线程获取任务绝大多数的情况。
线程需要从任务缓存模块中不断地取任务执行，帮助线程从阻塞队列中获取任务，实现线程管理模块和任务管理模块之间的通信。这部分策略由getTask方法实现，流程如图5所示。getTask这部分进行了多次判断，为的是控制线程的数量，使其符合线程池的状态。如果线程池现在不应该持有那么多线程，则会返回null值。工作线程Worker会不断接收新任务去执行，而当工作线程Worker接收不到任务的时候，就会开始被回收。

// 从队列获取任务，总体是先判断线程池状态和线程数量，然后再从队列获取任务
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out? 上一次获取是否超时了?for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.// rs >=STOP 或者 rs == SHUTDOWN并且队列是空，返回nullif (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling? 是否允许worker线程被回收？如果允许核心线程超时或者线程数超过corePoolSize则允许boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 当前线程数大于maximumPoolSize或者允许线程回收并且上次获取超时了，并且当前线程数大于1或者队列为空if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}// 注意从队列获取任务这段代码是catch了InterruptedException的（BlockQueue的poll、take中都使用了lock.lockInterruptibly），在线程池停止时，会中断worker，// Worker设置interrupt之后从队列取数据的行为会终止，再回到循环开始如果判断线程池状态终止了，则返回null，随即Worker退出。try {// worker允许被停掉，则从队列里获取任务时最多等待keepAliveTime；否则，获取不到任务就一直阻塞Runnable r = timed ?// 超时后poll会返回nullworkQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :// 只有allowCoreThreadTimeOut是false并且 wc <= corePoolSize，timed才是false，也就是不允许核心线程被回收的前提下，核心线程会走take// 这也就是一般情况下，我们不允许核心线程超时，如果不关闭线程池会发现main方法一直不退出的原因workQueue.take();if (r != null)return r;// r == null才会走到这里，也就是没走上面的workQueue.take()和return r，表示本轮获取任务超时了，因此将timeOut置为truetimedOut = true;} catch (InterruptedException retry) {// workerQueue.poll 或者 workerQueue.take被中断，线程池停止时会发出中断timedOut = false;}}}// ArrayBlockingQueue的poll方法，使用lock.lockInterruptibly()加锁，能够响应中断
public E poll(long timeout, TimeUnit unit) throws InterruptedException {long nanos = unit.toNanos(timeout);// It's a coding style made popular by Doug Lea.// It's an extreme optimization that probably isn't necessary;// you can expect the JIT to make the same optimizations.// (you can try to check the machine code yourself!)// Nevertheless, copying to locals produces the smallest// bytecode, and for low-level code it's nice to write code// that's a little closer to the machine.// Doug Lea的极致优化final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0) {if (nanos <= 0)return null;nanos = notEmpty.awaitNanos(nanos);}return dequeue();} finally {lock.unlock();}}

任务拒绝

任务拒绝模块是线程池的保护部分，线程池有一个最大的容量，当线程池的任务缓存队列已满，并且线程池中的线程数目达到maximumPoolSize时，就需要拒绝掉该任务，采取任务拒绝策略，保护线程池。拒绝策略是一个接口，其设计如下：

public interface RejectedExecutionHandler {void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

用户可以通过实现这个接口去定制拒绝策略，也可以选择JDK提供的四种已有拒绝策略，其特点如下：

3.4 Worker线程管理

Worker线程

Worker实现了Runnable接口，并持有一个线程thread，一个初始化的任务firstTask。

thread是在调用构造方法时通过ThreadFactory来创建的线程，thread持有的任务就是this。
firstTask用它来保存传入的第一个任务，这个任务可以有也可以为null。如果这个值是非空的，那么线程就会在启动初期立即执行这个任务，也就对应核心线程创建或者队列满了添加非核心线程时的情况；如果这个值是null，那么就需要创建一个线程去执行任务列表（workQueue）中的任务，也就是非核心线程的创建。

Worker执行任务的模型如图6所示：

Worker是通过继承AQS，使用AQS来实现独占锁这个功能。没有使用可重入锁ReentrantLock，而是使用AQS，为的就是实现不可重入的特性去反应线程现在的执行状态。

lock方法一旦获取了独占锁，表示当前线程正在执行任务中。
如果正在执行任务，则不应该中断线程。
如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可以对该线程进行中断。
线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来中断空闲的线程，interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态；如果线程是空闲状态则可以安全回收。

// 通过继承AQS实现不可重入的独占锁
private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/*** This class will never be serialized, but we provide a* serialVersionUID to suppress a javac warning.*/private static final long serialVersionUID = 6138294804551838833L;/** 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;/*** Creates with given first task and thread from ThreadFactory.* @param firstTask the first task (null if none)*/Worker(Runnable firstTask) {// state表示worker的状态. -1 是初始状态(禁止中断直到runWorker) 0表示未锁状态 1表示锁住状态setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker  */public void run() {runWorker(this);}// Lock methods//// The value 0 represents the unlocked state.// The value 1 represents the locked state.protected boolean isHeldExclusively() {return getState() != 0;}protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected boolean tryRelease(int unused) {setExclusiveOwnerThread(null);setState(0);return true;}public void lock()        { acquire(1); }public boolean tryLock()  { return tryAcquire(1); }public void unlock()      { release(1); }public boolean isLocked() { return isHeldExclusively(); }void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}}

Worker线程增加

addWorker方法的功能是增加一个线程，addWorker方法有两个参数：firstTask、core。

firstTask参数用于指定新增的线程执行的第一个任务，该参数可以为空；
core参数为true表示在新增线程时会判断当前活动线程数是否少于corePoolSize，false表示新增线程前需要判断当前活动线程数是否少于maximumPoolSize。

总体来讲，addWorker分为两个步骤

对线程池的状态和线程数等信息进行校验，不符合创建线程条件则返回false
创建worker

// 增加线程，并使它运行，最后返回是否成功
// firstTask参数用于指定新增的线程执行的第一个任务，该参数可以为空；
// core参数为true表示在新增线程时会判断当前活动线程数是否少于corePoolSize，
// false表示新增线程前需要判断当前活动线程数是否少于maximumPoolSize
private boolean addWorker(Runnable firstTask, boolean core) {// 检查线程池状态和线程数量是否符合预期，不符合预期直接返回falseretry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 1. 检查线程池状态// Check if queue empty only if necessary. RUNNING是最小的，rs >= SHUTDOWN表示线程池正在关闭// 这里写成(rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())) 更好理解些// 即线程池状态是 大于SHUTDOWN、线程池状态是SHUTDOWN并且firstTask不是空、线程池状态是SHUTDOWN并且队列是空if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {// 2. 检查线程数量int wc = workerCountOf(c);// 线程数量大于最大值（2 ^ 29 - 1）或者 core为true且大于corePoolSize 或者 core为false且大于maximumPoolSizeif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// WorkerCount增加成功则跳出循环往下走if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)// 线程池状态改变了，重新执行大循环，即需要重新检查下线程池的状态 continue retry;// else CAS failed due to workerCount change; retry inner loop，线程池状态没变，执行小循环，即不需要检查线程池状态了，只需要看能否增加线程数即可}}// 线程池状态和线程数量检查通过，开始创建Workerboolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;// t等于null的话说明创建线程失败了if (t != null) {// Doug Lea的极致优化final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();// 添加一个workerworkers.add(w);// largestPoolSize记录线程池同时运行过的最多的线程数量int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// worker添加成功，启动线程。t.start();workerStarted = true;}}} finally {// worker中的线程是null或者其他情况没将workerStarted置为true，则进行回滚if (! workerStarted)// workers.remove(w)、decrementWorkerCount()、tryTerminate();addWorkerFailed(w);}return workerStarted;}

Worker线程执行任务

在Worker类中的run方法调用了runWorker方法来执行任务，runWorker方法的执行过程如下：

while循环不断地通过getTask()方法获取任务。
getTask()方法从阻塞队列中取任务。
如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态。
执行任务。
如果getTask结果为null则跳出循环，执行processWorkerExit()方法，销毁线程。

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// Worker的构造函数里可以看到Worker将AQS的state初始化为-1，// 在tryRelease中直接赋值为0，所以在Worker加锁之前要先重置为0。// 为什么不直接从0开始呢，可以看到线程池的interruptIdleWorkers方法里会尝试给worker加锁然后interrupt 线程，这// 里加锁肯定是会失败的，这么做也是禁止worker刚创建就被interrupt。w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 核心线程addWorker时会设置task，此外都是从队列获取taskwhile (task != null || (task = getTask()) != null) {// 加锁，防止interruptIdleWorkers中断正在执行任务的workerw.lock();// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted.  This// requires a recheck in second case to deal with// shutdownNow race while clearing interrupt// 如果线程池处于STOP状态则要确保线程被中断，对应runStateAtLeast(ctl.get(), STOP)是true 并且 !wt.isInterrupted()，则wt.interrupt()// 否则，确保线程不被中断，也就是runStateAtLeast(ctl.get(), STOP)是false，Thread.interrupted()之后线程中断状态被置为false，再根据线程池状态判断是否要中断线程if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {// beforeExecute抛异常的话，会跳出循环，并且completedAbruptly是true，这时不会处理task，并且会走到processWorkerExit杀掉当前线程，然后再把异常抛出去beforeExecute(wt, task);Throwable thrown = null;try {// 执行任务，抛出异常的话会传递到afterExecute里，此时也会跳出循环杀掉当前的线程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 {// task是空并且getTask返回的也是空则销毁worker// 对于核心线程，如果设置了allowCoreThreadTimeOut为true，则核心线程在keepAliveTime后还获取不到任务，会释放核心线程// 对于非核心线程，在keepAliveTime后还获取不到任务，会释放非核心线程processWorkerExit(w, completedAbruptly);}}private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}// 尝试让线程池进入TERMINATED状态tryTerminate();int c = ctl.get();// 状态小于STOP说明还有未执行完的任务要处理if (runStateLessThan(c, STOP)) {// 看下Worker线程退出是不是因为执行过程中发生了异常if (!completedAbruptly) {// 没发生异常，如果allowCoreThreadTimeOut是false，则min是corePoolSize； 否则，队列非空则min是1，队列是空则min是0int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;// 线程数大于min，则不需要增加workerif (workerCountOf(c) >= min)return; // replacement not needed}// 分配一个空的非核心worker去把剩余任务执行完addWorker(null, false);}}

线程池关闭

shutdown方法

// shutdown方法是先设置状态再interrupt的，这和getTask里面的处理是吻合的，
// 先设置状态确保让worker先退出，避免interrupt的过程中线程又去获取任务。
public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 如果配置了security manager，则检查下caller是否有关闭线程的权限checkShutdownAccess();// 循环CAS将线程池状态置为SHUTDOWNadvanceRunState(SHUTDOWN);// 只清理闲置的Worker（例如不在执行中且未被中断的非核心线程）interruptIdleWorkers();onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}tryTerminate();}// 线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来中断空闲的线程
private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;// 能获取到锁，说明线程是空闲状态，可以对该线程进行中断，使线程被安全地回收// 如果worker在执行runWorker，则会先加锁，那么这里的w.tryLock()就返回false了if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}

shutdownNow方法

public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 循环CAS将线程池状态置为STOPadvanceRunState(STOP);// 把所有的worker都interruptinterruptWorkers();// 把任务队列中尚未处理的任务拿出来交给调用线程自行处理，tasks = drainQueue();} finally {mainLock.unlock();}tryTerminate();return tasks;}private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers)w.interruptIfStarted();} finally {mainLock.unlock();}}void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}

ThreadPoolExecutor源码分析相关推荐

JAVA线程池(ThreadPoolExecutor)源码分析
JAVA5提供了多种类型的线程池,如果你对这些线程池的特点以及类型不太熟悉或者非常熟悉,请帮忙看看这篇文章(顺便帮忙解决里面存在的问题,谢谢!): http://xtu-xiaoxin.ite ...
Java并发编程实战（chapter_3）（线程池ThreadPoolExecutor源码分析）
为什么80%的码农都做不了架构师?>>> 这个系列一直没再写,很多原因,中间经历了换工作,熟悉项目,熟悉新团队等等一系列的事情.并发课题对于Java来说是一个又重要又难的一大块 ...
java多线程系列：ThreadPoolExecutor源码分析
前言这篇主要讲述ThreadPoolExecutor的源码分析,贯穿类的创建.任务的添加到线程池的关闭整个流程,让你知其然所以然.希望你可以通过本篇博文知道ThreadPoolExecutor是怎么 ...
c++ 线程池_JAVA并发编程：线程池ThreadPoolExecutor源码分析
前面的文章已经详细分析了线程池的工作原理及其基本应用,接下来本文将从底层源码分析一下线程池的执行过程.在看源码的时候,首先带着以下两个问题去仔细阅读.一是线程池如何保证核心线程数不会被销毁,空闲线程数 ...
java多线程系列：ThreadPoolExecutor源码分析，java基础面试笔试题
我总结出了很多互联网公司的面试题及答案,并整理成了文档,以及各种学习的进阶学习资料,免费分享给大家. 扫描二维码或搜索下图红色VX号,加VX好友,拉你进[程序员面试学习交流群]免费领取.也欢迎各位一起 ...
threadpoolexecutor创建线程池_线程池ThreadPoolExecutor源码分析
什么是线程池创建线程要花费昂贵的资源和时间,如果任务来了才创建那么响应时间会变长,而且一个进程能创建的线程数量有限.为了避免这些问题,在程序启动的时候就创建若干线程来响应出来,它们被称为线程池,里面 ...
【java】java中的线程池 ThreadPoolExecutor源码分析
文章目录 1.概述 4.源码 4.1 关键属性 4.2 构造函数 4.4 状态控制 4.5 ThreadLocalMap 4.6 execute方法源码分析 4.7 addWorker方法源码分析 4 ...
聊聊高并发（四十）解析java.util.concurrent各个组件（十六） ThreadPoolExecutor源码分析
ThreadPoolExecutor是Executor执行框架最重要的一个实现类,提供了线程池管理和任务管理是两个最基本的能力.这篇通过分析ThreadPoolExecutor的源码来看看如何设计和实 ...
Java线程池ThreadPoolExecutor源码分析
继承关系 Executor接口 public interface Executor {void execute(Runnable command); } ExecutorService接口 publi ...
[JUC-4]ThreadPoolExecutor源码分析
相关博文: https://segmentfault.com/a/1190000008693801 https://www.jianshu.com/p/fa1eac9710c8 //构造器1 publ ...

ThreadPoolExecutor源码分析