本文参考：

文章框架参考：https://www.wormholestack.com/archives/668/

线程池设计解析源码长文：https://www.javadoop.com/post/java-thread-pool

美团的线程池文章，动态线程池引子：https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html

1. 引子

线程池作为日常开发中最常用的JUC工具，通过池化的思想提升了资源的利用率。之前最多就是了解其API的使用，七个参数八股信手拈来，但是对于其中的原理还是不甚了解，包括它是如何管理线程，如何实现拒绝策略的，这篇文章主要就是基于这个扫盲的目的而来的，整体来说站在了许多巨人的肩膀上，后面可以自己再不断往里面补充新的想法。

2. 线程池源码分析

2.1. 总览

Jdk 1.8 juc 包下面的关系

• Executor 为顶层接口，是最简单的，只暴露了一个 void execute(Runnable command) 方法

• ExecutorService 接口继承了 Executor 接口，额外添加了许多方法，所以一般来说后面都会用这个接口

• AbstractExecutorService 抽象类实现了 ExecutorService 接口，里面实现了一些方法，同时为子类提供了一些额外的方法直接使用

• ThreadPoolExecutor 就是线程池的接口

另外，我们还涉及到下图中的这些类：

Executors 类是个工具类，里面的方法都是静态方法，我们最常用的生成 ThreadPoolExecutor 实例的方法都在里面，包括 newFixedThreadPool、newCachedThreadPool。

另外，由于线程池支持获取线程执行的结果，所以，引入了 Future 接口，RunnableFuture 继承自此接口，然后我们最需要关心的就是它的实现类 FutureTask。在线程池的使用过程中，我们是往线程池提交任务（task），每个任务是实现了 Runnable 接口的，其实就是先将 Runnable 的任务包装成 FutureTask，然后再提交到线程池。FutureTask 这个类名的含义：它首先是一个任务（Task），然后具有 Future 接口的语义，即可以在将来（Future）得到执行的结果。

2.2. Executor

就一个接口带上一个最简单的抽象方法，入参是Runnable，寓意提交一个任务，至于任务是如何被执行的，则完全交给内部的实现，方法的注释上也说了，deiscretion 自由决断权。

public interface Executor {/*** Executes the given command at some time in the future.  The command* may execute in a new thread, in a pooled thread, or in the calling* thread, at the discretion of the {@code Executor} implementation.** @param command the runnable task* @throws RejectedExecutionException if this task cannot be* accepted for execution* @throws NullPointerException if command is null*/void execute(Runnable command);
}

所以理论上我只要能够执行command，具体实现并不关心，参考下面的实现。只不过我们这次是研究的线程池，线程池刚好是通过池化线程的方式执行 command 的。

class DirectRun implements Executor {@Overridepublic void execute(Runnable command) {// 直接执行 command 任务，不开启新县城command.run();}
}class ThreadRun implements Executor {@Overridepublic void execute(Runnable command) {// 开启一个线程执行 command 任务new Thread(command).run();}
}

2.3. ExecutorService

一般我们定义一个线程池，都是用的这个接口，它额外提供了很多抽象函数方法

ExecutorService threadPool1 = Executors.newFixedThreadPool(1);

public interface ExecutorService extends Executor {// 关闭线程池，已提交的任务继续执行（在任务队列中），不接受继续提交新任务 状态流转RUNNING -> SHUTDOWNvoid shutdown();// 关闭线程池，尝试停止正在执行的所有任务，不接受继续提交新任务 它和shutdown方法相比，区别在于它会去停止当前正在进行的任务  状态流转(RUNNING or SHUTDOWN) -> STOP// 返回值是等待执行的任务List<Runnable> shutdownNow();// 线程池是否已关闭boolean isShutdown();// 如果调用了 shutdown() 或 shutdownNow() 方法后，所有任务结束了，那么返回true // 必须在调用了 shutdown 和 shutdownNow 以后才会返回 trueboolean isTerminated();// 等待所有任务完成，并设置超时时间 实际应用中是，先调用 shutdown 或 shutdownNow， 然后再调这个方法等待所有的线程真正地完成，返回值意味着有没有超时boolean awaitTermination(long timeout， TimeUnit unit) throws InterruptedException;// 提交一个 Callable 任务<T> Future<T> submit(Callable<T> task); // 提交一个 Runnable 任务，第二个参数将会放到 Future 中，作为固定的返回值，因为 Runnable 的 run 方法本身并不返回任何东西<T> Future<T> submit(Runnable task， T result); // 提交一个 Runnable 任务Future<?> submit(Runnable task); // 执行所有任务，返回 Future 类型的一个 list<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;// 执行所有任务，返回 Future 类型的一个 list，有超时时间<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks，long timeout， TimeUnit unit)throws InterruptedException; // 只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果<T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException， ExecutionException;// 只有其中的一个任务结束了，就可以返回，返回执行完的那个任务的结果，超过指定的时间，抛出 TimeoutException 异常<T> T invokeAny(Collection<? extends Callable<T>> tasks，long timeout， TimeUnit unit)throws InterruptedException， ExecutionException， TimeoutException; 
}

2.4. AbstractExecutorService

AbstractExecutorService 抽象类派生自 ExecutorService 接口，提供了多个 protected 方法供子类使用

public abstract class AbstractExecutorService implements ExecutorService {/*** 将 Runnable 包装成 FutureTask 提交到线程池中执行，Runnable run() 方法无返回值，所以自定义 value*/protected <T> RunnableFuture<T> newTaskFor(Runnable runnable， T value) { return new FutureTask<T>(runnable， value);}/*** 将 Callable 任务包装成 FutureTask 提交到线程池中执行*/protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {return new FutureTask<T>(callable);}/*** 提交任务*/public Future<?> submit(Runnable task) {  if (task == null) throw new NullPointerException();// 将任务包装成 FutureTask，由于传入的 value 为 null，所以返回结果直接 RunnableFuture<Void>RunnableFuture<Void> ftask = newTaskFor(task， null); // 交给执行器执行，execute 方法由具体的子类来实现execute(ftask); return ftask;}/*** @throws RejectedExecutionException {@inheritDoc}* @throws NullPointerException       {@inheritDoc}*/public <T> Future<T> submit(Runnable task， T result) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task， result);execute(ftask);return ftask;}/*** @throws RejectedExecutionException {@inheritDoc}* @throws NullPointerException       {@inheritDoc}*/public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}// 本文中 invokeAny 和 invokeAll 方法并非重点，这里省略
}

其中主要是三个 submit 重载方法，支持传入 Runable，Callable，他们内部的实现都会分别调用newTaskFor 方法将其包装为 FutureTask 类，主要是因为这三个方法都需要获取结果。FutureTask 实现了 Runnable 接口，也可以放入 execute() 方法中执行，同时实现了 Future 这个获取结果的接口。

现在为止就可以看到后面实现AbstractExecutorService接口的线程池有两种提交任务的方式了：

1. 使用最顶层的 Executor 直接调用 void execute(Runnable command)，但是没有返回值
2. 使用 AbstractExecutorService 抽象类调用其中的三种 Future submit() 方法，可以将任务执行的返回值带出来

但是从上面的 submit() 方法也可以看出来，FutureTask 最底层的调用其实也是通过 void execute(Runnable command) 做的，只是在 AbstractExecutorService 中还没有具体实现，这需要交给子类的具体实现去做，ThreadPoolExecutor 线程池实现类就是做了这个事情。

==上面 Runnable 和 Callable 只是提交了具体的任务，最终实现还是得分别调用其 run() 或者 call() 方法，任务做的事情也是写在这两个方法里面的。==具体原理还得看后面线程池的代码慢慢研究下。

2.5. ThreadPoolExecutor

ThreadPoolExecutor 就是线程池的具体实现，它是通过构造函数传递核心参数的

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;}

构造函数核心参数

其中简单看一下构造函数中各参数的含义，它们在 ThreadPoolExecutor 中都是 volatile 属性

corePoolSize：线程池核心线程大小

是线程池中的一个最小的线程数量，即使这些线程处理空闲状态，他们也不会被销毁，除非设置了allowCoreThreadTimeOut，简单来说线程池分为两个部分，核心线程池和非核心线程池，核心线程池中的线程一旦创建便不会被销毁，非核心线程池中的线程在创建后如果长时间没有被使用则会被销毁。

maximumPoolSize：线程池最大线程数量

整个线程池的大小，此值大于等于1。线程池不会无限制的去创建新线程，它会有一个最大线程数量的限制，这个数量即由maximunPoolSize指定。工作队列满，且线程数等于最大线程数，此时再提交任务则会调用拒绝策略。maximumPoolSize - corePoolSize = 非核心线程池的大小

keepAliveTime:多余的空闲线程存活时间

非核心线程池中的线程在 keepAliveTime 时间内没有被使用就会被销毁，时间单位由 TimeUnit unit 决定。

当线程空闲时间达到 keepAliveTime 值时，多余的线程会被销毁直到只剩下 corePoolSize 个线程为止。

TimeUnit unit:空闲线程存活时间单位

keepAliveTime的计量单位

BlockingQueue workQueue:任务队列

阻塞队列用来存储任务，当有新的请求线程处理时，如果核心线程池已满，新来的任务会放入 workQueue 中，等待线程处理，JUC提供的阻塞队列有很多，例 ArrayBlockingQueue，LinkedBlockingQueue，PriorityBlockingQueue，SynchronousQueue 等

ThreadFactory:工厂类对象

线程池的创建传入了此参数时，是通过工厂类中的 newThread()方法来实现。

RejectedExecutionHandler handler:拒绝策略

如果线程池中没有空闲线程，已存在 maximumPoolSize 个线程，且阻塞队列 workQueue 已满，这时再有新的任务请求线程池执行，会触发线程池的拒绝策略，可以通过参数 handler 来设置拒绝策略，注意只有有界队列例如 ArrayBlockingQueue 或者指定大小的 LinkedBlockingQueue 等拒绝策略才有用，因为无解队列拒绝策略永远不会被触发。

阻塞队列

任务缓冲模块是线程池能够管理任务的核心部分。线程池的本质是对任务和线程的管理，而做到这一点最关键的思想就是将任务和线程两者解耦，不让两者直接关联，才可以做后续的分配工作。线程池中是以生产者消费者模式，通过一个阻塞队列来实现的。阻塞队列缓存任务，工作线程从阻塞队列中获取任务。

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是：**在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。**阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

下图中展示了线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素：

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

拒绝策略

ThreadPoolExecutor 内定义好了一些拒绝策略，都是静态类实现了 RejectedExecutionHandler 接口

public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }// 只要线程池没有被关闭，那么由提交任务的线程自己来执行这个任务。public void rejectedExecution(Runnable r， ThreadPoolExecutor e) {if (!e.isShutdown()) { // 线程池未关闭r.run(); // 当前调用者线程同步执行}}
}public static class AbortPolicy implements RejectedExecutionHandler {public AbortPolicy() { }// 不管怎样，直接抛出 RejectedExecutionException 异常 默认的策略，如果我们构造线程池的时候不传相应的 handler 的话，那就会指定使用这个public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}
}public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy() { }// 不做任何处理，直接丢掉这个任务public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}
}public static class DiscardOldestPolicy implements RejectedExecutionHandler { public DiscardOldestPolicy() { }// 如果线程池没有被关闭的话， 把队列队头的任务(也就是等待了最长时间即将执行的)直接扔掉，然后提交当前任务r到等待队列中public void rejectedExecution(Runnable r， ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}}
}

核心属性

接着我们来看下 ThreadPoolExecutor 中核心的属性变量

TreadPoolExecutor 使用一个 32 位整数来存放线程池的状态和当前池中的线程数，其中高 3 位用于存放线程池状态，低 29 位表示线程数。通过位运算计算，相比于基本运算，增强了计算速度。

// 用此变量保存当前池状态（高3位）和当前线程数（低29位）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING， 0));// COUNT_BITS 设置为 29(32-3)，意味着前三位用于存放线程状态，后29位用于存放线程数
private static final int COUNT_BITS = Integer.SIZE - 3;// 000 11111111111111111111111111111  这里得到的是 29 个 1，也就是说线程池的最大线程数是 2^29-1=536870911
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;  // 将整数 c 的低 29 位修改为 0，就得到了线程池的状态
private static int runStateOf(int c)     { return c & ~CAPACITY; } // 将整数 c 的高 3 为修改为 0，就得到了线程池中的线程数
private static int workerCountOf(int c)  { return c & CAPACITY; } 

线程池中的各个状态也是通过位运算计算

// 线程池的状态存放在高 3 位中 运算结果为 111跟29个0：111 00000000000000000000000000000 .接受新的任务，处理等待队列中的任务   
private static final int RUNNING    = -1 << COUNT_BITS;// 000 00000000000000000000000000000 .不接受新的任务提交，但是会继续处理等待队列中的任务
private static final int SHUTDOWN   =  0 << COUNT_BITS;// 001 00000000000000000000000000000 .不接受新的任务提交，不再处理等待队列中的任务，中断正在执行任务的线程
private static final int STOP       =  1 << COUNT_BITS; // 010 00000000000000000000000000000 .所有的任务都销毁了，workCount 为 0。线程池的状态在转换为 TIDYING 状态时，会执行钩子方法 terminated()
private static final int TIDYING    =  2 << COUNT_BITS; // 011 00000000000000000000000000000 .terminated() 方法结束后，线程池的状态就会变成这个
private static final int TERMINATED =  3 << COUNT_BITS;

