文章目录
- 1. 引子
- 2. 线程池源码分析
- 2.1. 总览
- 2.2. Executor
- 2.3. ExecutorService
- 2.4. AbstractExecutorService
- 2.5. ThreadPoolExecutor
- 构造函数核心参数
- 阻塞队列
- 拒绝策略
- 核心属性
- 线程池状态
- Worker 类
- execute() 方法
- addWorker() 方法
- runWorker() 方法
- getTask() 方法
- processWorkerExit() 方法
- 3. 实际问题
本文参考:
文章框架参考: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 {
@Override
public void execute(Runnable command) {
// 直接执行 command 任务,不开启新县城
command.run();
}
}
class ThreadRun implements Executor {
@Override
public void execute(Runnable command) {
// 开启一个线程执行 command 任务
new Thread(command).run();
}
}
2.3. ExecutorService
一般我们定义一个线程池,都是用的这个接口,它额外提供了很多抽象函数方法
ExecutorService threadPool1 = Executors.newFixedThreadPool(1);
public interface ExecutorService extends Executor {
// 关闭线程池,已提交的任务继续执行(在任务队列中),不接受继续提交新任务 状态流转RUNNING -> SHUTDOWN
void shutdown();
// 关闭线程池,尝试停止正在执行的所有任务,不接受继续提交新任务 它和shutdown方法相比,区别在于它会去停止当前正在进行的任务 状态流转(RUNNING or SHUTDOWN) -> STOP
// 返回值是等待执行的任务
List<Runnable> shutdownNow();
// 线程池是否已关闭
boolean isShutdown();
// 如果调用了 shutdown() 或 shutdownNow() 方法后,所有任务结束了,那么返回true
// 必须在调用了 shutdown 和 shutdownNow 以后才会返回 true
boolean 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接口的线程池有两种提交任务的方式了:
- 使用最顶层的 Executor 直接调用 void execute(Runnable command),但是没有返回值
- 使用 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 Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
// 执行任务的真正线程;由线程工厂创建,如果工厂创建失败则为 null
final 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 runWorker
this.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 重新获取 c
c = 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, break
for (;;) {
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,这里重新读取 CTL
c = 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 中的唯一一把 ReentrantLock
final ReentrantLock mainLock = this.mainLock;
// 整个线程池的全局锁,放在 try 外面
// 因为关闭一个线程池需要这个锁,至少我持有锁的期间,线程池不会被关闭
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
// 小于 SHUTTDOWN 那就是 RUNNING
if (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方法的执行过程如下:
- while循环不断地通过getTask()方法获取任务。线程还是复用的上面 t.start() 的 t.
- getTask()方法从阻塞队列中取任务。获取任务的先后顺序与线程启动的先后顺序无关,线程有空闲就可以 getTask()。
- 如果线程池正在停止,那么要保证当前线程是中断状态,否则要保证当前线程不是中断状态。
- 执行传入进来的 Runnable 任务,worker.thread#start -> worker#run -> runWorker(this)
- 如果getTask结果为null则跳出循环,执行processWorkerExit()方法,销毁线程。
执行流程如下图所示:
// 此方法由 worker 线程启动后调用,用一个 while 循环来不断从等待队列中获取任务并执行
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean 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,所以转换为 Error
thrown = 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 >= STOP
if (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 存在,返回 null
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 到 workQueue 中获取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // 超时未获取到,返回 null
workQueue.take(); // corePoolSize 之内的线程一直会阻塞在这里等待任务返回
if (r != null)
return r;
// 这里说明是 poll() 超时了,返回值为 null
timedOut = true;
} catch (InterruptedException retry) {
// 如果此 worker 发生了中断,采取的方案是重试
// 解释下为什么会发生中断,这个要去看 setMaximumPoolSize 方法。
// 如果开发者将 maximumPoolSize 调小了,导致其小于当前的 workers 数量,
// 那么意味着超出的部分线程要被关闭。重新进入 for 循环,自然会有部分线程会返回 null
timedOut = 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 {
// 线程中断,会抛出 InterruptedException
t.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 adjusted
decrementWorkerCount();
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 线程超时了。
示意图如下: