线程池详解

一，为什么要用线程池

① Java的线程模型是基于操作系统的原生线程模型实现的，所以说Java线程实际上是基于内核实现的，创建，析构，同步都需要从用户态切换至内核态，这样带来的性能损耗是很大的。

② 每个Thread都需要有一个内核线程支持，也就意味着每个线程都要消耗一定的内核资源，一个线程的消耗大概时1M

③ 当线程资源过多时，很难进行管理，工程师想要对线程进行管理或者更多功能扩展，需要花费更多的精力

二，池化技术

个人理解池化技术更多的是代表这一种思想，他在各个领域都有体现，在计算机中：内存池，线程池，连接池，在金融中也有现金池，还有在面试中令人口诛笔伐的人才池。这种思想是为了让资源统一起来更好管理，随用随取。

三，ThreadPoolExecutor 设计架构图

• Executor

只提供了execute(Runnable)方法，用户只需将任务提交，而无需知道线程是如何创建的

• ExecutorService

（1）扩充执行任务的能力，补充可以为一个或一批异步任务生成Future的方法；

（2）提供了管控线程池的方法，比如停止线程池的运行。

• AbstractExecutorService

实现了 ExecutorService 接口，实现了除 execute 以外的所有方法，只将一个最重要的 execute 方法交给 ThreadPoolExecutor 实现

四，线程池的构造以及参数

线程池由线程存储池HashSet以及阻塞队列BlockQueue组成，同时他还包含了RejectedExecutionHandler拒绝策略

① 线程池构造参数

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {

• **corePoolSize：**核心线程数。当前线程数小于核心线程，在提交新任务时，不管线程的状态，都会创建线程，并把新任务交给新线程处理，这些核心线程不会被回收，除非设置了allowCoreThreadTimeOut进行回收

• **workQueue：**阻塞队列。当前的线程数大于或者等于核心线程数时，提交新任务会直接进入阻塞队列中，线程去阻塞队列中拉取

• maximumPoolSize： 最大线程数。提交新的任务，如果提交队列满了，但线程数小于最大线程数，此时会创建新的线程来处理任务。如果提交任务时队列已满了但线程数也到达了最大线程数，此时就会触发拒绝策略。

• **RejectedExecutionHandler：**它对应了拒绝策略，一共有四种拒绝策略

  • **AbortPolicy：**丢弃任务并抛出异常，这也是默认策略；
  • **CallerRunsPolicy：**用调用者所在的线程来执行任务，所以开头的问题「线程把任务丢给线程池后肯定就马上返回了?」我们可以回答了，如果用的是 CallerRunsPolicy 策略，提交任务的线程（比如主线程）提交任务后并不能保证马上就返回，当触发了这个 reject 策略不得不亲自来处理这个任务。
  • **DiscardOldestPolicy：**丢弃阻塞队列中靠最前的任务，并执行当前任务。
  • **DiscardPolicy：**直接丢弃任务，不抛出任何异常，这种策略只适用于不重要的任务。

• **KeepAliveTime：**线程存活时间，如果非核心线程在此时间内处于休眠状态就会被回收

• **ThreadFactory：**可以用此参数对线程池进行命名，指定 defaultUncaughtExceptionHandler（异常抛出策略）,甚至可以设定线程为守护线程

② WorkQueue

WorkQueue它本质上是一个阻塞队列。而WorkQueue设计思想是为了将整个线程池进行解耦，将任务和线程分为两个部分。

private final BlockingQueue<Runnable> workQueue;

WorkQueue关键阶段

• 线程数>=核心线程：当这种情况发生时，会将任务提交至阻塞队列中，线程自行去offer()消费任务
• **阻塞队列已满&&线程数<最大线程数：**当这种情况发生时，会新增Worker，并将任务交给Worker

无界队列与有界队列对于线程池的影响

其实在上面我们可以发现，在阻塞队列已满的情况下，会创建新的线程来处理任务。但是如果队列是无界的呢，没错！此时队列永远无法到达队列已满的情况，将导致无法创建除核心线程以外的线程，在有大量的请求的情况下，消费能力不会改变，但是生产能力不断提升，导致阻塞队列堆满，抛出OOM异常。

以下是不同阻塞队列的介绍，线程池搭配不同的阻塞队列可以实现不同的效果：

③ Worker

worker是线程池将对于Thread线程的一种封装，他提供了更加高级的功能，以及更加方便管理。

• **Lock：**worker运行时，对整个worker对象上锁，保证对象运行安全
• **firstTask：**直接分配得到的任务，不是从阻塞队列中获取的。
• **Thread：**执行任务的线程

五，线程池的运行流程

六，线程池的状态

• **Running：**接收新的任务，并能继续处理workQueue的任务
• **SHUTDOWN：**不再接收新的任务，但c是在运行的任务会继续运行，并且会处理完BlockQueue中的任务
• **STOP：**不在接收新的任务，也不暂停此时所有线程，也不处理BlockQueue中剩余的任务
• **TIDYING：**所有任务都完结了，线程数量为0，即为当前状态，进入此状态后，会调用terminated()进入TERMINATED状态
• **TERMINATED：**调用terminated()即为此状态

七，线程池源码分析

前言：了解简单的线程池内容后，我们来到最重磅的环境，源码分析。对于源码分析，很多同学肯定都是头疼的，看源码肯定也不是一股脑去看，我们要懂得带着问题去找答案，所以在进行源码分析前，大家可以看看这些面试问题来进入源码世界。

1. 线程池的状态是如何被维护的，线程池的数量又是怎么被维护的
2. 提交任务之后会发生什么，是不是就直接返回了？
3. execute()和sumbit()的区别是什么？
4. 线程是怎么被添加的？
5. 线程是怎么获取任务的？
6. 为什么说使用无界队列会使线程池无法创建非核心线程
7. Worker不是加了AQS吗，那么shutdownNow()是怎么把Worker关闭的
8. 为什么Worker使用AQS不使用ReentrantLock
9. shutdown()之后，还会创建线程吗？
10. 如果一个线程发生异常，会发生什么，会放回线程池吗？一定会抛出异常吗？
11. 如何优雅的关闭线程池

1，ctl——线程池状态与数量的优雅表达

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;

我们看上面的源码可以发现，大佬Doug Lea将线程池的状态和线程数量用一个参数进行完美的表达

int ctl是32位，前3位表示状态，后29位用于表示线程池的线程数量

于是便有了这些方法来进行计算，记住这些，这些方法在后续的旅行中会用到

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; }		//获取ctl值

