【Java 并发】ThreadPool
1 为什么使用线程池
在实际使用中, 线程是很占用系统资源的, 如果对线程管理不善很容易导致系统问题。
因此, 在大多数并发框架中都会使用线程池来管理线程, 使用线程池管理线程主要有如下好处:
- 降低资源消耗。通过复用已存在的线程和降低线程关闭的次数来尽可能降低系统性能损耗
- 提升系统响应速度。通过复用线程, 省去创建线程的过程, 因此整体上提升了系统的响应速度
- 提高线程的可管理性。线程是稀缺资源, 如果无限制的创建, 不仅会消耗系统资源, 还会降低系统的稳定性, 因此, 需要使用线程池来管理线程。
2 线程池添加任务的流程
上面是向线程池中添加任务的流程图:
- 判断线程池的线程个数是否大于核心线程池个数, 如果不是, 则新创建一个线程执行刚提交的任务, 否则进入第 2 步
- 判断当前阻塞队列是否已满, 如果未满, 则将提交的任务放置在阻塞队列中, 否则进入第 3 步
- 判断线程池的线程个数是否大于最大线程个数, 如果没有, 则创建一个新的线程来执行任务, 否则进入第 4 步
- 按照配置的拒绝策略 (没有配置有默认值) 进行处理
3 创建线程池
3.1 创建线程池的几个重要参数
从上面的流程图可以知道向线程池添加任务的流程, 从中可以提取出几个关键的参数
- 核心线程个数
- 最大线程个数
- 阻塞队列
- 拒绝策略
线程池中的线程数个数是逐渐增长的, 不是一开始就创建好的。
线程个数先从 0 增长到核心线程个数, 核心个数的线程一旦达到了就不会增加了。
后续如果核心线程不够处理任务了, 线程数还会继续增加, 一直达到最大线程个数。
和核心线程数不同的是, 不是核心线程的这些线程在空闲的时候, 会销毁的。
所以在创建线程池除了需要指定上面的 4 个参数, 还有指定另外 3 个
- 线程空闲多长时间进行销毁
- 线程空闲多长时间的单位: 秒, 分钟等
- 线程工厂, 用于线程池中创建线程
3.2 线程池的定义
从 Java 线程池 Executor 框架体系可以看出 ThreadPoolExecutor 是线程池的定义。
3.2 ThreadPoolExecutor 的构造函数
public class ThreadPoolExecutor extends AbstractExecutorService {
// 指定 核心线程数 最大线程数 存活时间 存活时间单位 阻塞队列
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
//调用自身 7 个参数的构造函数, 默认工厂 Executors.DefaultThreadFactory 默认拒绝策略 AbortPolicy
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler);
}
// 指定 核心线程数 最大线程数 存活时间 存活时间单位 阻塞队列 线程工厂
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
//调用自身 7 个参数的构造函数, 默认拒绝策略 AbortPolicy
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);
}
// 指定 核心线程数 最大线程数 存活时间 存活时间单位 阻塞队列 拒绝策略
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
//调用自身 7 个参数的构造函数, 默认工厂 Executors.DefaultThreadFactory
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
}
// 指定 核心线程数 最大线程数 存活时间 存活时间单位 阻塞队列 线程工厂 拒绝策略
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
// 核心线程, 最大线程 存活时间 必须大于 0, 最大线程数必须大于等于核心线程数
if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0)
throw new IllegalArgumentException();
// 阻塞队列 线程工厂 拒绝策略 不能为空
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ? null : AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
}
corePoolSize
核心线程个数。 当提交一个任务时, 如果当前核心线程池的线程个数没有达到corePoolSize, 则会创建新的线程来执行所提交的任务, 即使当前核心线程池有空闲的线程。
如果当前核心线程的个数已经达到了 corePoolSize, 则不再重新创建线程。
maximumPoolSize
线程池能创建线程的最大个数。 如果当阻塞队列已满时, 并且当前线程池线程个数没有超过 maximumPoolSize 的话, 就会创建新的线程来执行任务。
keepAliveTime
空闲线程存活时间。如果当前线程池的线程个数已经超过了 corePoolSize, 并且线程空闲时间超过了 keepAliveTime 的话, 就会将这些空闲的非核心线程销毁, 这样可以尽可能降低系统资源消耗。
线程池中维护了一个变量 allowCoreThreadTimeOut, 这个字段决定了能回收的线程是否包含核心线程。
allowCoreThreadTimeOut, 默认为 false, 表示线程回收只针对非核心线程数。
allowCoreThreadTimeOut, 为 true, 则核心线程超过了空闲时间, 也会被回收, 但是这种做法违背了线程池的理念: 线程复用。
可以通过 线程池的 allowsCoreThreadTimeOut() 方法进行设置。
unit
时间单位。为 keepAliveTime 指定时间单位。
workQueue
阻塞队列。用于保存任务的阻塞队列, 可以使用 ArrayBlockingQueue, LinkedBlockingQueue, SynchronousQueue, PriorityBlockingQueue 等。
threadFactory
创建线程的工程类。可以通过指定线程工厂为每个创建出来的线程设置更有意义的名字, 如果出现并发问题, 也方便查找问题原因。
handler
拒绝策略。 当线程池的阻塞队列已满和指定的线程都已经开启, 说明当前线程池已经处于饱和状态了, 那么就需要采用一种策略来处理这种情况。
3.3 阻塞队列
创建 ThreadPoolExecutor 时需要指定一个阻塞队列。官方推荐的有 3 个
1. SynchronousQueue 同步队列, 内部没有任何容量的阻塞队列, 任何一次插入操作的元素都要等待另一边执行删除/读取操作, 否则进行插入操作的线程就要一直等待。
2. LinkedBlockingQueue 基于链表结构实现的无界限 (理论上的无界限, 最大值为 Integer.MAX_VALUE) 的队列。 因为队列是无限的, 不会达到上限, 所以不会触发线程达到最大线程数的情况, 所以的线程都是核心线程。
这种队列可以提高线程池吞吐量, 但代价是牺牲内存空间, 甚至会导致内存溢出。 LinkedBlockingQueue 可以在声明的时候指定容量, 指定了容量的话, 就会变成一个有限队列
3. ArrayBlockingQueue 基于数组实现的有限队列。 这种有界队列有利于防止资源耗尽, 但可能更难调整和控制
Java 还提供了另外 4 种队列可以选择
1. PriorityBlockingQueue 支持优先级排序的无界阻塞队列。存储在里面的元素要么实现了 Comparable 接口或者在内部指定了比较器 Comparator, 才能对里面的元素进行比较。PriorityBlockingQueue 只保证优先级最高的元素始终排在队列的头部,
但是不保证优先级一样的元素的顺序, 也不保证除了优先级最高的元素以外的元素, 都能处于正确排序的位置
2. DelayQueue 基于二叉堆实现的延迟队列, 时间没到任务取不出来。同时具备: 无界队列、阻塞队列、优先队列的特征。存储在里面的元素需要实现 Delayed 接口。
3. LinkedBlockingDeque 双端队列。基于链表实现, 既可以从尾部插入/取出元素, 还可以从头部插入元素/取出元素
4. LinkedTransferQueue 由链表结构组成的无界阻塞队列。这个队列比较特别的时, 采用一种预占模式。消费者线程取元素时, 如果队列不为空, 直接获取。为空, 生成一个节点 (节点元素为 null) 入队, 消费者线程被等待在这个节点上, 后面生产者线程入队时发现有一个元素为null的节点, 生产者线程就不入队了, 直接就将元素填充到该节点, 并唤醒该节点等待的线程, 被唤醒的消费者线程取走元素。
3.4 拒绝策略
AbortPolicy
直接拒绝所提交的任务, 并抛出 RejectedExecutionException 异常
CallerRunsPolicy
只用调用者所在的线程来执行任务
DiscardPolicy
不处理直接丢弃掉任务
DiscardOldestPolicy
丢弃掉阻塞队列中存放时间最久的任务, 执行当前任务
4 线程池状态
线程池有 5 种状态
public static final int SIZE = 32;
// Integer.SIZE = 32, count_bits = 32 - 3 = 29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 当创建线程池后, 初始时, 线程池处于 RUNNING 状态
// -536870912, 二进制 11100000 00000000 00000000 00000000, 最高位是 111 表示运行中
private static final int RUNNING = -1 << COUNT_BITS;
// 如果调用了 shutdown() 方法, 则线程池处于 SHUTDOWN 状态, 此时线程池不能够接受新的任务, 它会等待所有任务执行完毕
// 0, 二进制 00000000 00000000 00000000 00000000, 最高位是 000 表示关闭
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 如果调用了shutdownNow() 方法, 则线程池处于 STOP 状态, 此时线程池不能接受新的任务, 并且会去尝试终止正在执行的任务
// 536870912, 二进制 00100000 00000000 00000000 00000000, 最高位是 001 表示停止
private static final int STOP = 1 << COUNT_BITS;
// 线程池中的所有任务都已终止, 则会变为, 都是从 stop / shutdown 转为 tidying
// 1073741824, 二进制 01000000 00000000 00000000 00000000, 最高位是 010 表示终止前的事项都准备完成
private static final int TIDYING = 2 << COUNT_BITS;
// 线程池彻底终止
// 1610612736, 二进制 01100000 00000000 00000000 00000000, 最高位是 010 表示最终的停止
private static final int TERMINATED = 3 << COUNT_BITS;
线程池状态的切换过程:
5 线程池的使用
5.1 自定义线程池 ThreadPoolExecutor
// 自定义线程工厂类
public static class MyThreadFactory implements ThreadFactory {
private final AtomicInteger threadNumber = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread("My-Thread" + threadNumber.getAndIncrement());
}
int corePoolSize = 5;
int maximumPoolSize = 10;
long keepAliveTime = 1000L;
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue(20);
// 创建线程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, workQueue, new MyThreadFactory(), new AbortPolicy());
// 自定义 Runable 任务
Runnable task = () -> System.out.println("Finish");
// 提交任务
threadPool.execute(task);
// 关闭线程池
threadPool.shutdown();
}
5.2 官方提供的线程池工具类 Executors
Exceutors 内部提供了几种常用的线程池配置
| 线程池名 | 核心线程数 | 最大线程数 | 线程存活时间 (单位: 毫秒) | 阻塞队列类型 |
|---|---|---|---|---|
| FixedThreadPool (固定容量线程池) | 用户配置 | 等于核心线程数 | 0 | LinkedBlockingQueue |
| CachedThreadPool (缓存线程池) | 0 | Integer.MAX_VALUE | 60 | SynchronousQueue |
| SingleThreadExecutor (单线程线程池) | 1 | 1 | 0 | LinkedBlockingQueue |
| ScheduledThreadPool (定时线程池) | 用户配置 | Integer.MAX_VALUE | 0 (内部配置的单位是纳秒) | DelayedWorkQueue |
上面是比较常见的 4 个线程池, 在 JDK 8 新增了一个工作窃取线程池
3.3.2.1 FixedThreadPool
固定容量线程池。 其特点
- 最大线程数等于核心线程数
- 阻塞队列是理论没有界限的 LinkedBlockQueue
基于上面 2 个的配置, 可以得到 FixedThreadPool 的特点:
任务提交后, 会不断创建核心线程, 核心线程个数达到了配置的上限了, 任务会进入到无限的阻塞队列中, 永远不会创建出非核心线程。
适用于并发比较高的场景
3.3.2.2 CachedThreadPool
缓存线程池。其特点
- 核心线程是为 0, 最大线程数为 Integer.MAX_VALUE, 理论无上限
- 空闲线程存活时间 60 毫秒
- 阻塞队列是没有任何容量的 SynchronousQueue
基于上面 3 个配置, 可以得到 CachedThreadPool 的特点:
任务提交后, 就不断创建非核心线程数, 或者交给空闲的非核心线程处理, 中间没有队列缓存, 有任务过来必定有线程立即进行处理, 一旦空闲线程时间超过 60 毫秒, 就对其进行销毁。
适用于任务量大但耗时低的场景
3.3.2.3 SingleThreadExecutor
单线程线程池。其特点
- 核心线程数和最大线程数都是 1
- 阻塞队列是理论没有界限的 LinkedBlockQueue
基于上面 3 个配置, 可以得到 SingleThreadExecutor 的特点:
只会有一个核心线程在处理提交的任务, 提交的任务比消费的快时, 任务会缓存在理论无容量上限的阻塞队列中。
适用于消费效率比任务产生快或者生产端生产效率慢但是任务耗时高的情景
3.3.2.4 ScheduledThreadPool
定时线程池。其特点
- 最大线程数理论无限
- keepAliveTime 时间为 0, 直接过期
- 阻塞队列是延迟队列
基于上面 3 个配置, 可以到的 ScheduledThreadPool 的特点:
任务提交后, 延迟处理。在处理中, 和正常的线程池差不多, 优先创建核心线程数, 存入阻塞队列, 创建非核心线程数, 不同的是非核心线程数空闲了, 就立即进行销毁。
适用于执行定时或周期性的任务
3.3.2.5 WorkStealingPool
WorkStealPool 是一个拥有多个任务队列的线程池, 空闲的线程可以帮助其他的线程处理任务。
假设共有两个线程同时执行, A, B
假如 A 线程中的队列里面分配了 5 个任务, 而 B 线程的队列中分配了 1 个任务, 当 B 线程执行完任务后, 它会主动的去 A 线程中窃取其他的任务进行执行。
具体的实现是基于一个新的线程池 ForkJoinPool, 有兴趣的可以自行了解一下, 这里不展开了。
5.3 定时线程池
在 Java 中要实现定时任务的话, 可以使用到哪些现成的类呢?
- Timer
- ScheduledThreadPoolExecutor
2 者都能达到周期性执行任务和给定时间延迟执行任务的功能。不同的是 Timer 只能使用一个后台线程执行任务, 而 ScheduledThreadPoolExecutor 则可以通过构造函数来指定后台线程的个数, 能达到这种效果是因为
ScheduledThreadPoolExecutor 本身就是一个线程池的实现。
ScheduledThreadPoolExecutor 类的 UML 图如下:
从 UML图 可以看出, ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor, 具备了线程的 execute, submit 等提交异步任务的基础功能。
同时还实现了 ScheduledExecutorService, 该接口定义了 ScheduledThreadPoolExecutor 能够延时执行任务和周期执行任务的功能的 4 个接口方法
public interface ScheduledExecutorService extends ExecutorService {
// 达到给定的延时时间后, 执行任务。这里传入的是实现 Runnable 接口的任务, 因此通过 ScheduledFuture.get() 获取结果为 null
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
// 达到给定的延时时间后, 执行任务。这里传入的是实现Callable接口的任务, 因此, 返回的是任务的最终计算结果
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
// 以上一个任务开始的时间计时, period 时间过去后, 检测上一个任务是否执行完毕, 如果上一个任务执行完毕
// 则当前任务立即执行, 如果上一个任务没有执行完毕, 则需要等上一个任务执行完毕后立即执行
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
// 当达到延时时间 initialDelay 后, 任务开始执行。上一个任务执行结束后到下一次, 任务执行, 中间延时时间间隔为 delay。 以这种方式, 周期性执行任务
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
}
5.4 核心线程池的立即初始
声明了线程池后, 这时候实际是没有线程的, 单提交了一个任务后, 才会创建出线程进行处理
ThreadPoolExecutor 提供了 2 个方法, 可以立即初始化出核心线程:
- prestartCoreThread() 立即创建出一个核心线程
- prestartAllCoreThreads() 所有的核心线程都会被创建
5.6 线程池的关闭
官方提供了 2 个关闭线程池的方法
- shutdown(): 不会立即终止线程池, 而是要等所有任务缓存队列中的任务都执行完后才终止, 但再也不会接受新的任务
- shutdownNow(): 立即终止线程池, 并尝试打断正在执行的任务, 并且清空任务缓存队列, 返回尚未执行的任务
6 线程池的代码实现
6.1 提交任务 execute()
execute 在 ThreadPoolExecutor 类中提交任务的入口 (submit 同样最终还是会调用到 execute 方法)。
在 execute 中主要
- 对任务做非空判断
- 根据各种判断, 做出对任务对应的处理, 如创建核心线程, 添加到阻塞队列等
public void execute(Runnable command) {
// 任务不能为空
if (command == null)
throw new NullPointerException();
// ctl 是一个 AtomicInteger, 初始值为 -536870912, 二进制为 11100000 00000000 00000000 00000000
// ctl 一个 32 位, 一个字段存储了 2 种信息, 前面 3 位表示当前线程池的状态, 后面 29 位表示当前线程池中的线程个数
int c = ctl.get();
// workerCountOf 用入参的值 & 上 536870911, 二进制 00011111 11111111 11111111 11111111
// & 操作, 2 位都是 1, 结果才是 1, 其他的为 0
// ctl 后面 29 位存的是当前的线程个数, 所以 workerCountOf 方法得到的是当前线程的个数
// 判断线程池当前线程数是否小于核心线程数
if (workerCountOf(c) < corePoolSize) {
// 线程数小于核心线程数, 创建核心线程, 处理任务
// addWorker 主要是用户创建线程处理指定的任务, 参数二则决定了创建的线程是核心线程还是非核心线程
if (addWorker(command, true))
return;
// 获取最新的 control
c = ctl.get();
}
// isRunning 判断当前的线程池是否为运行中, 内部的逻辑就是判断 入参是否小于状态值 SHUTDOWN
// 线程池中的所有状态中只有 Running 的最高位是 1, 也就是负数, 所以无论多少个线程数, 正在运行中的状态一定都是负数, 必定小于其他的状态
// 线程池处于运行中, 同时任务添加到阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
// 再做一次检查
int recheck = ctl.get();
// 线程池从运行中变为非运行中了, 同时删除任务成功
if (!isRunning(recheck) && remove(command))
// 调用拒绝策略处理任务
reject(command);
// 线程池线程个数为 0
else if (workerCountOf(recheck) == 0)
// 通过上面的 workQueue.offer 明确队列中已经添加了一个任务, 但是这时候线程数为 0, 手动添加一个
// 为了保证线程池有一个线程来执行任务
addWorker(null, false);
// 如果执行到这里, 有两种情况:
// 1. 线程池已经不是 RUNNING 状态；
// 2. 线程池是 RUNNING 状态, 但 workerCount >= corePoolSize 并且 workQueue 已满
// 尝试添加一个非核心线程
} else if (!addWorker(command, false))
// 添加失败, 调用拒绝策略处理任务
reject(command);
}
}
在执行 execute() 方法时如果状态一直是 RUNNING 时, 执行过程如下:
- 如果 workerCount < corePoolSize, 则创建并启动一个线程来执行新提交的任务
- 如果 workerCount >= corePoolSize, 且线程池内的阻塞队列未满, 则将任务添加到该阻塞队列中
- 如果 workerCount >= corePoolSize && workerCount < maximumPoolSize, 且线程池内的阻塞队列已满, 则创建并启动一个线程来执行新提交的任务
- 如果 workerCount >= maximumPoolSize, 并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常
6.2 添加工作者线程 addWorker()
private boolean addWorker(Runnable firstTask, boolean core) {
// Java 标签
retry:
for (;;) {
int c = ctl.get();
// 获取当前的线程池的状态
// runStateOf 的实现逻辑, 使用入参 & 上 11100000 00000000 00000000 00000000 (& 2 位都是 1 才是 1)
int rs = runStateOf(c);
// 条件 1 : 线程池为非运行状态
// 条件 2 : 线程池为关闭状态, 并且添加的任务为 null, 同时队列为不为空
// 条件 1 满足同时条件 2 不满足的情况下, 直接返回 false
// 整理后为, 线程池的状态为 stop/tidying/terminated 并且添加的任务不为空同时队列为空, 直接返回 false
if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
return false;
for (;;) {
// 获取当前的线程个数
int wc = workerCountOf(c);
// CAPACITY = 536870911, 二进制表示 00011111 11111111 11111111 11111111, 能达到的最大线程数
// 当前的线程数大于等于最大线程数
// 或者需要创建核心线程的话, 当前的线程数大于等于核心线程数
// 或者需要创建非核心线程数的话, 当前的线程数大于等于最大线程数
// 上面 3 种情况, 满足一种就直接返回 fasle
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 通过 CAS 给 ctl 加 1
if (compareAndIncrementWorkerCount(c))
// CAS 加 1 成功, 调节池 retry 指定的循环, 执行下面的操作
break retry;
c = ctl.get();
// CAS 给 ctl 加 1 失败, 重新获取新的状态
// 状态值改变了, 重新回到 retry 标签, 开始新的循环
if (runStateOf(c) != rs)
continue retry;
}
// 线程池数量 + 1 成功后, 开始工作线程的创建
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 创建一个工作者线程, Worker 对任务和工作线程等进行了包装, 后面分析
w = new Worker(firstTask);
// 获取里面的线程
final Thread t = w.thread;
if (t != null) {
// 获取线程池维护的 ReentrantLock
final ReentrantLock mainLock = this.mainLock;
// 上锁
mainLock.lock();
try {
// 获取当前的线程池状态
int rs = runStateOf(ctl.get());
// 当前的线程池状态为运行中 或者 (线程池状态为 shutdown 并且 任务为 null)
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
// 线程 t 已经启动了
if (t.isAlive())
throw new IllegalThreadStateException();
// 将当前的工作线程添加到 ThreadPoolExecutor 的 HashSet 中
workers.add(w);
// 获取当前的工作线程数
int s = workers.size();
// 确保 ThreadPoolExecutor 的出现过最大线程池数的 largestPoolSize 为最大值
if (s > largestPoolSize)
largestPoolSize = s;
// 添加成功
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 工作线程添加成功
if (workerAdded) {
// 启动工作线程
t.start();
workerStarted = true;
}
}
} finally {
// 工作线程启动失败
if (!workerStarted)
// 1. 尝试从 ThreadPoolExecutor 的 HashSet 移除这个线程
// 2. 通过 CAS 给 ctl - 1
// 3. 尝试关闭线程池
addWorkerFailed(w);
}
}
}
上面的逻辑基本就是添加工作线程的逻辑。而工作线程如何执行任务的逻辑在 Worker 这个内部类。
6.3 工作线程 Worker 定义
Worker 是 ThreadPoolExecutor 的内部类, 继承了 AbstractQueuedSynchronizer (AQS) 和 实现了 Runnable 2 个接口。
AbstractQueuedSynchronizer: 抽象类, 提供了一套简单的锁分配机制。
简单的来说 AQS 用 volatile 修饰共享变量 state, 线程通过 CAS 去改变状态符, 成功则获取锁成功, 失败则进入等待队列, 等待被唤醒。
这里不直接使用 ReentrantLock, 是因为 Worker 的获取锁是要不可重入的, 所以自身实现 AbstractQueuedSynchronizer, 自定义获取锁不可重入。
这里的 Worker 可以看做是一个包装类, 包装了任务执行者的 Thread 和执行的任务 Runnable。
一般情况下, 他都是以任务的角色存在, 当调用了 Worker 内 Thread 的 start 方法, 将会转变为任务执行者。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
// 设置 AbstractQueuedSynchronizer 的中的状态为 -1
// -1 为初始值, 0 为 unlock 状态, 1 为 lock 状态
setState(-1);
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
// worker 执行的逻辑
runWorker(this);
}
}
6.4 工作线程 Woker 执行任务 runWorker()
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
public void run() {
// worker 启动时执行的逻辑
runWorker(this);
}
final void runWorker(Worker w) {
// 执行的逻辑, 可以把上面的 Worker 当然任务, 不包含线程的功能
// 获取当前的线程
Thread wt = Thread.currentThread();
// 获取入参 worker 的任务
Runnable task = w.firstTask;
// 将入参的任务置为 null
w.firstTask = null;
// 将 AQS 的 state 设置为无锁状态, 即设置为 0
w.unlock();
// 是否因为异常退出循环
boolean completedAbruptly = true;
try {
// 死循环 入参的 work 的任务不为空或者从其他 worker 获取任务不为空
while (task != null || (task = getTask()) != null) {
// 给入参的 worker 上锁, 设置 state 为 1
w.lock();
// interrupt 设置线程的中断标识为 true
// interrupted() 将当前线程的中断标识设置为 false, 返回值为上次的中断标识
// isInterrupted() 判断线程的中断标识是什么, true 为需要中断
// 下面的逻辑整理后如下:
// 1. 当前的线程池状态为 STOP/TIDYING/TERMINATED, 当前线程的中断标识为 false
// 2. 当前的线程池状态为 STOP/TIDYING/TERMINATED, 当前线程的中断标识为 true, 将当前线程的中断标识先设置为 false
// 作用, 保证线程池状态为 STOP/TIDYING/TERMINATED 下, 当前线程的中断标识为 true
if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
// 尝试中断当前线程 (设置中断标识为 true)
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 {
// 任务置为 null
task = null;
// 设置入参的 Worker 完成的任务数 + 1
w.completedTasks++;
// 解锁
w.unlock();
}
}
// 当前是 Worker 线程正常的结束, 不是因为异常结束
completedAbruptly = false;
} finally {
// 线程退出逻辑处理
processWorkerExit(w, completedAbruptly);
}
}
}
3.4.4.1 工作线程 runWorker() 子流程获取任务 getTask()
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
// 获取任务
private Runnable getTask() {
// 上一次获取任务是否超时
// 第一次默认为 false, 经过下面的流程, 获取任务为 null, 会变成 true, 然后回到循环的开头
boolean timedOut = false;
for (;;) {
int c = ctl.get();
// 当前线程池的状态
int rs = runStateOf(c);
// 线程池进入关闭了
// 1. 线程池状态为 SHUTDOWN/STOP/TERMINATED 状态
// 2. 线程池状态为 STOP/TERMINATED 状态或 workQueue 为空
// 上面 2 个条件都为 true, 返回 null, 让执行线程结束
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// 通过 CAS 给 ctrl - 1
decrementWorkerCount();
// 返回 null
return null;
}
// 当前线程池的线程个数
int wc = workerCountOf(c);
// 线程获取任务是否可以超时 核心线程配置了可以超时消耗 或者 当前的线程数大于核心线程数
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// getTask 方法返回 null 会导致当前线程消耗
// 所以在线程池个数大于 1 或者队列为空的情况下, 可以尝试判断当前线程是否需要销毁
// 1. 当前的线程大于最大线程数
// 2. 线程获取任务可以超时同时上一次获取任务已经超时了
// 2 个条件满足一个, 这个线程就可以销毁了
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
// 通过 CAS 给 ctl 减 1
if (compareAndDecrementWorkerCount(c))
// 返回 null 结束
return null;
// 通过 CAS 给 ctl 减 1 失败, 重试
continue;
}
try {
// 获取任务可以超时的话, 用带时间的阻塞方法从队列中获取任务
Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
// 有结果, 返回结果
if (r != null)
return r;
// 设置这次获取任务超时了
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
}
3.4.4.3 工作线程 runWorker() 子流程共享线程的退出 processWorkerExit()
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 参数二 completedAbruptly: 是否有线程中断导致的退出, 正常的退出为 false
if (completedAbruptly)
// 通过 CAS 给 ctl 减 1
decrementWorkerCount();
// 获取 ThreadPoolExecutor 的可重入锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 已经完成的任务数追加 w 已经完成的任务数
completedTaskCount += w.completedTasks;
// workers 集合异常这个线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 尝试终止
tryTerminate();
int c = ctl.get();
// 当前线程池的状态为 RUNNING 或 SHUTDOWN
if (runStateLessThan(c, STOP)) {
// 当前线程是正常退出的
if (!completedAbruptly) {
// 允许存活的线程数 允许核心线程过期的话, 等于 0 否则等于核心线程数
int min = allowCoreThreadTimeOut ? 0, : corePoolSize;
// 允许的最小线程数为 0 同时有阻塞队列有任务
if (min == 0 && !workQueue.isEmpty())
// 最小的线程数变为 1
min = 1;
// 当前的线程数大于等于最小的线程数, 直接结束
if (workerCountOf(c) >= min)
return;
}
// 当前线程是因异常需要结束 或 当前的线程数小于需要的最小线程数
// 创建一个任务为空的非核心线程
addWorker(null, false);
}
}
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// 当前的线程池为运行状态
// 当前的线程池的状态为 TIDYING 或 TERMINATED
// 当前线程池状态为 SHUTDOWN 同时阻塞队列不为空
// 三个满足其中一个直接结束
if (isRunning(c)
|| runStateAtLeast(c, TIDYING)
|| (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()))
return;
// 线程池线程个数不等于 0
if (workerCountOf(c) != 0) {
// ONLY_ONE 为 true
// 从线程集合 HashSet<Worker> workers 中获取一个进行中断
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 通过 CAS 将 ctl 设置为 1073741824, 二进制 01000000 00000000 00000000 00000000, 也就是把线程数设置为 0, 状态修改为 tidying
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 空方法, 子类进行重新
terminated();
} finally {
// 通过 CAS 将 ctl 设置为 1610612736, 二进制 01100000 00000000 00000000 00000000, 状态修改为 terminated
ctl.set(ctlOf(TERMINATED, 0));
// 唤醒所有阻塞在 ThreadPoolExecutor 的可重入锁 ReentrantLock mainLock 的线程
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
}
}
}
6.5 线程池的关闭 shutdown()
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 通过 SecurityManager 设置当前线程池中的线程都是不可访问的
checkShutdownAccess();
// 设置线程池的状态为 shutdown
advanceRunState(SHUTDOWN);
// 将线程池中的所有线程打上中断标识, 即调用线程的 interrupt 方法
interruptIdleWorkers();
// 空方法, 回调
onShutdown();
} finally {
mainLock.unlock();
}
// 再次调用 tryTerminate 进行线程池的状态改变和线程的关闭
tryTerminate();
}
6.5 线程池的关闭 shutdownNow()
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 通过 SecurityManager 设置当前线程池中的线程都是不可访问的
checkShutdownAccess();
// 设置线程池的状态为 stop
advanceRunState(STOP);
// 将线程池中的所有线程打上中断标识, 即调用线程的 interrupt 方法
interruptWorkers();
// 清空当前队列中的任务
tasks = drainQueue();
} finally {
mainLock.unlock();
}
// 再次调用 tryTerminate 进行线程池的状态改变和线程的关闭
tryTerminate();
// 返回所有未执行的任务
return tasks;
}
6.6 定时线程的实现
从 ScheduledThreadPoolExecutor 实现 ThreadPoolExecutor 可以看出, 其本身的实现基本都是差不多的,
不同的是普通的线程池使用的任务的类型为 Runnable 或者 Callable, 而 ScheduledThreadPoolExecutor 调用到 ScheduledExecutorService 的 4 个定时方法, 内部的任务类型实际为 ScheduledFutureTask,
对里面的 run 方法进行了重写, 执行完成, 设置下次执行时间等操作, 重新放入自定义的优先级阻塞队列等, 具体的实现可以看一下源码, 有着 ThreadPoolExecutor 的基础, 很容易看懂的 (下面只截了一部分)。
private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
/**
* 重复任务的周期 (以纳秒为单位)
* 正值表示固定频率执行,
* 负值表示固定延迟执行,
* 0 表示不是重复任务,
*/
private final long period;
public void run() {
// 判断 period 是否等于 0, false 表示当前任务就是一个普通的任务
boolean periodic = isPeriodic();
// 当前线程池是否能运行这个任务
if (!canRunInCurrentRunState(periodic))
cancel(false);
} if (!periodic)
// 普通任务, 直接执行
ScheduledFutureTask.super.run();
// 周期任务, 执行完成后同时重置状态
else if (ScheduledFutureTask.super.runAndReset()) {
// 操作成功了
// 设置下次执行时间
setNextRunTime();
// 重新投递这个任务
reExecutePeriodic(outerTask);
}
}
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
if (canRunInCurrentRunState(true)) {
// 添加到队列中
super.getQueue().add(task);
if (!canRunInCurrentRunState(true) && remove(task))
task.cancel(false);
else
// 当前线程数小于核心线程, 创建一个核心线程
// 当前的线程数等于 0, 创建一个非核心线程
// 总之就是确保有线程执行任务
ensurePrestart();
}
}
}
7 如何合理配置线程池参数
要想合理的配置线程池, 就必须首先分析任务特性, 可以从以下几个角度来进行分析
- 任务的性质: CPU 密集型任务, IO 密集型任务和混合型任务
- 任务的优先级: 高, 中和低
- 任务的执行时间: 长, 中和短
- 任务的依赖性: 是否依赖其他系统资源, 如数据库连接
任务性质不同的任务可以用不同规模的线程池分开处理。
CPU 密集型任务配置尽可能少的线程数量, 如配置 Ncpu+1 个线程的线程池。 IO 密集型任务则由于需要等待 IO 操作, 线程并不是一直在执行任务, 则配置尽可能多的线程, 如 2 x Ncpu。
混合型的任务, 如果可以拆分, 则将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务, 只要这两个任务执行的时间相差不是太大, 那么分解后执行的吞吐率要高于串行执行的吞吐率, 如果这两个任务执行时间相差太大, 则没必要进行分解。
可以通过 Runtime.getRuntime().availableProcessors 方法获得当前设备的 CPU 个数。
优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理。它可以让优先级高的任务先得到执行, 需要注意的是如果一直有优先级高的任务提交到队列里, 那么优先级低的任务可能永远不能执行
执行时间不同的任务可以交给不同规模的线程池来处理, 或者也可以使用优先级队列, 让执行时间短的任务先执行。
依赖数据库连接池的任务, 因为线程提交 SQL 后需要等待数据库返回结果, 如果等待的时间越长 CPU 空闲时间就越长, 那么线程数应该设置越大, 这样才能更好的利用 CPU
IO 密集型 = Ncpu * 2
IO 密集型 = Ncpu / (1 - 阻塞系数), 阻塞系数 = 阻塞时间 /(阻塞时间 + 计算时间)
8 参考
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：chenni525@qq.com进行投诉反馈，一经查实，立即删除！
- Python教程
- 深入理解 MySQL 中的 HAVING 关键字和聚合函数
- Qt之QChar编码（1）
- MyBatis入门基础篇
- 用Python脚本实现FFmpeg批量转换
- Git推送本地文件到仓库
- 软件工程：黑盒测试等价分类法相关知识和多实例分析
- Redis：原理速成+项目实战——Redis实战7（优惠券秒杀+细节解决超卖、一人一单问题）
- 多维时序 | MATLAB实现RIME-LSSVM【23年新算法】基于霜冰优化算法(RIME)优化最小二乘向量机(LSSVM)多变量时间序列预测
- 算法通关村番外篇-LeetCode编程从0到1系列五
- 在分类任务中准确率（accuracy）、精确率（precision）、召回率（recall）和 F1 分数是常用的性能指标，如何在python中使用呢？
- Hikvision综合安防管理平台files；.css接口存在任意文件读取漏洞 附POC软件
- A Tour of Go部分练习
- 算法训练营第五十五天|392.判断子序列 115.不同的子序列
- 专业146总分400+重庆大学841信号与系统考研经验重大电子信息与通信工程