【java并发编程】ThreadPoolExecutor
一什么是线程池线程池（Thread Pool）是一种基于池化思想管理线程的工具，经常出现在多线程服务器中，如MySQL。线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分
一 什么是线程池
线程池（Thread Pool）是一种基于池化思想管理线程的工具，经常出现在多线程服务器中，如MySQL。
线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。
而本文描述线程池是JDK中提供的ThreadPoolExecutor类。
当然，使用线程池可以带来一系列好处：
- 降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
- 提高响应速度：任务到达时，无需等待线程创建即可立即执行。
- 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
- 提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。
二 线程池使用场景
java中经常需要用到多线程来处理一些业务，我们非常不建议单纯使用继承Thread或者实现Runnable接口的方式来创建线程，那样势必有创建及销毁线程耗费资源、线程上下文切换问题。同时创建过多的线程也可能引发资源耗尽的风险，这个时候引入线程池比较合理，方便线程任务的管理。java中涉及到线程池的相关类均在jdk1.5开始的java.util.concurrent包中，涉及到的几个核心类及接口包括：Executor、Executors、ExecutorService、ThreadPoolExecutor、FutureTask、Callable、Runnable等。
2.1场景1：快速响应用户请求
描述：用户发起的实时请求，服务追求响应时间。比如说用户要查看一个商品的信息，那么我们需要将商品维度的一系列信息如商品的价格、优惠、库存、图片等等聚合起来，展示给用户。
分析：从用户体验角度看，这个结果响应的越快越好，如果一个页面半天都刷不出，用户可能就放弃查看这个商品了。而面向用户的功能聚合通常非常复杂，伴随着调用与调用之间的级联、多级级联等情况，业务开发同学往往会选择使用线程池这种简单的方式，将调用封装成任务并行的执行，缩短总体响应时间。另外，使用线程池也是有考量的，这种场景最重要的就是获取最大的响应速度去满足用户，所以应该不设置队列去缓冲并发任务，调高corePoolSize和maxPoolSize去尽可能创造多的线程快速执行任务。
2.2场景2：快速处理批量任务
描述：离线的大量计算任务，需要快速执行。比如说，统计某个报表，需要计算出全国各个门店中有哪些商品有某种属性，用于后续营销策略的分析，那么我们需要查询全国所有门店中的所有商品，并且记录具有某属性的商品，然后快速生成报表。
分析：这种场景需要执行大量的任务，我们也会希望任务执行的越快越好。这种情况下，也应该使用多线程策略，并行计算。但与响应速度优先的场景区别在于，这类场景任务量巨大，并不需要瞬时的完成，而是关注如何使用有限的资源，尽可能在单位时间内处理更多的任务，也就是吞吐量优先的问题。所以应该设置队列去缓冲并发任务，调整合适的corePoolSize去设置处理任务的线程数。在这里，设置的线程数过多可能还会引发线程上下文切换频繁的问题，也会降低处理任务的速度，降低吞吐量。
三 ThreadPoolExecutor线程池的设计
四 线程池的创建及核心参数
Java中创建线程池很简单，只需要调用Executors中相应的便捷方法即可，比如Executors.newFixedThreadPool(int nThreads)，但是便捷不仅隐藏了复杂性，也为我们埋下了潜在的隐患（OOM，线程耗尽）。
2.1 Executors创建线程池方法
| 方法名 | 功能 |
| newFixedThreadPool(int nThreads) | 创建固定大小的线程池 |
| newSingleThreadExecutor() | 创建只有一个线程的线程池 |
| newCachedThreadPool() | 创建一个不限线程数上限的线程池，任何提交的任务都将立即执行 |
创建线程池应该直接使用ThreadPoolExecutor的构造方法。没错，上述Executors方法创建的线程池就是ThreadPoolExecutor。
2.2 ThreadPoolExecutor构造方法
Executors中创建线程池的快捷方法，实际上是调用了ThreadPoolExecutor的构造方法（定时任务使用的是ScheduledThreadPoolExecutor），该类构造方法参数列表如下：
// Java线程池的完整构造函数
public ThreadPoolExecutor(
int corePoolSize, // 线程池长期维持的线程数，即使线程处于Idle状态，也不会回收。
int maximumPoolSize, // 线程数的上限
long keepAliveTime, TimeUnit unit, // 超过corePoolSize的线程的idle时长，
// 超过这个时间，多余的线程会被回收。
BlockingQueue<Runnable> workQueue, // 任务的排队队列
ThreadFactory threadFactory, // 新线程的产生方式
RejectedExecutionHandler handler) // 拒绝策略
)有7个参数，这些参数中，比较容易引起问题的有corePoolSize, maximumPoolSize, workQueue以及handler：
- corePoolSize和maximumPoolSize设置不当会影响效率，甚至耗尽线程；
ThreadPoolExecutor中参数详解
- corePoolSize：核心线程数，也是线程池中常驻的线程数，线程池初始化时默认是没有线程的，当任务来临时才开始创建线程去执行任务
- maximumPoolSize：最大线程数，在核心线程数的基础上可能会额外增加一些非核心线程，需要注意的是只有当workQueue队列填满时才会创建多于corePoolSize的线程(线程池总线程数不超过maxPoolSize)
- keepAliveTime：非核心线程的空闲时间超过keepAliveTime就会被自动终止回收掉，注意当corePoolSize=maxPoolSize时，keepAliveTime参数也就不起作用了(因为不存在非核心线程)；
- unit：keepAliveTime的时间单位
- workQueue：用于保存任务的队列，可以为无界、有界、同步移交三种队列类型之一，当池子里的工作线程数大于corePoolSize时，这时新进来的任务会被放到队列中
- threadFactory：创建线程的工厂类，默认使用Executors.defaultThreadFactory()，也可以使用guava库的ThreadFactoryBuilder来创建
- handler：线程池无法继续接收任务(队列已满且线程数达到maximunPoolSize)时的饱和策略，取值有AbortPolicy、CallerRunsPolicy、DiscardOldestPolicy、DiscardPolicy
任务调度是线程池的主要入口，当用户提交了一个任务，接下来这个任务将如何执行都是由这个阶段决定的。了解这部分就相当于了解了线程池的核心运行机制。
首先，所有任务的调度都是由execute方法完成的，这部分完成的工作是：检查现在线程池的运行状态、运行线程数、运行策略，决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。其执行过程如下：
- 首先检测线程池运行状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING的状态下执行任务。
- 如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。
- 如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
- 如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
- 如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。
其执行流程如下图所示：
举个栗子：现有一个线程池，corePoolSize=10，maxPoolSize=20，队列长度为100，那么当任务过来会先创建10个核心线程数，接下来进来的任务会进入到队列中直到队列满了，会创建额外的线程来执行任务(最多20个线程)，这个时候如果再来任务就会执行拒绝策略.
2.3 workQueue队列
- SynchronousQueue(同步移交队列)：队列不作为任务的缓冲方式，可以简单理解为队列长度为零
- LinkedBlockingQueue(无界队列)：队列长度不受限制，当请求越来越多时(任务处理速度跟不上任务提交速度造成请求堆积)可能导致内存占用过多或OOM
- ArrayBlockintQueue(有界队列)：队列长度受限，当队列满了就需要创建多余的线程来执行任务
2.4 handler拒绝策略
- AbortPolicy：中断抛出异常
- DiscardPolicy：默默丢弃任务，不进行任何通知
- DiscardOldestPolicy：丢弃掉在队列中存在时间最久的任务
- CallerRunsPolicy：让提交任务的线程去执行任务(对比前三种比较友好一丢丢)
五 常见的几种创建线程池方式
5.1 newFixedThreadPool
上图说明：
- 如果当前运行的线程数小于 corePoolSize， 如果再来新任务的话，就创建新的线程来执行任务；
- 当前运行的线程数等于 corePoolSize 后， 如果再来新任务的话，会将任务加入 LinkedBlockingQueue；
- 线程池中的线程执行完 手头的任务后，会在循环中反复从 LinkedBlockingQueue 中获取任务来执行；
为什么不推荐使用FixedThreadPool
FixedThreadPool 使用无界队列 LinkedBlockingQueue（队列的容量为 Integer.MAX_VALUE）作为线程池的工作队列会对线程池带来如下影响 ：
- 当线程池中的线程数达到 corePoolSize 后，新任务将在无界队列中等待，因此线程池中的线程数不会超过 corePoolSize；
- 由于使用无界队列时 maximumPoolSize 将是一个无效参数，因为不可能存在任务队列满的情况。所以，通过创建 FixedThreadPool的源码可以看出创建的 FixedThreadPool 的 corePoolSize 和 maximumPoolSize 被设置为同一个值。
- 由于 1 和 2，使用无界队列时 keepAliveTime 将是一个无效参数；
- 运行中的 FixedThreadPool（未执行 shutdown()或 shutdownNow()）不会拒绝任务，在任务比较多的时候会导致 OOM（内存溢出）。
5.2 newSingleThreadExector
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
}
上图说明 :
- 如果当前运行的线程数少于 corePoolSize，则创建一个新的线程执行任务；
- 当前线程池中有一个运行的线程后，将任务加入 LinkedBlockingQueue
- 线程执行完当前的任务后，会在循环中反复从LinkedBlockingQueue 中获取任务来执行；
为什么不推荐使用SingleThreadExecutor
SingleThreadExecutor 使用无界队列 LinkedBlockingQueue 作为线程池的工作队列（队列的容量为 Intger.MAX_VALUE）。SingleThreadExecutor 使用无界队列作为线程池的工作队列会对线程池带来的影响与 FixedThreadPool 相同。说简单点就是可能会导致 OOM，
5.3 newCachedThreadPool
CachedThreadPool 的corePoolSize 被设置为空（0），maximumPoolSize被设置为 Integer.MAX.VALUE，即它是无界的，这也就意味着如果主线程提交任务的速度高于 maximumPool 中线程处理任务的速度时，CachedThreadPool 会不断创建新的线程。极端情况下，这样会导致耗尽 cpu 和内存资源。
上图说明：
- 首先执行 SynchronousQueue.offer(Runnable task) 提交任务到任务队列。如果当前 maximumPool 中有闲线程正在执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)，那么主线程执行 offer 操作与空闲线程执行的 poll 操作配对成功，主线程把任务交给空闲线程执行，execute()方法执行完成，否则执行下面的步骤 2；
- 当初始 maximumPool 为空，或者 maximumPool 中没有空闲线程时，将没有线程执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)。这种情况下，步骤 1 将失败，此时 CachedThreadPool 会创建新线程执行任务，execute 方法执行完成；
为什么不推荐使用CachedThreadPool
CachedThreadPool允许创建的线程数量为 Integer.MAX_VALUE ，可能会创建大量线程，从而导致 OOM。
5.4 newScheduledThreadPool
ScheduledThreadPoolExecutor 的执行主要分为两大部分：
- 当调用 ScheduledThreadPoolExecutor 的 scheduleAtFixedRate() 方法
或者 scheduleWithFixedDelay() 方法时，会向 ScheduledThreadPoolExecutor 的 DelayQueue 添加一个实现了 RunnableScheduledFuture 接口的 ScheduledFutureTask 。
- 线程池中的线程从 DelayQueue 中获取 ScheduledFutureTask，然后执行任务。
ScheduledThreadPoolExecutor 为了实现周期性的执行任务，对 ThreadPoolExecutor做了如下修改：
- 使用 DelayQueue 作为任务队列；
- 获取任务的方不同
- 执行周期任务后，增加了额外的处理
- 线程 1 从 DelayQueue 中获取已到期的 ScheduledFutureTask（DelayQueue.take()）。到期任务是指 ScheduledFutureTask的 time 大于等于当前系统的时间；
- 线程 1 执行这个 ScheduledFutureTask；
- 线程 1 修改 ScheduledFutureTask 的 time 变量为下次将要被执行的时间；
- 线程 1 把这个修改 time 之后的 ScheduledFutureTask 放回 DelayQueue 中（DelayQueue.add())。
使用的构造方式为new ThreadPoolExecutor(var1, 2147483647, 0L, TimeUnit.NANOSECONDS, new ScheduledThreadPoolExecutor.DelayedWorkQueue())，支持定时周期性执行，注意一下使用的是延迟队列，弊端同newCachedThreadPool一致
5.5 关闭线程池
- shutdownNow()：立即关闭线程池(暴力)，正在执行中的及队列中的任务会被中断，同时该方法会返回被中断的队列中的任务列表
- shutdown()：平滑关闭线程池，正在执行中的及队列中的任务能执行完成，后续进来的任务会被执行拒绝策略
- isTerminated()：当正在执行的任务及对列中的任务全部都执行（清空）完就会返回true
5.6 小结
所以根据上面分析我们可以看到，FixedThreadPool和SigleThreadExecutor中之所以用LinkedBlockingQueue无界队列，是因为设置了corePoolSize=maxPoolSize，线程数无法动态扩展，于是就设置了无界阻塞队列来应对不可知的任务量；而CachedThreadPool则使用的是SynchronousQueue同步移交队列，为什么使用这个队列呢？因为CachedThreadPool设置了corePoolSize=0，maxPoolSize=Integer.MAX_VALUE，来一个任务就创建一个线程来执行任务，用不到队列来存储任务；SchduledThreadPool用的是延迟队列DelayedWorkQueue。在实际项目开发中也是推荐使用手动创建线程池的方式，而不用默认方式，关于这点在《阿里巴巴开发规范》中是这样描述的：
六 类结构 & 核心代码
6.1 任务执行流程
线程池的工作线程通过Woker类实现，在ReentrantLock锁的保证下，把Woker实例插入到HashSet后，并启动Woker中的线程。 从Woker类的构造方法实现可以发现: 线程工厂在创建线程thread时，将Woker实例本身this作为参数传入，当执行start方法启动线程thread时，本质是执行了Worker的runWorker方法。 firstTask执行完成之后，通过getTask方法从阻塞队列中获取等待的任务，如果队列中没有任务，getTask方法会被阻塞并挂起，不会占用cpu资源；
execute()方法
//队列就是我们自定义的同步队列
private final BlockingQueue<Runnable> workQueue;
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();
if (workerCountOf(c) < corePoolSize) {
//workerCountOf获取线程池的当前线程数；小于corePoolSize，执行addWorker创建新线程执行command任务
if (addWorker(command, true))
return;
c = ctl.get();
}
// double check: c, recheck
// 线程池处于RUNNING状态，把提交的任务成功放入阻塞队列中
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// recheck and if necessary 回滚到入队操作前，即倘若线程池shutdown状态，就remove(command)
//如果线程池没有RUNNING，成功从阻塞队列中删除任务，执行reject方法处理任务
if (! isRunning(recheck) && remove(command))
reject(command);
//线程池处于running状态，但是没有线程 false表示创建非核心线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false); //未生成task且队列中有的会优先执行队列中的
}
// 往线程池中创建新的线程失败，则reject任务
else if (!addWorker(command, false))
reject(command);
}
- 为什么需要double check线程池的状态?
在多线程环境下，线程池的状态时刻在变化，而ctl.get()是非原子操作，很有可能刚获取了线程池状态后线程池状态就改变了。判断是否将command加入workque是线程池之前的状态。倘若没有double check，万一线程状态不正常，会导致队列中的任务永远无法执行了。这里也加了一个线程池线程数量的检查，如果不做检查会导致队列中还有任务，没有线程处理，所以对应的添加一个空任务的最大线程来处理任务。
addWorker 方法
private final ReentrantLock mainLock = new ReentrantLock();
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
//每次for循环都需要获取最新的ctl值
int c = ctl.get();
//获取当前线程池状态
int rs = runStateOf(c);
//线程池还在跑或者线程池关闭了队列中还有任务就可以继续往下创建线程否则创建失败
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
//获取工作线程
int wc = workerCountOf(c);
/**
*校验线程池已有线程数量是否超限
*1.工作线程是否大于最大上限CAPACITY
*2.或者创建的是核心线程且工作线程数>=corePoolSize
*3.或者创建的是非核心线程且工作线程数>=maximumPoolSize
*步骤2、3中工作线程数大于什么，取决于入参core
*/
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//使用CAS的方法给ctl的worker的数量加1,成功则跳出最外层循环
if (compareAndIncrementWorkerCount(c))
break retry;
//CAS不成功则重新获取ctl的值
c = ctl.get();
//如果CAS不成功的原因是状态变了则重新进行外层循环
if (runStateOf(c) != rs)
continue retry;
}
}
//workerStarted表示woker是否被执行
boolean workerStarted = false;
//workerAdded表示worker是否成功添加到workers
boolean workerAdded = false;
Worker w = null;
try {
//创建一个新的工作线程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//先持有锁，再创建线程
final ReentrantLock mainLock = this.mainLock;
//可重入锁加锁 这个是为了防止，线程shutdown了我们还往里面添加线程
mainLock.lock();
try {
//获取线程池的工作状态
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//检查线程是否已经已启动
if (t.isAlive())
throw new IllegalThreadStateException();
//把worker添加到workers里面
workers.add(w);
//工作线程数
int s = workers.size();
//如果工作线程数s大于largestPoolSize,设置largestPoolSize为s
if (s > largestPoolSize)
largestPoolSize = s;
//设置workerAdded为true
workerAdded = true;
}
} finally {
//可重入锁解锁
mainLock.unlock();
}
//如果workerAdded为true,启动线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//如果工作线程启动失败，则删除此工作线程
if (! workerStarted)
addWorkerFailed(w);
}
//返回工作线程启动结果
return workerStarted;
}
在第一部分中，首先还是从类型为AtomicInteger的ctl中获得最新值，然后调用runStateOf来获得当前线程池的运行状态。
下面的if判断，这块我们其实需要以相反的判断去思考——也就是，如果if中的判断是false，那么就不用return false就不往下执行了。那么，什么情况下，程序不往下执行了呢？
if (rs >= SHUTDOWN &&
!(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
（1）、case1：线程状态终止。
（2）、case2：!(在线程终止，且当线程没有任务，任务队列不为空的时候
case1 &&! case2 那这里主要是说程池未关闭，或者线程池已关闭，但是队列中还有任务的时候，要允许创建线程往下执行，否则不允许创建线程。
代码在往下，就是一个无限的循环了，在循环里我们发现，如果线程池中的线程数等于或者超过了最大线程数量（CAPACITY），或者已经等于或者超越了核心线程数（corePoolSize）/最大线程数（maximumPoolSize），那么就会直接返回false，没有“机会”继续下去了。而具体是对比corePoolSize还是maximumPoolSize，是根据我们addWorker方法第二个入参boolean core来决定的。
跳过这层阻碍，我们就要真正的去给当前的工作线程数加1了。这次依然采用的是CAS的方式去加1。如果加一成功，则Part1执行完毕，跳出循环，开始Part2的旅程。如果加一不成功，说明与其他线程操作冲突了，那么会重新获取最新的ctl值，再次循环执行上面的步骤。
worker 类
private final class Worker
extends AbstractQueuedSynchronizer
implements 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) {
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);
}
// 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) {
}
}
}
}runWorker 核心方法
通过getTask方法从阻塞队列中获取等待的任务，如果队列中没有任务，getTask方法会被阻塞并挂起，不会占用cpu资源；
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，再从workerQueue中取task(getTask())
while (task != null || (task = getTask()) != null) {
w.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
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 {
processWorkerExit(w, completedAbruptly);
}
}
从runWorker的代码逻辑中，我们能够看出来，它的主要处理逻辑就是执行外部指定的firstTask或者从阻塞队列中获得待执行的任务，然后调用run方法进行执行。这里我们可以思考一下是最大线程先执行还是队列中的先执行 ？
- 执行优先级 是 核心线程>非核心线程>队列
- 提交优先级 是 核心线程>队列>非核心线程
同时也与Spring类似，提供了beforeExecute和afterExecute的前置和后置的处理。默认都是空的方法，如果我们需要，可以对其进行继承实现。
那这个时候，就会有同学有疑问了，在介绍线程池构造方法入参的时候，不是有两个参数，是用来控制线程最大空闲时间的吗（keepAliveTime和unit），也就是说，超过这个时间，线程就消亡了。那么，怎么在源码中没有发现在哪呢？其实，他们都在获取队列中的任务getTask()方法中呢。我们下面将会对这个方法进行解析。
getTask()方法
下面来看一下getTask()方法，这里面涉及到keepAliveTime的使用，从这个方法我们可以看出线程池是怎么让超过corePoolSize的那部分worker销毁的。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
/** 对线程池状态的判断，两种情况会 workerCount-1，并且返回 null
1. 线程池状态为 shutdown，且 workQueue 为空（反映了 shutdown 状态的线程池还是
要执行 workQueue 中剩余的任务的）
2. 线程池状态为 stop（shutdownNow()会导致变成 STOP）（此时不用考虑 workQueue
的情况）
**/
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/ timed 变量用于判断是否需要进行超时控制。
// allowCoreThreadTimeOut 默认是 false，也就是核心线程不允许进行超时；
// wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；
// 对于超过核心线程数量的这些线程，需要进行超时控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/**
1. 线程数量超过 maximumPoolSize 可能是线程池在运行时被调用了 setMaximumPoolSize()
被改变了大小，否则已经 addWorker()成功不会超过 maximumPoolSize
2. timed && timedOut 如果为 true，表示当前操作需要进行超时控制，并且上次从阻塞队列中
获取任务发生了超时.其实就是体现了空闲线程的存活时间
**/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
/**
根据 timed 来判断，如果为 true，则通过阻塞队列 poll 方法进行超时控制，如果在
keepaliveTime 时间内没有获取到任务，则返回 null.否则通过 take 方法阻塞式获
取. 队列中的任务
**/
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true; /* 如果 r==null，说明已经超时了,队列中仍然没有任务，此时设置timedOut=true，在下次自旋的时候进行回收*/
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
allowCoreThreadTimeOut为false，核心线程即使空闲也不会被销毁；倘若为ture，在keepAliveTime内核心线程仍空闲则会被销毁。
如果线程允许空闲等待而不被销毁timed == false，workQueue.take任务: 如果阻塞队列为空，当前线程会被挂起等待；当队列中有任务加入时，线程被唤醒，take方法返回任务，并执行；
如果线程不允许无休止空闲timed == true, workQueue.poll任务: 如果在keepAliveTime时间内，阻塞队列还是没有任务，则返回null；
submit() 方法
AbstractExecutorService.submit()实现了ExecutorService.submit() 可以获取执行完的返回值, 而ThreadPoolExecutor 是AbstractExecutorService.submit()的子类，所以submit方法也是ThreadPoolExecutor`的方法。
// submit方法在AbstractExecutorService中的实现
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。通过Executor.execute方法提交FutureTask到线程池中等待被执行，最终执行的是FutureTask的run方法；但是这里要注意如果只是提交个异步执行的线程不需要返回结果尽量用execute()方法，如果需要异步结果可以使用submit()方法，因为如果不需要返回结果，仅仅只是异步调用了submit()以后异常是得不到有效处理的。
¶FutureTask
public class FutureTask<V> implements RunnableFuture<V> 可以将FutureTask提交至线程池中等待被执行(通过FutureTask的run方法来执行)
- 内部状态
/* The run state of this task, initially NEW.
* ...
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
内部状态的修改通过sun.misc.Unsafe修改
- get方法
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
//阻赛等待任务执行完成
s = awaitDone(false, 0L);
//返回结果或者抛出异常
return report(s);
}
FutureTask 通过get()方法获取任务执行结果。如果任务处于未完成的状态(state <= COMPLETING)，就调用awaitDone方法(后面单独讲解)等待任务完成。任务完成后，通过report方法获取执行结果或抛出执行期间的异常
//返回执行结果或抛出异常
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}
内部通过awaitDone方法对主线程进行阻塞，具体实现如下:
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
- 如果主线程被中断，则抛出中断异常；
- 判断FutureTask当前的state，如果大于COMPLETING，说明任务已经执行完成，则直接返回；
- 如果当前state等于COMPLETING，说明任务已经执行完，这时主线程只需通过yield方法让出cpu资源，等待state变成NORMAL；
- 通过WaitNode类封装当前线程，并通过UNSAFE添加到waiters链表；
- 最终通过LockSupport的park或parkNanos挂起线程；
removeWaiter() 把等待的线程移出队列
private void removeWaiter(WaitNode node) {
if (node != null) {
node.thread = null;//首先置空线程
retry:
for (;;) { // restart on removeWaiter race
//依次遍历查找
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
s = q.next;
if (q.thread != null)
pred = q;
else if (pred != null) {
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,q, s)) //cas替换
continue retry;
}
break;
}
}
}
如果线程被中断，首先清除中断状态，调用removeWaiter移除等待节点，然后抛出InterruptedException
run方法
public void run() {
if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
首先利用cas修改state状态为COMPLETING，设置返回结果，然后使用 lazySet(UNSAFE.putOrderedInt)的方式设置state状态为NORMAL。结果设置完毕后，调用finishCompletion()方法唤醒等待线程
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}七 如何设置线程个数
- 类比于现实世界中的人类通过合作做某件事情，我们可以肯定的一点是线程池大小设置过大或者过小都会有问题，合适的才是最好。
- 如果我们设置的线程池数量太小的话，如果同一时间有大量任务/请求需要处理，可能会导致大量的请求/任务在任务队列中排队等待执行，甚至会出现任务队列满了之后任务/请求无法处理的情况，或者大量任务堆积在任务队列导致 OOM。这样很明显是有问题的！ CPU 根本没有得到充分利用。
- 但是，如果我们设置线程数量太大，大量线程可能会同时在争取 CPU 资源，这样会导致大量的上下文切换，从而增加线程的执行时间，影响了整体执行效率。
2.1 理论上公式
CPU 密集型任务(N+1)
这种任务消耗的主要是 CPU 资源，可以将线程数设置为 N（CPU 核心数）+1，比 CPU 核心数多出来的一个线程是为了防止线程偶发的缺页中断，或者其它原因导致的任务暂停而带来的影响。一旦任务暂停，CPU 就会处于空闲状态，而在这种情况下多出来的一个线程就可以充分利用 CPU 的空闲时间。
I/O 密集型任务(2N)
这种任务应用起来，系统会用大部分的时间来处理 I/O 交互，而线程在处理 I/O 的时间段内不会占用 CPU 来处理，这时就可以将 CPU 交出给其它线程使用。因此在 I/O 密集型任务的应用中，我们可以多配置一些线程，具体的计算方法是 2N。
2.2判断CPU&IO 密集
CPU 密集型简单理解就是利用 CPU 计算能力的任务比如你在内存中对大量数据进行排序。但凡涉及到网络读取，文件读取这类都是 IO 密集型，这类任务的特点是 CPU 计算耗费时间相比于等待 IO 操作完成的时间来说很少，大部分时间都花在了等待 IO 操作完成上.在我们的项目实际实践过程中更多的时候我们需要读写数据库,调用远程接口.所以我们日常碰到的场景更多的是IO密集型的系统.
2.3 实际配置
理论上是没问题，但是到底要设置多少个线程合适，还得看自己的业务使用量，所以最好的处理方式便是监控与动态跟进，那么要怎么监控与动态跟进呢，这里实际上可以自己操作活着使用开源的平台类似于 接入流程 | Hippo4j 等的一些监控与动态处理手段
开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！
更多推荐
2
0- 0
已为社区贡献4条内容
所有评论(0)