线程池状态

在这里，介绍下线程池中的各个状态和状态变化的转换过程：

• RUNNING -1：这是最正常的状态：接受新的任务，并且也能处理阻塞队列中的任务
• SHUTDOWN 0：不接受新的任务提交，但是会继续处理阻塞队列中的任务
• STOP 1：不接受新的任务提交，不再处理阻塞队列中的任务，中断正在执行任务的线程
• TIDYING 2：所有的任务都销毁了，workCount 为 0。线程池的状态在转换为 TIDYING 状态时，会执行钩子方法 terminated()
• TERMINATED 3：terminated() 方法结束后，线程池的状态就会变成这个

RUNNING 定义为 -1，SHUTDOWN 定义为 0，其他的都比 0 大，所以等于 0 的时候不能提交任务，大于 0 的话，连正在执行的任务也需要中断。

各个状态的转换过程有以下几种：

• RUNNING -> SHUTDOWN：当调用了 shutdown() 后，会发生这个状态转换，这也是最重要的
• (RUNNING or SHUTDOWN) -> STOP：当调用 shutdownNow() 后，会发生这个状态转换，这下要清楚 shutDown() 和 shutDownNow() 的区别了
• SHUTDOWN -> TIDYING：当任务队列和线程池都清空后，会由 SHUTDOWN 转换为 TIDYING
• STOP -> TIDYING：当任务队列清空后，发生这个转换
• TIDYING -> TERMINATED：这个前面说了，当 terminated() 方法结束后

Worker 类

另外，我们还要看看一个内部类 Worker，因为 Doug Lea 把线程池中的线程包装成了一个个 Worker，翻译成工人，就是线程池中做任务的线程。所以到这里，我们知道任务是 Runnable（内部变量名叫 task 或 command），线程是 Worker。

里面的属性主要是 thread、firstTask、completedTasks

它继承了抽象类 AbstractQueuedSynchronizer，用 AQS 来实现独占锁，为的就是实现不可重入的特性去反映线程现在执行的状态。同时它还实现了 Runnable 接口，后面在 addWorker() 方法中会调用。

private final class Workerextends AbstractQueuedSynchronizerimplements Runnable
{private static final long serialVersionUID = 6138294804551838833L;// 执行任务的真正线程；由线程工厂创建，如果工厂创建失败则为 nullfinal Thread thread;// 这里的 Runnable 是任务 这个线程起来以后需要执行的第一个任务，那么第一个任务就是存放在这里的(线程可不止执行这一个任务)// 也可以为 null，这样线程起来了，自己到任务队列 BlockingQueue 中取任务（getTask 方法）Runnable firstTask;// 用于存放此线程完成的任务数，通过 volatile 保证变量的可见性（防止 CPU 缓存的内存不可见问题）volatile long completedTasks;//  Worker 只有这一个构造方法，传入 firstTask，也可以传 null，后面 addWorker() 方法会调用Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;// 调用 ThreadFactory 来创建一个新的线程，但是并没有启动，启动交给后面的 addWorker() 方法// Runnable 方法传入的 this 就是 Worker，其本身也实现了 Runnable 接口，thread#start() 就会执行下面的 run 方法this.thread = getThreadFactory().newThread(this);}// 重写 Runnable#run()方法，thread#start() 就会执行public void run() {// 调用了外部类 ThreadPoolExecutor 中的 runWorker 方法runWorker(this);}// 下面都是通过 AQS 的操作，来获取线程的执行权，用了独占锁，后面可以再去了解下/*// 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) {}}}*/
}