2，submit&execute——步入线程池的大门

execute方法

public void execute(Runnable command) {if (command == null)throw new NullPointerException();//获取ctlint c = ctl.get();//查询运行线程的数目是否小于corePoolSizeif (workerCountOf(c) < corePoolSize) {//增加新的worker,true代表该线程为核心线程if (addWorker(command, true))return;//添加失败，重新获取ctl，因为线程池状态可能在这个时候改变c = ctl.get();}/**前置条件：1. 运行线程 > 核心线程2. 添加线程失败如果线程池正在运行，则将任务添加至阻塞队列**/if (isRunning(c) && workQueue.offer(command)) {// 重新获取ctl值int recheck = ctl.get();// 再次判断线程池的运行状态，如果不是运行状态，由于之前已经把command添加到workQueue中了，// 这时需要移除该command// 执行过后通过handler使用拒绝策略对该任务进行处理，整个方法返回if (! isRunning(recheck) && remove(command))reject(command);/** 获取线程池中的有效线程数，如果数量是0，则执行addWorker方法，让一个非核心线程去处理阻塞队列中的任务*/else if (workerCountOf(recheck) == 0)addWorker(null, false);}/** 如果执行到这里，有两种情况：* 1. 线程池已经不是RUNNING状态；* 2. 线程池是RUNNING状态，但workerCount >= corePoolSize并且workQueue已满。* 这时，再次创建 非核心线程 直接处理当前的任务。* 如果失败则拒绝该任务*/else if (!addWorker(command, false))reject(command);
}

我们来梳理一下大致流程

1. 如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务；
2. 如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中；
3. 如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务；
4. 如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

