【ThreadPoolExecutor】自定义线程池详解（一篇透彻）_资讯

【ThreadPoolExecutor】自定义线程池详解（一篇透彻）

创始人

2025-05-30 23:49:35

0次

这世界上没有优秀的理念，只有脚踏实地的结果。

上一篇写了Executors创建线程池的四种方法【查看】

接下来主要对ThreadPoolExecutor类的核心参数及工作原理进行说明。

一、简介

ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务。

二、依赖关系梳理

1、Executor线程池相关顶级接口，它将任务的提交与任务的执行分离开来

2、ExecutorService继承并扩展了Executor接口，提供了Runnable、FutureTask等主要线程实现接口扩展

3、AbstractExecutorService提供 ExecutorService 执行方法的默认实现。此类使用 newTaskFor 返回的 RunnableFuture 来实现 commit，invokeAny 和 invokeAll 方法，该方法默认为此包中提供的 FutureTask 类。

4、ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务

三、线程池的生命周期

线程有线程的生命周期，线程池也有线程池的生命周期，其主要为以下几个状态，线程池的运行状态，并不是用户显示设置的，而是伴随着线程池的运行，由内部来维护。

生命周期状态：

RUNNING：运行状态能够接受新提交的任务，并且能够处理阻塞队列中的任务
SHUTDOWN：关闭状态，不再接受新提交的任务，能继续处理阻塞队列中的任务
STOP：停止状态，不接受新的任务，也不处理阻塞队列中的任务，中断正在处理的任务线程
TIDYING：整理状态所有的任务都已经完成终止，workCount（有效线程数为0）
TERMINATED：终止状态在terminated方法执行之后进入该状态

生命周期图形化：

四、参数介绍

1.1 构造参数

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue 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;}

1.2 ThreadPoolExecutor包含了7个核心参数：

参数	名称
corePoolSize	核心线程数
maximumPoolSize	最大线程数
keepAliveTime	空闲线程存活时间
unit	keepAliveTime时间单位
workQueue	保存任务的工作队列
threadFactory	线程工厂，提供线程的创建方式
handler	拒绝策略

详细介绍一下：

corePoolSize：核心线程数

线程池会维护一个最小的线程数量，即使这些线程没有任务需要执行处于空闲状态，他们也不会被销毁。
当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理，总之一定会保持核心数量的线程。
设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭。

maximumPoolSize：最大线程数

线程池最大创建线程数量，线程池创建超过最大线程数就不会再创建线程。
当线程数>=corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务。
当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常。

keepAliveTime：空闲线程存活时间

一个线程如果处于空闲状态，并且当前的线程数量大于corePoolSize，那么在指定时间后，这个空闲线程会被销毁，这里的指定时间由keepAliveTime来设定。

unit：keepAliveTime时间单位

TimeUnit.DAYS 天
TimeUnit.HOURS 小时
TimeUnit.MINUTES 分钟
TimeUnit.SECONDS 秒
TimeUnit.MILLISECONDS 毫秒
TimeUnit.MICROSECONDS 微秒
TimeUnit.NANOSECONDS 纳秒

workQueue：工作队列

队列线程数关系：

线程数少于corePoolSize，对于提交的新任务会创建一个新的线程处理，并不会把任务放入队列
如果线程数介于corePoolSize和maximumPoolSize之间，新提交的任务会被放入阻塞队列中
如果线程池处于饱和状态，即无法创建线程也无法存放在阻塞队列，那么新任务将交由拒绝策略来处理

阻塞队列：（此处不做详细介绍）

SynchronousQueue：是BlockingQueue的一种，是线程安全的。但是它和其BlockingQueue不同的是它的capacity是0。即SynchronousQueue不存储任何元素。只是维护一组线程，在等待着把元素加入或移出队列，相当于直接交接任务给具体执行的线程。
ArrayBlockingQueue：基于数组的有界阻塞队列，按FIFO排序。新任务进来后，会放到该队列的队尾，有界的数组可以防止资源耗尽问题。
LinkedBlockingQueue：采用链表实现的无界队列，按FIFO排序。如果使用时没有预定义容量，当所有corePoolSize线程都在处理任务时，将导致新任务都会在队列中等待，不会创建超过corePoolSize个线程。这种场景下maximumPoolSize的值对于线程数量没有任何影响。

threadFactory：线程工厂

提供了线程池中线程的创建方式，这里使用了工厂模式ThreadFactory创建新线程。

handler（RejectedExecutionHandler）：拒绝策略

如果线程池处于饱和状态，没有足够的线程数或者队列空间来处理提交的任务，或者是线程池已经处于关闭状态但还在处理进行中的任务，那么继续提交的任务就会根据线程池的拒绝策略处理。
两种情况会拒绝处理任务：
- 当线程数已经达到maxPoolSize并且队列已满，会拒绝新任务。
- 当线程池被调用shutdown()后，会等待线程池里的任务执行完毕，再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝。
ThreadPoolExecutor类提供了四种拒绝策略：
- AbortPolicy (默认) 丢弃任务并抛出运行时异常。
- CallerRunsPolicy 调用者执行该任务。
- DiscardPolicy 直接丢弃且不抛出异常，什么都不会发生。
- DiscardOldestPolicy 如果线程关闭，则抛弃任务，否则从队列中踢出最先进入队列的任务，也就是存活时间最久的任务（弃老策略）。

五、线程池的执行顺序

接下来我们用一张图片来描述一下线程池的执行顺序

