文章目录
- 1 问题背景
- 2 前言
- 3 线程池是什么
- 4 线程池核心设计与实现
- 4.1 总体设计
- 4.2 运行机制
- 4.3 生命周期管理
- 4.4 任务执行机制
- 4.4.1 任务调度
- 4.4.2 任务缓冲
- 4.4.3 任务申请
- 4.4.4 任务拒绝
- 4.5 Worker线程管理
- 4.5.1 Worker线程
1 问题背景
前面总结了工作中的解决方案——电商项目之百万级别的临时订单数据补偿解决方案。其中应用到了线程池,今天简单了解一下线程池的基础知识。
参考自:Java线程池实现原理及其在美团业务中的实践
2 前言
- 本篇博客旨在对线程池的各个参数有一个基本的全局了解,基本不涉及案例demo代码。
- 不必背诵这些知识点,有不懂之处需要上下文反复阅读,反复思考。
- 本文描述线程池是JDK1.8中提供的
ThreadPoolExecutor
类。- 本文基本都是从Java线程池实现原理及其在美团业务中的实践搬来的,有需要了解更详细的信息可以直接看它。
3 线程池是什么
线程池是池化思想的一种体现。
线程过多的缺点:
- 创建、销毁线程的开销
- 调度线程的开销
线程池维护多个线程,它解决了
- 避免创建、销毁线程的开销
- 避免线程数量膨胀导致过分的调度问题
总结:线程池就是解决了线程过多的缺点。可以理解成线程池是一个管理线程的管理者。
线程池的优点:
- 提高资源利用率。重复利用已创建的线程。
- 提高响应速度。任务到达时,无需等待线程创建即可立刻执行任务。
- 提高线程的可管理性。线程是稀缺资源,无限制创建线程会使得线程不合理分布导致降低系统的稳定性。使用线程池进行统一分配、调优和监控。
- 提供更多更强大的功能。比如延时、定时线程池,允许任务延时执行或定时执行。
4 线程池核心设计与实现
4.1 总体设计
Java中线程池核心实现类是
ThreadPoolExecutor
。基于JDK1.8的源码,ThreadPoolExecutor
的继承关系如下:
Executor
:该接口的思想是将任务提交和任务执行解耦。用户只需提供Runnable对象(任务的运行逻辑)给Executor,由Executor完成线程的调配和任务的执行。ExecutorService
:增加了一些能力:(1)补充了为一个或一批异步任务生成Future的方法;(2)提供了管控线程池的方法。比如停止线程池的运行。AbstractExecutorService
:将执行任务的流程串联起来。ThreadPoolExecutor
:实现最复杂的运行部分。一方面维护自身的生命周期;另一方面同时管理线程和任务,使两者良好结合从而执行并行任务。
4.2 运行机制
ThreadPoolExecutor
如何同时维护线程和执行任务的呢?其运行机制如下所示:
线程池内部使用了生产者消费者模型,将任务和线程解耦。
线程池的运行主要分成两部分:
- 任务管理: 充当生产者的角色。当任务提交后,线程池会判断该任务后续的流转:(1)直接申请线程执行任务;(2)缓冲到阻塞队列中等待执行;(3)拒绝该任务。
- 线程管理: 充当消费者的角色。线程被统一维护在
图中的线程池
内,根据任务请求进行线程分配,线程执行完任务后会继续获取新的任务去执行,最终当线程获取不到任务的时候,线程就会被回收。
接下来,会按照以下三个部分去详细讲解线程池运行机制:
- 线程池如何维护自身状态。
- 线程池如何管理任务。
- 线程池如何管理线程。
4.3 生命周期管理
线程池运行的状态是由其内部维护的。它使用一个变量维护两个值:运行状态(runState) 和线程数量(workerCount)。源码如下:
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
ctl
这个变量包含两部分的信息:高3位保存runState
,低29位保存workerCount
。两部分信息之间互不干扰。
用一个变量存储两个值,可以避免在做相关决策时出现不一致的情况。不必为了维护两者的一致而占用锁资源。
线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数,如下所示:
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
线程 池的五种状态如下:
运行状态 | 状态描述 |
---|---|
RUNNING | 能接收新提交的任务,也能处理阻塞队列中的任务 |
SHUTDOWN | 关闭状态,不再接收新提交的任务,但仍会继续处理阻塞队列中的任务 |
STOP | 不接收新任务,也不处理阻塞队列中的任务,中断正在处理任务的线程 |
TYDYING | 所有的任务已终止,workerCount(有效线程)为0 |
TERMINATED | 在terminated() 方法执行完后进入该状态 |
线程池生命周期的状态转换如下:
4.4 任务执行机制
4.4.1 任务调度
任务调度是线程池的主要入口,当用户提交了一个任务,接下来这个任务将如何执行都是由这个阶段决定的。
所有任务调度都是由
execute()
完成的,其执行过程如下:
- 检测线程池的状态,如果不是RUNNING,则直接拒绝。线程池要保证在RUNNING的状态下执行任务。
- 如果
workerCount < corePoolSize
,则创建并启动一个线程来执行新提交的任务。 - 如果
workerCount >= corePoolSize
,并且阻塞队列未满,则将任务添加到该阻塞队列中。 - 如果
workerCount >= corePoolSize && workerCount < maxmimumPoolSize
,且线程池内的阻塞队列已满,则创建并启动一个线程来执行新提交的任务。 - 如果
workerCount >= maxmimumPoolSize
,并且阻塞队列已满,则根据拒绝策略来处理任务。默认的方式是直接抛异常。
任务调度流程图如下:
4.4.2 任务缓冲
线程池中以生产者消费者模式,通过一个阻塞队列实现,达到将任务与线程解耦的目的。阻塞队列缓存任务,工作线程从阻塞队列中获取任务。
阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是:在队列为空时,获取元素的线程会等待队列变为非空。
下图中展示了线程1往阻塞队列中添加元素,而线程2从阻塞队列中移除元素:
使用不同的队列可以实现不一样的任务存取策略。在这里,我们可以再介绍下阻塞队列的成员:
名称 | 描述 |
---|---|
ArrayBlockingQueue | 一个用数组实现的有界队列,此队列按照先进先出(FIFO)原则对元素进行排序。支持公平锁和非公平锁。 |
LinkedBlockingQueue | 一个用链表结构组成的有界队列,此队列按照先进先出(FIFO)原则对元素进行排序。队列的默认长度为Integer.MAX_VALUE ,所以默认创建的队列有容量危险。 |
PriorityBlockingQueue | 一个支持线程优先级的无界队列,默认自然序进行排序。也可以实现compareTo() 来指定元素排序规则。不能保证同优先级元素的顺序。 |
DelayQueue | 一个实现PriorityBlockingQueue实现延时获取的队列。在创建元素时,可以指定多久才能从队列中获取当前元素。只有延时期满后才能从队列中获取元素。 |
SynchronousQueue | 一个不存储元素的阻塞队列,每一个put操作必须等待take操作,否则不能添加元素。支持公平锁和非公平锁。Executors.newCachedThreadPool() 就使用了SynchronousQueue,这个线程池根据需要(新任务到来时)创建新的线程,如果有空闲线程则会重复使用,线程空闲了60s后会被回收。 |
LinkedTransferQueue | 一个有链表结构组成的无界阻塞队列,相当于其他队列。LinkedTransferQueue多了transfer和tryTransfer方法。 |
LinkedBlockingDeque | 一个由链表结构组成的双向阻塞队列。队列头部和尾部都可以添加移除元素。多线程并发时,可以将锁的竞争最多降到一半。 |
4.4.3 任务申请
由4.4.1任务调度小节中得知,任务的执行有2种可能:
- 任务直接由新创建的线程执行
- 线程从任务队列获取任务然后执行,执行完任务的空闲线程再次从队列中申请任务再去执行。
第一种情况 仅出现在线程初始创建的时候,第二种则是线程获取任务绝大多数的情况。
线程需要从任务缓存模块中不断获取任务执行,帮助线程从阻塞队列中获取任务,实现任务管理模块与线程管理模块之间的通信。这部分策略由
getTask()
实现,其执行流程如下:
getTask()
进行了多次判断,目的是为了控制线程的数量,使其符合线程池的状态。如果线程池现在不应该持有那么多的线程,则会返回null值。工作线程worker会不断接收新任务执行,当工作线程worker接收不到任务时,就会开始被回收。
4.4.4 任务拒绝
任务拒绝模块是线程池的保护部分,线程池有一个最大的容量,当线程池的任务缓存队列已满,并且线程池中的线程数目达到maximumPoolSize时,就需要拒绝掉该任务,采取任务拒绝策略,保护线程池。
拒绝策略是一个接口,其设计如下:
public interface RejectedExecutionHandler {void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
用户可以通过实现这个接口去定制拒绝策略,也可以选择JDK提供的四种已有拒绝策略,其特点如下:
序号 | 名称 | 描述 |
---|---|---|
1 | ThreadPoolExecutor.AbortPolicy | 丢弃任务并抛出RejectedExecutionException 异常。这是线程池默认的拒绝策略。在任务不能再提交时,抛出异常,及时反馈 程序运行状态。如果是比较关键的业务,推荐使用该策略,这样子在系统不能承受更大的并发量的时候,能够及时通过异常发现。 |
2 | ThreadPoolExecutor.DiscardPolicy | 丢弃任务,但不抛异常。使用该策略会使得我们无法及时发现系统的异常状态,建议是一些无关紧要的业务采用此策略。 |
3 | ThreadPoolExecutor.DiscardOldestPolicy | 丢弃队列最前面的任务,然后重新提交被拒绝的任务。是否采用此种策略,还得根据实际业务是否允许丢弃老任务来认真衡量。 |
4 | ThreadPoolExecutor.CallerRunsPolicy | 由调用线程(提交任务的线程)处理该任务。这种情况是需要所有任务都执行完毕,那么就适合大量计算的任务类型去执行。多线程仅仅是增大吞吐量的手段,最终必须要让每个任务都执行完毕。 |
4.5 Worker线程管理
4.5.1 Worker线程
线程池为了掌握线程的状态并维护线程的生命周期,设计了线程池内的工作线程Worker。部分代码如下:
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{final Thread thread;//Worker持有的线程Runnable firstTask;//初始化的任务,可以为null
}
记录:后续的知识都是围绕线程管理如何运行的,笔者都不太懂,打算不继续写下去了,需要了解详情可以看美团的文章。基本是说申请添加线程,回收线程等,都会有涉及锁的操作。所以创建线程、回收线程会消耗一定的系统资源。文章末尾还给出线程池的一整套解决方案,动态参数化设置线程池的核心参数、监控、告警、权限修改等。