sumbit方法

public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;
}

两者区别

返回类型不同

从上面的代码可以发现，execute是返回void，而submit返回的Future类，可以通过Future关停任务or获取返回值

✨异常抛出

虽说，这里submit代码只有仅仅四行，大家乍一看好像没什么明面上的区别了，实际上这里隐藏在一个大秘密

execute不会抛出异常也无法捕获，submit可以捕获到异常

要想彻底了解这一结论，我们还得往后面走，不过现在我们可以了解以下知识：

• 怎么样获取execute异常？
• sumbit怎么捕获异常，他是怎么抛出的？

怎么样获取execute异常？

使用 execute 执行如果发生了异常，是捕获不到的，默认会执行 ThreadGroup 的 uncaughtException 方法

所以如果你想监控执行 execute 方法时发生的异常，需要通过 threadFactory 来指定一个 UncaughtExceptionHandler，这样就会执行上图中的 1，进而执行 UncaughtExceptionHandler 中的逻辑,如下所示:

//1.实现一个自己的线程池工厂
ThreadFactory factory = (Runnable r) -> {//创建一个线程Thread t = new Thread(r);//给创建的线程设置UncaughtExceptionHandler对象 里面实现异常的默认逻辑t.setDefaultUncaughtExceptionHandler((Thread thread1, Throwable e) -> {// 在此设置统计监控逻辑System.out.println("线程工厂设置的exceptionHandler" + e.getMessage());});return t;
};// 2.创建一个自己定义的线程池，使用自己定义的线程工厂
ExecutorService service = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,new LinkedBlockingQueue(10),factory);//3.提交任务
service.execute(()->{int i=1/0;
});

Sumbit怎么捕获异常，他是怎么抛出的？

Future future = executor.submit(myCallable);
try {future.get(3));
} catch (InterruptedException e) {e.printStackTrace();
} catch (ExecutionException e) {e.printStackTrace();
}

在执行sumbit的时候异常会被保存，在get的时候才会进行抛出，请看源码

而在FutureTask使用run方法时，会将异常存入

try {result = c.call();ran = true;
} catch (Throwable ex) {result = null;ran = false;setException(ex);
}
if (ran)set(result);

3，Worker——线程池中的精灵

Worker类

