动态线程池（DynamicTp）技术文档

概述

动态线程池（Dynamic Thread Pool，简称DynamicTp）是一个基于配置中心的轻量级动态线程池框架，内置监控告警功能，集成常用中间件线程池管理。它解决了传统线程池使用中的三大痛点：参数配置困难、运行时无法动态调整、运行状态无法实时感知。

核心特性

🚀 零代码侵入

所有配置均放在配置中心，服务启动时从配置中心拉取配置生成线程池对象，使用时直接从Spring容器中获取，对业务代码完全无侵入。

📊 实时监控告警

提供20+种监控指标，支持多种告警维度：

• 配置变更通知
• 线程池活性报警
• 队列容量阈值报警
• 拒绝触发报警
• 任务执行/等待超时报警

🔧 多配置中心支持

支持主流配置中心：

• Nacos
• Apollo
• Zookeeper
• Consul
• Etcd
• Polaris
• ServiceComb

🛠️ 中间件集成

已集成管理常用第三方组件的线程池：

• Web服务器：Tomcat、Jetty、Undertow
• RPC框架：Dubbo、gRPC、Motan、Brpc、Tars、SofaRpc
• 消息队列：RocketMQ、RabbitMQ
• 其他：Hystrix、OkHttp3、Liteflow、Thrift

架构设计

graph TB
    subgraph "配置中心"
        Config[Nacos/Apollo/Zookeeper]
    end
    
    subgraph "DynamicTp核心"
        Listener[配置变更监听模块]
        Manager[线程池管理模块]
        Monitor[监控模块]
        Notifier[通知告警模块]
        Adapter[适配器模块]
    end
    
    subgraph "业务应用"
        App[应用程序]
        ThreadPool[线程池实例]
    end
    
    Config --> Listener
    Listener --> Manager
    Manager --> ThreadPool
    ThreadPool --> Monitor
    Monitor --> Notifier
    Notifier --> Adapter
    Adapter --> App
    App --> ThreadPool

核心组件

DtpExecutor

增强型线程池，继承自ThreadPoolExecutor，提供动态参数调整能力：

public class DtpExecutor extends ThreadPoolExecutor {
    // 动态设置核心参数
    public void setCorePoolSize(int corePoolSize);
    public void setMaximumPoolSize(int maximumPoolSize);
    public void setKeepAliveTime(long time, TimeUnit unit);
    
    // 监控相关方法
    public String getThreadPoolName();
    public void setNotifyEnabled(boolean notifyEnabled);
    public void setNotifyItems(List<NotifyItem> notifyItems);
}

线程池类型

类型	类名	适用场景	特点
通用线程池	DtpExecutor	CPU密集型任务	标准ThreadPoolExecutor增强
饥饿线程池	EagerDtpExecutor	IO密集型任务	任务先创建线程，队列为后备
调度线程池	ScheduledDtpExecutor	定时任务	支持周期性任务执行
有序线程池	OrderedDtpExecutor	需要顺序执行	保证相同key的任务顺序执行

配置详解

基础配置示例

dynamictp:
  enabled: true
  collectorTypes: micrometer,logging
  monitorInterval: 5
  platforms:
    - platform: ding
      urlKey: your-webhook-key
      receivers: 13800138000

  executors:
    - threadPoolName: orderServiceTp
      executorType: common
      corePoolSize: 10
      maximumPoolSize: 50
      queueCapacity: 1000
      queueType: VariableLinkedBlockingQueue
      rejectedHandlerType: CallerRunsPolicy
      keepAliveTime: 60
      notifyItems:
        - type: capacity
          enabled: true
          threshold: 80
        - type: reject
          enabled: true
          threshold: 1

配置参数说明

参数	类型	默认值	说明
threadPoolName	String	-	线程池名称，唯一标识
corePoolSize	int	1	核心线程数
maximumPoolSize	int	Integer.MAX_VALUE	最大线程数
queueCapacity	int	1024	队列容量
queueType	String	LinkedBlockingQueue	队列类型
rejectedHandlerType	String	AbortPolicy	拒绝策略
keepAliveTime	long	60	线程空闲时间(s)
notifyItems	List	-	告警配置项

接入步骤

1. 添加依赖

<dependency>
    <groupId>org.dromara</groupId>
    <artifactId>dynamic-tp-spring-boot-starter</artifactId>
    <version>1.2.2</version>
</dependency>

<dependency>
    <groupId>org.dromara</groupId>
    <artifactId>dynamic-tp-spring-boot-starter-nacos</artifactId>
    <version>1.2.2</version>
</dependency>

2. 配置中心配置

在Nacos中创建配置dynamic-tp-demo-dtp-dev.yml：

dynamictp:
  executors:
    - threadPoolName: userServiceTp
      corePoolSize: 5
      maximumPoolSize: 20
      queueCapacity: 200
      notifyItems:
        - type: capacity
          threshold: 70
        - type: reject
          threshold: 1

3. 应用配置

spring:
  application:
    name: dynamic-tp-demo
nacos:
  config:
    server-addr: 127.0.0.1:8848
    data-ids: dynamic-tp-demo-dtp-dev.yml
    auto-refresh: true

4. 启用DynamicTp

@SpringBootApplication
@EnableDynamicTp
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

5. 使用线程池

@Service
public class UserService {
    
    @Resource
    private ThreadPoolExecutor userServiceTp;
    