execute() 方法

之前在 AbstractExecutorService 抽象类里面看到最终任务的执行都是要依赖子类的 execute 方法的实现，现在总算可以看 ThreadPoolExecutor#execute() 方法了。调用了不少内部方法，看着比较简洁，但也相对抽象。主要做的是线程申请相关的动作，即 addWorker()。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();// 一个变量保存当前池状态（高3位）和当前线程数（低29位）int c = ctl.get();	// 如果当前线程数小于和线程数，那么直接添加一个 worker 来执行任务if (workerCountOf(c) < corePoolSize) {// 添加任务成功，就直接返回，执行的结果会包装到 FutureTask 中。// 返回 false 代表线程池不允许提交任务if (addWorker(command, true))return;// addWorker = false 重新获取 cc = ctl.get();}// 执行到这里，要么当前线程数大于等于 corePoolSize，要么刚刚 addWorker 失败了// 如果线程处于 RUNNING 状态，则把这个任务添加到任务队列 workQueue 中if (isRunning(c) && workQueue.offer(command)) {// 如果当前线程数大于等于 corePoolSize，就会进到这里// 二次状态检查int recheck = ctl.get();// 如果线程池已不处于 RUNNING 状态，那么移除已经入队的这个任务，并且执行拒绝策略if (! isRunning(recheck) && remove(command))reject(command);// 否则如果线程池还是 RUNNING 的，并且线程数为 0，那么开启新的线程，这里的目的是担心前面任务提交到队列中，但是没有可用的线程了else if (workerCountOf(recheck) == 0)// 创建Worker，并启动里面的Thread，为什么任务传null，线程启动后会自动从阻塞队列拉取任务来执行addWorker(null, false);}// 如果 workQueue 队列满了，那么进入到这个分支// 以 maximumPoolSize 为界创建新的工作 worker，如果失败，说明当前线程数已经达到 maximumPoolSize，执行拒绝策略else if (!addWorker(command, false))// 失败执行拒绝策略reject(command);
}

简单总结下上述代码的逻辑

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

二次检查是为了应对并发情况，从上次判断线程池状态到现在线程池可能会被关闭，由于线程池关闭后不能再继续添加任务了，此时就需要回滚刚才的添加任务到队列中的操作，并执行拒绝策略。其实在 JUC 包里面二次检查的代码蛮多的，核心原因在于执行这些代码序列都不是原子的操作，序列中任何时刻都有可能会被外界改变线程池状态，在执行一些相对耗时的操作判断以后，可以去二次获取一些变量值。

引用美团线程池篇图 执行流程如下图所示

其中判断线程池是否还在运行，其实在 addWorker 方法刚开始也判断了一次，然后在 execute 方法里又额外判断了一次

addWorker() 方法

Worker 线程增加就是用的这个方法，该方法不考虑线程池是在哪个阶段增加的该线程，该步骤仅仅完成增加线程，并使它运行，最后返回是否成功这个结果。

第一个参数 firstTask 是准备提交给这个线程执行的任务，当为 null 时，线程启动后会自动从阻塞队列拉任务执行。

引用美团线程池篇图 执行流程如下图所示：

第二个参数 core 为 true 代表使用corePoolSize作为创建线程的界限，也就说创建这个线程的时候，如果线程池中的线程总数已经达到核心线程数，那么不能响应这次创建线程的请求，如果是 false，代表使用maximumPoolSize作为界限。

引用美团线程池篇图 执行流程如下图所示：

break retry 跳到retry处，且不再进入循环
continue retry 跳到retry处，且再次进入循环.