// 此处可以看出 worker 既是一个 Runnable 任务，也实现了 AQS（实际上是用 AQS 实现了一个独占锁，这样由于 worker 运行时会上锁，执行 shutdown，setCorePoolSize，setMaximumPoolSize等方法时会试着中断线程（interruptIdleWorkers） ，在这个方法中断方法中会先尝试获取 worker 的锁，如果不成功，说明 worker 在运行中，此时会先让 worker 执行完任务再关闭 worker 的线程，实现优雅关闭线程的目的）
private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{private static final long serialVersionUID = 6138294804551838833L;// 实际执行任务的线程final Thread thread;// 如果当前线程数少于核心线程数，创建线程并将提交的任务交给 worker 处理处理，此时 firstTask 即为此提交的任务，如果 worker 从 workQueue 中获取任务，则 firstTask 为空Runnable firstTask;// 统计完成的任务数volatile long completedTasks;Worker(Runnable firstTask) {// 初始化为 -1，这样在线程运行前（调用runWorker）禁止中断，在 interruptIfStarted() 方法中会判断 getState()>=0setState(-1); this.firstTask = firstTask;// 根据线程池的 threadFactory 创建一个线程，将 worker 本身传给线程（因为 worker 实现了 Runnable 接口）this.thread = getThreadFactory().newThread(this);}public void run() {// thread 启动后会调用此方法runWorker(this);}// 1 代表被锁住了，0 代表未锁protected boolean isHeldExclusively() {return getState() != 0;}// 尝试获取锁protected boolean tryAcquire(int unused) {// 从这里可以看出它是一个独占锁，因为当获取锁后，cas 设置 state 不可能成功，这里我们也能明白上文中将 state 设置为 -1 的作用，这种情况下永远不可能获取得锁，而 worker 要被中断首先必须获取锁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(); }// 中断线程，这个方法会被 shutdowNow 调用，从中可以看出 shutdownNow 要中断线程不需要获取锁，也就是说如果线程正在运行，照样会给你中断掉，所以一般来说我们不用 shutdowNow 来中断线程，太粗暴了，中断时线程很可能在执行任务，影响任务执行void interruptIfStarted() {Thread t;// 中断也是有条件的，必须是 state >= 0 且 t != null 且线程未被中断// 如果 state == -1 ，不执行中断，再次明白了为啥上文中 setState(-1) 的意义if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}}

为什么Worker使用AQS不使用ReentrantLock

1. 因为ReentrantLock是可重入的锁，为了保证线程在执行任务的时候不被一些事件中断（shutdown，setCorePoolSize），所以采用独占锁

为什么构造Worker时要将state设置为-1

1. 因为tryAcquire是判断state是否为0，在线程刚创建期间还没执行任务，不应该被打断，所以应该将state设置为-1

runWorker——线程！启动！

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// unlock 会调用 tryRelease 方法将 state 设置成 0，代表允许中断，允许中断的条件上文我们在 interruptIfStarted() 中有提过，即 state >= 0w.unlock();boolean completedAbruptly = true;try {// 如果在提交任务时创建了线程，并把任务丢给此线程，则会先执行此 task// 否则从任务队列中获取 task 来执行（即 getTask() 方法）while (task != null || (task = getTask()) != null) {w.lock();// 如果线程池状态为 >= STOP（即 STOP，TIDYING，TERMINATED ）时，则线程应该中断// 如果线程池状态 < STOP, 线程不应该中断,如果中断了（Thread.interrupted() 返回 true，并清除标志位）,再次判断线程池状态（防止在清除标志位时执行了 shutdownNow() 这样的方法）,如果此时线程池为 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) {thrown = x; throw new Error(x);} finally {// 执行任务后，子类可实现此钩子方法作为统计之用afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// 如果执行到这只有两种可能，一种是执行过程中异常中断了，一种是队列里没有任务了，从这里可以看出线程没有核心线程与非核心线程之分，哪个任务异常了或者正常退出了都会执行此方法，此方法会根据情况将线程数-1processWorkerExit(w, completedAbruptly);}
}

代码讲解

这里说明一下第一个if判断，目的是：

• 如果线程池正在停止，那么要保证当前线程是中断状态；
• 如果不是的话，则要保证当前线程不是中断状态；

这里要考虑在执行该if语句期间可能也执行了shutdownNow方法，shutdownNow方法会把状态设置为STOP，回顾一下STOP状态：

不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。在线程池处于 RUNNING 或 SHUTDOWN 状态时，调用 shutdownNow() 方法会使线程池进入到该状态。

STOP状态要中断线程池中的所有线程，而这里使用Thread.interrupted()来判断是否中断是为了确保在RUNNING或者SHUTDOWN状态时线程是非中断状态的，因为Thread.interrupted()方法会复位中断的状态。

总结一下runWorker方法的执行过程：

1. while循环不断地通过getTask()方法获取任务；
2. getTask()方法从阻塞队列中取任务；
3. 如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态；
4. 调用task.run()执行任务；
5. 如果task为null则跳出循环，执行processWorkerExit()方法；
6. runWorker方法执行完毕，也代表着Worker中的run方法执行完毕，销毁线程。

这里的beforeExecute方法和afterExecute方法在ThreadPoolExecutor类中是空的，留给子类来实现。

completedAbruptly变量来表示在执行任务过程中是否出现了异常，在processWorkerExit方法中会对该变量的值进行判断。

核心关键点

从这里可以看出，在线程池中其实没有核心线程和非核心线程这两种类别（这里指没有类别是线程池中并没有区分不同worker之间的类别），不管是核心线程还是非核心线程，他们在执行完任务时都会执行processWorkerExit()方法

processWorkerExit方法——线程正常or异常退出的方法