六、线程池参数设置

线程池的参数设置一直是一个很头疼的问题。这块我没有给出具体答案，因为每个线程池的配置不是一成不变的，每个线程池都要根据实际的业务需求去修改参数配置。当然网上也会有很多教程去教怎么配置，但是这只是千篇一律的，还是建议大家根据实际情况去配置相关参数。

6.1 疑问❓

我们怎么去合理设置线程池的参数？又怎么高效快速无感知的设置线程池参数？

那么有没有方法去动态调整线程池。。。。。答案是：当然有。

七、线程池动态调整

ThreadPoolExecutor提供的方法实现动态调整，查看ThreadPoolExecutor源码：

//设置核心线程数
public void setCorePoolSize(int corePoolSize);
//设置最大线程数
public void setMaximumPoolSize(int maximumPoolSize);
//设置空闲线程存活时间
public void setKeepAliveTime(long time, TimeUnit unit);
//设置线程工厂
public void setThreadFactory(ThreadFactory threadFactory);
//设置拒绝策略
public void setRejectedExecutionHandler(RejectedExecutionHandler handler);
//设置是否允许核心线程超时关闭
public void allowCoreThreadTimeOut(boolean value);//获取核心线程数
public int getCorePoolSize();
//获取线程最大数
public int getMaximumPoolSize();
//获取空闲线程存活时间
public long getKeepAliveTime(TimeUnit unit);
//获取队列信息
public BlockingQueue getQueue();
//获取拒绝策略
public RejectedExecutionHandler getRejectedExecutionHandler();
//获取是否允许核心线程超时关闭
public boolean allowsCoreThreadTimeOut();

既然提供了相应方法那就好办了。比如我们可以通过写一个对外的API接口，通过接口传参来控制线程池的参数设置。

7.1 通过API接口实现

接下来我用代码简单实现一个对线程池的查看和修改参数代码：

@RestController
@RequestMapping(value = "/api/concurrent")
public class ConcurrentApi {@Autowiredprivate ApplicationContext applicationContext;@ApiOperation(value = "统计线程池使用情况", notes = "")@PostMapping(path = "/threadPool/stat")@SuppressWarnings("unchecked")public String statThreadPool(@Valid BaseReq req, @ApiIgnore BindingResult bindingResult) {Map handlerBeanMap = applicationContext.getBeansOfType(ThreadPoolExecutor.class);List> statList = Lists.newArrayList();for (Map.Entry entry : handlerBeanMap.entrySet()) {String k = entry.getKey();ThreadPoolExecutor v = entry.getValue();Map statMap = Maps.newLinkedHashMap();statMap.put("Name", k);statMap.put("PoolSize", v.getPoolSize());statMap.put("CorePoolSize", v.getCorePoolSize());statMap.put("MaximumPoolSize", v.getMaximumPoolSize());statMap.put("LargestPoolSize", v.getLargestPoolSize());statMap.put("ActiveCount", v.getActiveCount());statMap.put("TaskCount", v.getTaskCount());statMap.put("CompletedTaskCount", v.getCompletedTaskCount());statMap.put("QueueSize", v.getQueue().size());statList.add(statMap);}return JSON.toJSONString(statList);}@ApiOperation(value = "修改线程池参数", notes = "")@PostMapping(path = "/threadPool/update")@SuppressWarnings("unchecked")public void updateThreadPool(@Valid ThreadPoolParamReq req, @ApiIgnore BindingResult bindingResult) {Map handlerBeanMap = applicationContext.getBeansOfType(ThreadPoolExecutor.class);ThreadPoolExecutor threadPoolExecutor = handlerBeanMap.get(req.getThreadPoolName());if (req.getCorePoolSize() > 0) {threadPoolExecutor.setCorePoolSize(req.getCorePoolSize());}if (req.getMaximumPoolSize() > 0 && req.getMaximumPoolSize() >= req.getCorePoolSize()) {threadPoolExecutor.setMaximumPoolSize(req.getMaximumPoolSize());}}@Data@ApiModelstatic class ThreadPoolParamReq {@NotBlank@ApiModelProperty(value = "线程池名称", required = true)private String threadPoolName;@ApiModelProperty(value = "corePoolSize")private int corePoolSize;@ApiModelProperty(value = "maximumPoolSize")private int maximumPoolSize;}
}

7.2 通过配置中心（Nacos）实现

这块我就没有写具体代码去实现，但是大概可以说一下实现思路

1）实现原理：

利用Nacos作为配置中心，动态监听Nacos配置变更。

7.3 通过Dinamic-Tp实现

Dinamic-Tp是美团开源的一个动态调整线程池参数的框架。

有以下特性：

代码零侵入
通知告警
运行监控
任务增强
多配置中心支持：Nacos、Apollo、Zookeeper、Consul、Etcd、Polaris
中间件线程池管理
轻量简单：基于 SpringBoot 实现
兼容性：JUC 普通线程池和 Spring 中的 ThreadPoolTaskExecutor 也可以被框架监控

技术架构图

Dynamic-Tp官网：https://dynamictp.cn/

本文参考文章：https://blog.csdn.net/m0_67401270/article/details/126107525

词库加载错误:未能找到文件“E:\highferrum_mysql\Configuration\Dict_Stopwords.txt”。

上一篇：基于梯度的优化问题中不可导操作的处理方法总结

下一篇：Linux -- 磁盘存储管理管理交换分区（虚拟内存）