private boolean addWorker(Runnable firstTask, boolean core) {retry:// 死循环，等待中间逻辑进行中断 return, breakfor (;;) {int c = ctl.get();int rs = runStateOf(c);// 这个表达式非常不好理解，需要将其逻辑表达式做一下计算// 如果线程池已关闭，并满足以下条件之一，那么不创建新的 worker：// 1. 线程池状态大于 SHUTDOWN，其实也就是 STOP, TIDYING, 或 TERMINATED// 2. firstTask != null// 3. workQueue.isEmpty()// 简单分析下：// 还是状态控制的问题，当线程池处于 SHUTDOWN 的时候，不允许提交任务，但是已有的任务继续执行// 当状态大于 SHUTDOWN 时，不允许提交任务，且中断正在执行的任务// 多说一句：如果线程池处于 SHUTDOWN，但是 firstTask 为 null，且 workQueue 非空，那么是允许创建 worker 的// 这是因为 SHUTDOWN 的语义：不允许提交新的任务，但是要把已经进入到 workQueue 的任务执行完，所以在满足条件的基础上，是允许创建新的 Worker 的if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {// 根据 core 条件，将当前线程池工作线程数 wc 分别与 corePoolSize/maximumPoolSize 进行比较int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;//  如果成功，那么就是所有创建线程前的条件校验都满足了，准备创建线程执行任务了// 这里失败的话，说明有其他线程也在尝试往线程池中创建线程if (compareAndIncrementWorkerCount(c))break retry;// 前面 CAS，这里重新读取 CTLc = ctl.get();  // Re-read ctl// 如果因为其他线程的操作，导致线程池的状态发生了变更，那么就需要重新回到外层的 retry 进行下一轮循环判断if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}// 前面的两层 for 循环都是校验工作，下面总算要开始创建线程执行任务了// worker 是否已经启动 boolean workerStarted = false;// 是否已经将这个 work 添加到 workers 的 HashSet 中boolean workerAdded = false;Worker w = null;try {// worker 的构造方法w = new Worker(firstTask);// 取 worker 中的线程对象，之前说了，Worker的构造方法会调用 ThreadFactory 来创建一个新的线程final Thread t = w.thread;if (t != null) {// ThreadPoolExecutor 中的唯一一把 ReentrantLockfinal ReentrantLock mainLock = this.mainLock;// 整个线程池的全局锁，放在 try 外面// 因为关闭一个线程池需要这个锁，至少我持有锁的期间，线程池不会被关闭mainLock.lock();try {int rs = runStateOf(ctl.get());// 小于 SHUTTDOWN 那就是 RUNNINGif (rs < SHUTDOWN ||// 如果等于 SHUTDOWN，不接受新的任务，但是会继续执行等待队列中的任务(rs == SHUTDOWN && firstTask == null)) {// worker 里面的 thread 可不能是已经启动的if (t.isAlive()) throw new IllegalThreadStateException();// 加到 workers 这个 HashSet 中workers.add(w);// largestPoolSize 用于记录 workers 中的个数的最大值，动态记录int s = workers.size();if (s > largestPoolSize)// 因为 workers 是不断增加减少的，通过这个值可以知道线程池的大小曾经达到的最大值largestPoolSize = s;// 更改状态workerAdded = true;}} finally {// 上面的workers增加，workerAdded状态改变改变以后，就可以释放锁了mainLock.unlock();}// 如果worker添加成功if (workerAdded) {// 启动这个线程，worker.thread#start -> worker#run -> runWorker(this)t.start();// 更改状态workerStarted = true;}}} finally {// 如果线程没有启动，需要做一些清理工作，如前面 workCount 加了 1，将其减掉if (! workerStarted)addWorkerFailed(w);}// 返回线程是否创建成功return workerStarted;
}

从上面源码结合前面分析的 Worker 类源码可以看出来，if (workerAdded) 后 t.start() 启动 worker 中的线程，最终还是会调用 runWorker 方法

runWorker() 方法

runWorker方法的执行过程如下：

1. while循环不断地通过getTask()方法获取任务。线程还是复用的上面 t.start() 的 t.
2. getTask()方法从阻塞队列中取任务。获取任务的先后顺序与线程启动的先后顺序无关，线程有空闲就可以 getTask()。
3. 如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态。
4. 执行传入进来的 Runnable 任务，worker.thread#start -> worker#run -> runWorker(this)
5. 如果getTask结果为null则跳出循环，执行processWorkerExit()方法，销毁线程。

