动态线程池组件 (Dynamic Thread Pool)
组件作用
由于JDK自带的线程池无法满足业务需求，无法动态调整线程池的大小，因此需要自定义线程池组件。 目前使用了dynamic-tp，一个基于配置中心的轻量级动态可监控线程池。
功能特性
- 代码零侵入：所有配置均放在配置中心，服务启动时会从配置中心拉取配置生成线程池对象放到 Spring 容器中，使用时直接从 Spring 容器中获取，对业务代码零侵入
- 通知告警：提供多种通知告警维度（配置变更通知、活性报警、队列容量阈值报警、拒绝触发报警、任务执行或等待超时报警），触发配置阈值实时推送告警信息，已支持企微、钉钉、飞书、邮件、云之家报警，同时提供 SPI 接口可自定义扩展实现
- 运行监控：定时采集线程池指标数据（20 多种指标，包含线程池维度、队列维度、任务维度、tps、tp99等），支持通过 MicroMeter、JsonLog、JMX 三种方式定时获取，也可以通过 SpringBoot Endpoint 端点实时获取最新指标数据，同时提供 SPI 接口可自定义扩展实现
- 任务增强：提供任务包装功能（比 Spring 线程池任务包装更强大），实现 TaskWrapper 接口即可，如 MdcTaskWrapper、TtlTaskWrapper、SwTraceTaskWrapper、OpenTelemetryWrapper，可以支持线程池上下文信息传递
- 中间件线程池管理：集成管理常用第三方组件的线程池，已集成 Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3、Brpc、Tars、SofaRpc、RabbitMq 等组件的线程池管理（调参、监控报警）
- 多模式：提供了增强线程池 DtpExecutor，IO 密集型场景使用的线程池 EagerDtpExecutor，调度线程池 ScheduledDtpExecutor，有序线程池 OrderedDtpExecutor，可以根据业务场景选择合适的线程池
- 兼容性：JUC 普通线程池和 Spring 中的 ThreadPoolTaskExecutor 也可以被框架管理，@Bean 定义时加 @DynamicTp 注解即可
- 可靠性：依靠 Spring 生命周期管理，可以做到优雅关闭线程池，在 Spring 容器关闭前尽可能多的处理队列中的任务
- 高可扩展：框架核心功能都提供 SPI 接口供用户自定义个性化实现（配置中心、配置文件解析、通知告警、监控数据采集、任务包装等等）
组件用法
pom.xml引入依赖
<dependency>
<groupId>cn.com.rydeen.component</groupId>
<artifactId>rydeen-component-threadpool</artifactId>
</dependency>
配置中心配置线程池
dynamictp:
# 是否开启动态线程池，默认true
enabled: true
# 是否开启banner，默认true
enabled-banner: false
# 是否开启监控指标采集，默认false
enabled-collect: true
# 监控数据采集器类型（logging | micrometer | internal_logging），默认micrometer
collector-types: micrometer,logging
# 指标日志存储路径，仅用于“日志记录”类型。
log-path: /app/logs/platform/dtp
# 监控时间间隔（报警判断、指标采集），默认5s
monitor-interval: 5
# 通知平台需要配合 dynamic-tp-spring-boot-starter-extension-xxx 模块使用，
# 参考：https://central.sonatype.com/search?q=dynamic-tp-spring-boot-starter-extension
platforms:
- platform: wechat
platform-id: 1 # 平台id
url-key: 3a700-127-4bd-a798-c53d8b69c # 替换
receivers: "test1,test2" # 接受人企微名称
- platform: ding
# 平台id
platform-id: 2
# 企业钉钉url-key
url-key: f80dad441fcd655438f4a08dcd6a
# 替换，非sign模式可以没有此值
secret: SECb5441fa6f375d5b9d21
# 钉钉账号手机号
receivers: "18888888888"
- platform: lark
platform-id: 3
url-key: 0d944ae7-b24a-40
# 接受人飞书名称/openid
receivers: "test1,test2"
- platform: email
platformId: 4
receivers: "richie696@vip.qq.com"
# okhttp3 线程池配置
okhttp3-tp:
- thread-pool-name: okHttpTp
core-pool-size: 10
maximum-pool-size: 200
keep-alive-time: 60
# tomcat webserver 线程池配置
tomcat-tp:
thread-pool-name: tomcatTp
core-pool-size: 10
maximum-pool-size: 200
keep-alive-time: 60
executors:
# 线程池名称
- thread-pool-name: demoExecutor
# 线程池简化别名，用于发送通知时（当线程池名称较长时，可通过本参数精简名称，并用于发送通知）
thread-pool-alias-name: demoExecutor
# 核心线程池大小（默认：1）
core-pool-size: 1
# 最大线程池大小（默认：处理器可用核心数）
maximum-pool-size: 8
# 线程池空闲线程存活时间（默认：60）
# 注：当线程数大于内核数时，这是多余的空闲线程在终止之前等待新任务的最长时间。
keep-alive-time: 60
# 线程池空闲线程存活时间单位（默认：SECONDS）
unit: SECONDS
# 是否开启报警（默认：true）
notify-enabled: true
# 绑定的报警平台ID（需要配合platforms节点配置）
platform-ids:
- 1
- 4
# 报警项列表
notify-items:
# 是否启用
- enabled: true
platform-ids:
- ??
# 报警项类型（变动通知、活跃性阈值通知、容量阈值通知、队列拒绝通知、任务运行超时警报通知、任务队列等待超时警报通知）
#（参考：org.dromara.dynamictp.common.em.NotifyItemEnum）
type: capacity
# 报警阈值（当type=CAPACITY时，为线程池使用率阈值；当type=RUN_TIMEOUT时，为任务执行超时阈值；当type=QUEUE_TIMEOUT时，为任务在队列等待超时阈值）
threshold: 0
# 报警间隔（单位：秒，默认：120）
interval: 120
# 群集通知限制（默认：1）
cluster-limit: 1
# 线程池类型（参考：org.dromara.dynamictp.core.support.ExecutorType）
executor-type: common
# 线程池队列类型（参考：org.dromara.dynamictp.common.em.QueueTypeEnum，默认：VariableLinkedBlockingQueue）
queue-type: VariableLinkedBlockingQueue
# 线程池队列容量（默认：1024）
queueCapacity: 1024
# 是否使用公平策略（当queue-type=SYNCHRONOUS_QUEUE时，默认：非公平策略）
fair: false
# MemorySafeLBQ 的最大可用内存（单位：MB，默认：16）
max-free-memory: 16
# 当队列满了之后，拒绝策略（参考：org.dromara.dynamictp.common.em.RejectedTypeEnum，默认：AbortPolicy）
rejected-handler-type: AbortPolicy
# 是否允许核心线程池超时（默认：false）
allow-core-thread-time-out: false
# 线程名称前缀（默认：dtp）
thread-name-prefix: dtp
# 是否在服务停止时，等待任务执行完成后再停止服务（默认：false）
wait-for-tasks-to-complete-on-shutdown: false
# 线程池关闭时，等待任务完成的超时时间（单位：秒，默认：0）
await-termination-seconds: 0
# 是否预热所有核心线程（默认：false）
pre-start-all-core-threads: false
# 是否增强拒绝策略（默认：true）
reject-enhanced: true
# 任务执行超时阈值（仅用于统计）（单位：毫秒，默认：0）
run-timeout: 0
# 任务在队列等待超时阈值（仅用于统计）（单位：毫秒，默认：0）
queue-timeout: 0
# 任务包装器名称（默认：空）
task-wrapper-names: ["ttl", "mdc", "OTel", "custom"]
启动类加 @EnableDynamicTp 注解
@Slf4j
@EnableDynamicTp
@SpringBootApplication
public class DemoApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(DemoApplication.class).run(args);
}
}
代码引用
使用 @Resource 或 @Autowired 进行依赖注入，或通过 DtpRegistry.getExecutor("name") 获取
- 依赖注入方式使用，优先推荐依赖注入方式，不能使用依赖注入的场景可以使用方式2
@Resource
private ThreadPoolExecutor dtpExecutor1;
public void exec() {
dtpExecutor1.execute(() -> System.out.println("test"));
}
- 从 DtpRegistry 注册器获取（接口场景可用）
public static void main(String[] args) {
DtpExecutor dtpExecutor = DtpRegistry.getDtpExecutor("dtpExecutor1");
dtpExecutor.execute(() -> System.out.println("test"));
}
任务包装
由于子线程执行的上下文可能不一致，因此，框架提供了一些任务包装器，可以实现特定的功能
- MdcTaskWrapper 支持 MDC 上下文传递，名称：mdc
- TtlTaskWrapper 支持 ThreadLocal 上下文传递，名称：ttl
- SwTraceTaskWrapper 支持 Skywalking TID 传递，名称：swTrace
- OpenTelemetryWrapper，支持 OpenTelemetry 上下文传递，名称：OTel
- NamedRunnable 支持给任务添加名称
- 可以继承 TaskWrapper SPI 接口自定义任务包装器
自定义 TaskWrapper 示例
import lombok.extern.slf4j.Slf4j;
import org.dromara.dynamictp.core.support.task.wrapper.TaskWrapper;
@Slf4j
public class CustomTaskWrapper implements TaskWrapper {
@Override
public String name() {
return "custom";
}
@Override
public Runnable wrap(Runnable runnable) {
return new MyRunnable(runnable);
}
public static class MyRunnable implements Runnable {
private final Runnable runnable;
public MyRunnable(Runnable runnable) {
this.runnable = runnable;
}
@Override
public void run() {
log.info("before run");
runnable.run();
log.info("after run");
}
}
}
resources/META-INF/services 下配置 java spi 实现 文件名：org.dromara.dynamictp.core.support.task.wrapper.TaskWrapper 文件值：org.dromara.dynamictp.example.wrapper.CustomTaskWrapper
线程池配置文件加如下配置项
spring:
dynamic:
tp:
executors: # 动态线程池配置，省略其他项，具体看上述配置文件
- threadPoolName: dtpExecutor1
taskWrapperNames: ["ttl", "mdc", "swTrace", "custom"] # 任务包装器名称，继承 TaskWrapper 接口