private void processWorkerExit(Worker w, boolean completedAbruptly) {// 如果异常退出，cas 执行线程池减 1 操作if (completedAbruptly) decrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;// 加锁确保线程安全地移除 worker workers.remove(w);} finally {mainLock.unlock();}// woker 既然异常退出，可能线程池状态变了（如执行 shutdown 等），尝试着关闭线程池tryTerminate();int c = ctl.get();//  如果 woker 是异常退出的，重新新增一个 woker，如果是正常退出的，在 wokerQueue 为非空的条件下，确保至少有一个线程在运行以执行 wokerQueue 中的任务    iif (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false);}
}

整体流程如下：

减少线程数量 -> 移除线程 -> 尝试关闭线程池 -> 如果线程池异常退出则新加线程，如果阻塞队列不为空且是正常退出要保证有一个线程处理阻塞队列

getTask()方法——woker 从 workQueue 中取任务

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// 如果线程池状态至少为 STOP 或者// 线程池状态 == SHUTDOWN 并且任务队列是空的// 则减少线程数量，返回 null,这种情况下上文分析的 runWorker 会执行 processWorkerExit 从而让获取此 Task 的 woker 退出if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// 如果 allowCoreThreadTimeOut 为 true，代表任何线程在 keepAliveTime 时间内处于 idle 状态都会被回收，如果线程数大于 corePoolSize，本身在 keepAliveTime 时间内处于 idle 状态就会被回收boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;/** wc > maximumPoolSize的情况是因为可能在此方法执行阶段同时执行了setMaximumPoolSize方法；* timed && timedOut 如果为true，表示当前操作需要进行超时控制，并且上次从阻塞队列中获取任务发生了超时* 接下来判断，如果有效线程数量大于1，或者阻塞队列是空的，那么尝试将workerCount减1；* 如果减1失败，则返回重试。* 如果wc == 1时，也就说明当前线程是线程池中唯一的一个线程了。*/if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 阻塞获取 task，如果在 keepAliveTime 时间内未获取任务，说明超时了，此时 timedOut 为 trueRunnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}

整体流程如下：

• 判断线程池是否为STOP或者状态为SHUTDOWN但是阻塞队列为空

• 判断以下条件内容，来判断是否要取消本次任务获取

  1. 是否重新设置了maximumPoolSize参数
  2. 线程运作是否已经超时（设置了allowCoreThreadTimeOut 或者 当前线程数大于核心线程数)
  3. 线程数大于1，且阻塞队列已经为空

• 尝试获取任务

  1.如果设置了allowCoreThreadTimeOut 或者 线程池有非核心线程，则该线程会以一个keepAliveTime时间去获取任务

  2.如果线程池只有核心线程，则该线程会作为核心线程去阻塞获取任务

4，addWorker——线程的诞生

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();// 获取线程池的状态int rs = runStateOf(c);/** 这个if判断* 如果rs >= SHUTDOWN，则表示此时不再接收新任务；* 接着判断以下3个条件，只要有1个不满足，则返回false：* 1. rs == SHUTDOWN，这时表示关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务* 2. firsTask为空，不为空代码有新任务要接收，但是SHUTDOWN不在接收新任务* 3. 阻塞队列不为空，阻塞队列也为空了说明没有任务*/if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {// 获取线程数int wc = workerCountOf(c);// 如果超过了线程池的最大 CAPACITY（5 亿多，基本不可能）// 或者 超过了 corePoolSize（core 为 true） 或者 maximumPoolSize（core 为 false） 时// 则返回 falseif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 否则 CAS 增加线程的数量，如果成功跳出双重循环if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctl// 如果线程运行状态发生变化，跳到外层循环继续执行if (runStateOf(c) != rs)continue retry;// 说明是因为 CAS 增加线程数量失败所致，继续执行 retry 的内层循环}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 能执行到这里，说明满足增加 worker 的条件了，所以创建 worker,准备添加进线程池中执行任务w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {// 加锁，是因为下文要把 w 添加进 workers 中， workers 是 HashSet，不是线程安全的，所以需要加锁予以保证final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//  再次 check 线程池的状态以防执行到此步时发生中断等int rs = runStateOf(ctl.get());// 如果线程池状态小于 SHUTDOWN（即为 RUNNING），// 或者状态为 SHUTDOWN 但 firstTask == null（代表不接收任务，只是创建线程处理 workQueue 中的任务），则满足添加 worker 的条件if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {// 如果线程已启动，显然有问题（因为创建 worker 后，还没启动线程呢），抛出异常if (t.isAlive()) throw new IllegalThreadStateException();workers.add(w);int s = workers.size();// 记录最大的线程池大小以作监控之用if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}// 说明往 workers 中添加 worker 成功，此时启动线程if (workerAdded) {t.start();workerStarted = true;}}} finally {// 添加线程失败，执行 addWorkerFailed 方法，主要做了将 worker 从 workers 中移除，减少线程数，并尝试着关闭线程池这样的操作if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}