执行流程如下图所示：

// 此方法由 worker 线程启动后调用，用一个 while 循环来不断从等待队列中获取任务并执行
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// while循环调用 getTask() 获取任务while (task != null || (task = getTask()) != null) {// 加上 worker 的独占锁w.lock();// 如果线程池状态大于等于 STOP，那么意味着该线程也要中断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) {// 这里不允许抛出 Throwable，所以转换为 Errorthrown = x; throw new Error(x);} finally {// 也是一个钩子方法，将 task 和异常作为参数，留给需要的子类实现afterExecute(task, thrown);}} finally {// 置空 task，准备 getTask 获取下一个任务task = null;// 累加完成的任务数w.completedTasks++;// 释放掉 worker 的独占锁w.unlock();}}completedAbruptly = false;} finally {// 如果到这里，需要执行线程关闭：// 1. 说明 getTask 返回 null，也就是说，队列中已经没有任务需要执行了，执行关闭// 2. 任务执行过程中发生了异常// 第一种情况，已经在代码处理了将 workCount 减 1，这个在 getTask 方法分析中会说// 第二种情况，workCount 没有进行处理，所以需要在 processWorkerExit 中处理processWorkerExit(w, completedAbruptly);}
}

getTask() 方法

前面提到， Worker 线程启动后调用，会通过 while 循环来不断地通过 getTask 方法从等待队列中获取任务并执行达到线程回收。

getTask这部分进行了多次判断，为的是控制线程的数量，使其符合线程池的状态。如果线程池现在不应该持有那么多线程，则会返回null值。工作线程Worker会不断接收新任务去执行，而当工作线程Worker接收不到任务的时候，就会开始被回收，回收的方法在 processWorkerExit() 内。

// 此方法有三种可能：
// 1. 阻塞直到获取到任务返回。我们知道，默认 corePoolSize 之内的线程是不会被回收的，
//      它们会一直等待任务
// 2. 超时退出。keepAliveTime 起作用的时候，也就是如果这么多时间内都没有任务，那么应该执行关闭
// 3. 如果发生了以下条件，此方法必须返回 null:
//    - 池中有大于 maximumPoolSize 个 workers 存在(通过调用 setMaximumPoolSize 进行设置)
//    - 线程池处于 SHUTDOWN，而且 workQueue 是空的，前面说了，这种不再接受新的任务
//    - 线程池处于 STOP，不仅不接受新的线程，连 workQueue 中的线程也不再执行
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// 两种可能// 1. rs == SHUTDOWN && workQueue.isEmpty()// 2. rs >= STOPif (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// CAS 操作，减少工作线程数，工作线程应该在 shutdown() 方法里面就被 interrupt 了decrementWorkerCount();return null;}int wc = workerCountOf(c);// 1. 可以设置 allowCoreThreadTimeOut 允许 core 线程数被回收// 2. 当前线程数超过了核心线程数，发生超时关闭，这个是符合以前我们预期的boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 池中有大于 maximumPoolSize 个 workers 存在，返回 nullif ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 到 workQueue 中获取任务Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // 超时未获取到，返回 nullworkQueue.take(); //  corePoolSize 之内的线程一直会阻塞在这里等待任务返回if (r != null)return r;// 这里说明是 poll() 超时了，返回值为 nulltimedOut = true;} catch (InterruptedException retry) {// 如果此 worker 发生了中断，采取的方案是重试// 解释下为什么会发生中断，这个要去看 setMaximumPoolSize 方法。// 如果开发者将 maximumPoolSize 调小了，导致其小于当前的 workers 数量，// 那么意味着超出的部分线程要被关闭。重新进入 for 循环，自然会有部分线程会返回 nulltimedOut = false;}}
}

需要注意的是