    public void processUserData(List<User> users) {
        users.forEach(user -> 
            userServiceTp.execute(() -> processSingleUser(user))
        );
    }
    
    private void processSingleUser(User user) {
        // 业务处理逻辑
    }
}

监控告警

监控指标

DynamicTp提供丰富的监控指标：

pie title 监控指标分类
    "线程池维度" : 40
    "队列维度" : 25
    "任务维度" : 20
    "性能指标" : 15

主要监控指标

类别	指标	说明
线程池	corePoolSize	核心线程数
线程池	maximumPoolSize	最大线程数
线程池	activeCount	活动线程数
队列	queueSize	当前队列大小
队列	remainingCapacity	剩余容量
任务	completedTaskCount	已完成任务数
任务	rejectCount	拒绝任务数

告警配置

notifyItems:
  - type: capacity              # 队列容量告警
    enabled: true
    threshold: 80               # 阈值80%
    platforms: [ding,wechat]    # 告警平台
    interval: 120               # 告警间隔(s)
  
  - type: liveness             # 线程活性告警
    enabled: true
    threshold: 80
    
  - type: reject               # 拒绝告警
    enabled: true
    threshold: 1               # 拒绝1次即告警
    
  - type: run_timeout          # 运行超时告警
    enabled: true
    threshold: 1000            # 超时阈值(ms)

高级特性

任务包装器

支持任务执行前后增强处理：

public class MdcTaskWrapper implements TaskWrapper {
    @Override
    public String name() {
        return "mdc";
    }
    
    @Override
    public Runnable wrap(Runnable runnable) {
        Map<String, String> context = MDC.getCopyOfContextMap();
        return () -> {
            if (context != null) {
                MDC.setContextMap(context);
            }
            try {
                runnable.run();
            } finally {
                MDC.clear();
            }
        };
    }
}

自定义拒绝策略

public class CustomRejectHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 自定义拒绝逻辑
        if (executor instanceof DtpExecutor) {
            DtpExecutor dtpExecutor = (DtpExecutor) executor;
            // 触发告警
            AlarmManager.tryAlarmAsync(dtpExecutor, NotifyItemEnum.REJECT);
        }
        throw new RejectedExecutionException("Task rejected");
    }
}

性能优化建议

1. 参数调优策略

flowchart TD
    A[分析业务场景] --> B{CPU密集型?}
    B -->|是| C[核心线程数 = CPU核数]
    B -->|否| D[核心线程数 = CPU核数 * 2]
    
    C --> E[最大线程数 = 核心线程数 + 10]
    D --> F[最大线程数 = 核心线程数 * 2]
    
    E --> G[队列类型: LinkedBlockingQueue]
    F --> H[队列类型: SynchronousQueue]
    
    G --> I[设置合理的队列容量]
    H --> J[设置快速拒绝策略]

2. 监控指标建议

场景	核心监控指标	告警阈值建议
CPU密集型	CPU使用率、活动线程数	CPU > 80%，活动线程 > 核心线程
IO密集型	队列大小、拒绝数	队列 > 70%，拒绝 > 0
混合型	所有指标	综合监控，分级告警

故障排查

常见问题及解决方案

问题现象	可能原因	解决方案
线程池拒绝任务	队列满且线程数达到最大值	调整队列容量或最大线程数
任务执行超时	任务处理逻辑复杂	优化任务逻辑或增加超时时间
内存溢出	队列堆积过多任务	设置合理的队列容量和拒绝策略

诊断命令

通过Spring Boot Actuator端点查看线程池状态：

# 查看所有线程池状态
curl http://localhost:8080/actuator/dynamic-tp

# 查看特定线程池详情
curl http://localhost:8080/actuator/dynamic-tp/userServiceTp

最佳实践

1. 生产环境配置

dynamictp:
  collectorTypes: micrometer
  monitorInterval: 5
  platforms:
    - platform: ding
      urlKey: ${DING_WEBHOOK_KEY}
      receivers: ${DING_RECEIVERS}

  executors:
    - threadPoolName: criticalServiceTp
      corePoolSize: 10
      maximumPoolSize: 50
      queueCapacity: 500
      notifyItems:
        - type: capacity
          threshold: 60        # 生产环境建议较低阈值
        - type: reject
          threshold: 1
        - type: run_timeout
          threshold: 5000      # 5秒超时

2. 多环境配置管理

graph LR
    Dev[开发环境] -->|宽松配置| DevConfig[core: 2, max: 10]
    Test[测试环境] -->|模拟生产| TestConfig[core: 5, max: 20]
    Prod[生产环境] -->|严格配置| ProdConfig[core: 10, max: 50]
    
    DevConfig -.-> Nacos[Nacos配置中心]
    TestConfig -.-> Nacos
    ProdConfig -.-> Nacos

总结

DynamicTp作为一个成熟的动态线程池解决方案，提供了从配置管理、运行时调整到监控告警的全套能力。通过合理的配置和使用，可以显著提升应用的稳定性和可维护性。

关键优势：

• 🎯 零代码侵入，接入简单
• 📈 实时监控，全方位可观测
• 🔔 智能告警，事前预警
• 🔧 动态调整，无需重启
• 🛡️ 生产验证，稳定可靠

建议在生产环境中逐步推广使用，先从非核心业务开始，积累经验后再应用到关键业务系统。