整体流程如下：

• 检查线程池状态，判断是否需要添加线程
• 尝试增加线程数
• 尝试添加线程

八，线程池的实际应用

这里主要是是我本人的一些应用

① 数据聚合

在实际场景中，我通常会把需要把一些数据聚合起来，返回给用户，例如：商品信息包含了商品价格，商品活动，商品图片，商品库存等等数据，这些数据内容可能有伴随着调用与调用之间的级联，那我们可以选择使用线程池将这些调用封装成任务并行执行，缩短总体时间。

② 批量任务处理

③ 业务拆分，异步优化

对于业务中一些非强一致的流程进行拆分，进行异步优化。

九，线程池引发的一些问题与思考

取自于美团技术博客

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

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

事故原因：该服务展示接口内部逻辑使用线程池做并行计算，由于没有预估好调用的流量，导致最大核心数设置偏小，大量抛出RejectedExecutionException，触发接口降级条件，示意图如下：

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

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

事故原因：该服务处理请求内部逻辑使用线程池做资源隔离，由于队列设置过长，最大线程数设置失效，导致请求数量增加时，大量任务堆积在队列中，任务执行时间过长，最终导致下游服务的大量调用超时失败。示意图如下：

从上面两个案例我们可以得出一个结论，也是使用线程池一直需要考虑的一点——我们到底该如何配置线程池中的参数（阻塞队列大小，核心线程大小，maximum大小）

一些技术方案的平替

十，动态线程池

取自于美团技术博客

如我们上面所说，线程池中的配置一直都是我们要考虑的问题，那为什么我们不能根据线程池当前的状态，对线程池动态变化呢？动态线程池就是为了解决这方面问题

动态化线程池的核心设计包括以下三个方面：

1. 简化线程池配置：线程池构造参数有8个，但是最核心的是3个：corePoolSize、maximumPoolSize，workQueue，它们最大程度地决定了线程池的任务分配和线程分配策略。考虑到在实际应用中我们获取并发性的场景主要是两种：（1）并行执行子任务，提高响应速度。这种情况下，应该使用同步队列，没有什么任务应该被缓存下来，而是应该立即执行。（2）并行执行大批次任务，提升吞吐量。这种情况下，应该使用有界队列，使用队列去缓冲大批量的任务，队列容量必须声明，防止任务无限制堆积。所以线程池只需要提供这三个关键参数的配置，并且提供两种队列的选择，就可以满足绝大多数的业务需求，Less is More。
2. 参数可动态修改：为了解决参数不好配，修改参数成本高等问题。在Java线程池留有高扩展性的基础上，封装线程池，允许线程池监听同步外部的消息，根据消息进行修改配置。将线程池的配置放置在平台侧，允许开发同学简单的查看、修改线程池配置。
3. 增加线程池监控：对某事物缺乏状态的观测，就对其改进无从下手。在线程池执行任务的生命周期添加监控能力，帮助开发同学了解线程池状态。

3.3.2 功能架构

动态化线程池提供如下功能：