• 线程池处于 SHUTDOWN，而且 workQueue 是空的，该方法返回 null，这种不再接受新的任务。
• 线程池中有大于 maximumPoolSize 个 workers 存在，这种可能是因为有可能开发者调用了 setMaximumPoolSize() 将线程池的 maximumPoolSize 调小了，那么多余的 Worker 就需要被关闭
• 线程池处于 STOP，不仅不接受新的线程，连 workQueue 中的线程也不再执行
• 如果此 worker 发生了中断，采取的方案是重试，也就是说如果开发者将 maximumPoolSize 调小了，导致其小于当前的 workers 数量，那么意味着超出的部分线程要被关闭。重新进入 for 循环获取任务
• workQueue.take() corePoolSize 之内的线程一直会阻塞在这里等待任务返回，在未获取到前这些线程不会关闭，并且获取到了任务 task 以后，从 runWorker 逻辑里面看出来，一直在 while 循环里面，也不会走到 processWorkerExit() 方法里，不会回收线程。

// 动态线程池内就会用到这个
public void setMaximumPoolSize(int maximumPoolSize) {if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)throw new IllegalArgumentException();this.maximumPoolSize = maximumPoolSize;if (workerCountOf(ctl.get()) > maximumPoolSize)// 中断 worker，超出的线程需要被关闭interruptIdleWorkers();
}private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;if (!t.isInterrupted() && w.tryLock()) {try {// 线程中断，会抛出 InterruptedExceptiont.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}
}

processWorkerExit() 方法

线程池中线程的销毁依赖JVM自动的回收，线程池做的工作是根据当前线程池的状态维护一定数量的线程引用，防止这部分线程被JVM回收，当线程池决定哪些线程需要回收时，只需要将其引用消除即可（这个和 GC Root 回收机制有关）。Worker被创建出来后，就会不断地进行轮询，然后获取任务去执行，核心线程可以无限等待获取任务，非核心线程要限时获取任务。当Worker无法获取到任务，也就是获取的任务为空时，循环会结束，Worker会主动消除自身在线程池内的引用。

核心代码：workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) 超过线程存活时间 keepAliveTime 获取不到，就会将这个线程回收

try {// 获取不到任务，和上面 getTask() 有关系// workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) 超过线程存活时间 keepAliveTime 获取不到，就会将这个线程回收while (task != null || (task = getTask()) != null) {//执行任务}
} finally {processWorkerExit(w, completedAbruptly);//获取不到任务时，主动回收自己
}

线程回收的工作是在processWorkerExit方法完成的。

事实上，在这个方法中，将线程引用移出线程池就已经结束了线程销毁的部分。但由于引起线程销毁的可能性有很多，线程池还要判断是什么引发了这次销毁，是否要改变线程池的现阶段状态，是否要根据新状态，重新分配线程。

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;// 将获取不到 task 的线程移出线程池，删除了线程引用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; // replacement not needed}addWorker(null, false);}
}

从 workers.remove(w); 可以看出来，将获取不到 task 的线程移出线程池，然而刚开始创建的 core 线程是一定不会走到这里的，所以只会删除后面创建的 [corePoolSize, maximumPoolSize] 那些线程。

3. 实际问题

线程池使用面临的核心的问题在于：线程池的参数并不好配置。一方面线程池的运行机制不是很好理解，配置合理需要强依赖开发人员的个人经验和知识；另一方面，线程池执行的情况和任务类型相关性较大，IO密集型和CPU密集型的任务运行起来的情况差异非常大，这导致业界并没有一些成熟的经验策略帮助开发人员参考。

关于线程池配置不合理引发的故障，美团公司内部有较多记录，下面引用一些它举的例子：

Case1：2018年XX页面展示接口大量调用降级：

事故描述：XX页面展示接口产生大量调用降级，数量级在几十到上百。

事故原因：该服务展示接口内部逻辑使用线程池做并行计算，由于没有预估好调用的流量，阻塞队列满了以后，最大核心数设置偏小，没法满足大流量执行条件，拒绝策略大量抛出RejectedExecutionException，触发接口降级条件，示意图如下：

Case2：2018年XX业务服务不可用S2级故障

事故描述：XX业务提供的服务执行时间过长，作为上游服务整体超时，大量下游服务调用失败。

事故原因：该服务处理请求内部逻辑使用线程池做资源隔离，由于阻塞队列设置过长，最大线程数设置失效，导致请求数量增加时，大量任务堆积在队列中，任务执行时间过长，最终导致下游服务的大量调用超时失败。这是由于 corePoolSize设置过小导致任务执行速度低，并且阻塞队列设置过长无法调用非核心线程执行，request IO 线程超时了。

示意图如下：