• 动态调参：支持线程池参数动态调整、界面化操作；包括修改线程池核心大小、最大核心大小、队列长度等；参数修改后及时生效。
• 任务监控：支持应用粒度、线程池粒度、任务粒度的Transaction监控；可以看到线程池的任务执行情况、最大任务执行时间、平均任务执行时间、95/99线等。
• 负载告警：线程池队列任务积压到一定值的时候会通过大象（美团内部通讯工具）告知应用开发负责人；当线程池负载数达到一定阈值的时候会通过大象告知应用开发负责人。
• 操作监控：创建/修改和删除线程池都会通知到应用的开发负责人。
• 操作日志：可以查看线程池参数的修改记录，谁在什么时候修改了线程池参数、修改前的参数值是什么。
• 权限校验：只有应用开发负责人才能够修改应用的线程池参数。

参数动态化

JDK原生线程池ThreadPoolExecutor提供了如下几个public的setter方法，如下图所示：

JDK允许线程池使用方通过ThreadPoolExecutor的实例来动态设置线程池的核心策略，以setCorePoolSize为方法例，在运行期线程池使用方调用此方法设置corePoolSize之后，线程池会直接覆盖原来的corePoolSize值，并且基于当前值和原始值的比较结果采取不同的处理策略。对于当前值小于当前工作线程数的情况，说明有多余的worker线程，此时会向当前idle的worker线程发起中断请求以实现回收，多余的worker在下次idle的时候也会被回收；对于当前值大于原始值且当前队列中有待执行任务，则线程池会创建新的worker线程来执行队列任务，setCorePoolSize具体流程如下：

线程池内部会处理好当前状态做到平滑修改，其他几个方法限于篇幅，这里不一一介绍。重点是基于这几个public方法，我们只需要维护ThreadPoolExecutor的实例，并且在需要修改的时候拿到实例修改其参数即可。基于以上的思路，我们实现了线程池参数的动态化、线程池参数在管理平台可配置可修改，其效果图如下图所示：

用户可以在管理平台上通过线程池的名字找到指定的线程池，然后对其参数进行修改，保存后会实时生效。目前支持的动态参数包括核心数、最大值、队列长度等。除此之外，在界面中，我们还能看到用户可以配置是否开启告警、队列等待任务告警阈值、活跃度告警等等。关于监控和告警，我们下面一节会对齐进行介绍。

线程池监控

除了参数动态化之外，为了更好地使用线程池，我们需要对线程池的运行状况有感知，比如当前线程池的负载是怎么样的？分配的资源够不够用？任务的执行情况是怎么样的？是长任务还是短任务？

基于对这些问题的思考，动态化线程池提供了多个维度的监控和告警能力，包括：线程池活跃度、任务的执行Transaction（频率、耗时）、Reject异常、线程池内部统计信息等等，既能帮助用户从多个维度分析线程池的使用情况，又能在出现问题第一时间通知到用户，从而避免故障或加速故障恢复。

1. 负载监控和告警

线程池负载关注的核心问题是：基于当前线程池参数分配的资源够不够。对于这个问题，我们可以从事前和事中两个角度来看。事前，线程池定义了“活跃度”这个概念，来让用户在发生Reject异常之前能够感知线程池负载问题，线程池活跃度计算公式为：线程池活跃度 = activeCount/maximumPoolSize。这个公式代表当活跃线程数趋向于maximumPoolSize的时候，代表线程负载趋高。

事中，也可以从两方面来看线程池的过载判定条件，一个是发生了Reject异常，一个是队列中有等待任务（支持定制阈值）。以上两种情况发生了都会触发告警，告警信息会通过大象推送给服务所关联的负责人。

2. 任务级精细化监控

在传统的线程池应用场景中，线程池中的任务执行情况对于用户来说是透明的。比如在一个具体的业务场景中，业务开发申请了一个线程池同时用于执行两种任务，一个是发消息任务、一个是发短信任务，这两类任务实际执行的频率和时长对于用户来说没有一个直观的感受，很可能这两类任务不适合共享一个线程池，但是由于用户无法感知，因此也无从优化。动态化线程池内部实现了任务级别的埋点，且允许为不同的业务任务指定具有业务含义的名称，线程池内部基于这个名称做Transaction打点，基于这个功能，用户可以看到线程池内部任务级别的执行情况，且区分业务，任务监控示意图如下图所示：

3. 运行时状态实时查看，线程实时监控

用户基于JDK原生线程池ThreadPoolExecutor提供的几个public的getter方法，可以读取到当前线程池的运行状态以及参数，如下图所示：

动态化线程池基于这几个接口封装了运行时状态实时查看的功能，用户基于这个功能可以了解线程池的实时状态，比如当前有多少个工作线程，执行了多少个任务，队列中等待的任务数等等。效果如下图所示：

十一，工厂模式构造线程池

因为深入了解了线程池后，对于原先工厂模式构造线程池已经不那么感兴趣了，但是他毕竟也是属于Java线程池中的一部分，所以我还是决定补上

工厂方法构造线程池

• newFixThreadPool 固定大小线程池

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}

1，没有救急线程

2，阻塞队列未设置大小，大小“无限”

3，适用于任务量已知，任务量大的线程

• newCachedThreadPool 带缓冲的线程池

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
}

1，可以无限次创建线程，且核心线程为0，则代表所有线程都是救急线程，且存活时间为60秒

2，SynchronousQueue队列的容量为0，只有在有线程取任务的时候，才能放入任务，相当于一手交钱，一手交货(底层采用Park/unPark实现)

3，适用于任务量庞大，且每个任务时间较短的情况

• newSingleThreadExecutor 单线程的线程池

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}

1，该线程池只有一个线程，用来处理多任务排队处理的情况

和单线程的区别

• 当其中一个线程任务失败异常抛出后，还会创建一个新的线程来执行
• ExecutorService.newSingleThreadExecutor()线程池线程数始终为一，不允许修改

与newFixThreadExecutor(1)不同的是，newSingleThreadExecutor()中的FinalizableDelegatedExecutor，是装饰器模式，只暴露了一些调用方法的接口，不允许更改Executor内部信息，而newFixThreadExecutor返回的是ThreadPoolExecutor，可以通过setCoreCapacity来更改线程数量

• newScheduledThreadPool 定时执行任务线程池

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);
}

与Timer的区别：

1，Timer十分脆弱，当一系列任务中某些任务的时间非常长或者任务出现异常都会影响到后续任务，但是newScheduledThreadPool并不会

shceduleAtiFixedRate与scheduleWithFixedDelay的区别

两个方法都是定时循环执行某个任务，且参数意义都相同

1，shceduleAtiFixedRate 的延时执行时间 取决于 max(任务执行时间，延时时间)

2，scheduleWithFixedDelay 的延时执行时间 取决于 延时时间+任务执行时间

十二，一些问题（随时补充）

• 「阿里 Java 代码规范为什么不允许使用 Executors 快速创建线程池？」

​ 这个和我们之前聊到的无界队列有关系，如果是快速创建线程池，我们可以看到在newCachedThreadPool 方法的最大线程数设置成了 Integer.MAX_VALUE，而 newSingleThreadExecutor 方法创建 workQueue 时 LinkedBlockingQueue 未声明大小，这样就会导致无法创建非核心线程。

改Executor内部信息，而newFixThreadExecutor返回的是ThreadPoolExecutor，可以通过setCoreCapacity来更改线程数量**

• newScheduledThreadPool 定时执行任务线程池

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);
}

与Timer的区别：

1，Timer十分脆弱，当一系列任务中某些任务的时间非常长或者任务出现异常都会影响到后续任务，但是newScheduledThreadPool并不会

shceduleAtiFixedRate与scheduleWithFixedDelay的区别

两个方法都是定时循环执行某个任务，且参数意义都相同

1，shceduleAtiFixedRate 的延时执行时间 取决于 max(任务执行时间，延时时间)

2，scheduleWithFixedDelay 的延时执行时间 取决于 延时时间+任务执行时间

十二，一些问题（随时补充）

• 「阿里 Java 代码规范为什么不允许使用 Executors 快速创建线程池？」

​ 这个和我们之前聊到的无界队列有关系，如果是快速创建线程池，我们可以看到在newCachedThreadPool 方法的最大线程数设置成了 Integer.MAX_VALUE，而 newSingleThreadExecutor 方法创建 workQueue 时 LinkedBlockingQueue 未声明大小，这样就会导致无法创建非核心线程。