目录

• 系列总目录
• 背景
• 简单实现
• 美团开源方案DynamicTp
  • 架构
    • 监控模块
    • 通知告警模块
  • 特性
• 参考文章

---

系列总目录

• 业务场景实战汇总

---

背景

• 对于线程池核心参数的配置很大程度上依靠经验。然而，由于系统运行过程中存在的不确定性，我们很难一劳永逸地规划一个合理的线程池参数。在对线程池配置参数进行调整时，一般需要对服务进行重启，这样修改的成本就会偏高。一种解决办法就是，将线程池的配置放到平台侧，运行开发同学根据系统运行情况对核心参数进行动态配置
• 这里有两种实现方案，一种是用配置中心比如Nacos，Apollo监听配置的变化更新线程池参数查看最合适的。另一种是用美团开源的线程池框架DynamicTp

---

简单实现

• 根据配置中心变更配置动态 + 线程池setCorePoolSize底层支持，便可实现动态变更。setCorePoolSize根据需要增加或者减少核心线程数量，动态增加或者减少线程队列对应的Worker数量，核心代码

@Configuration
public class DynamicThreadPool implements InitializingBean {
   @Override
    public void afterPropertiesSet() throws Exception {
        // 按照配置中心初始化线程池，这里简单举例nacos
        threadPoolExecutor = new ThreadPoolExecutor(Integer.parseInt(coreSize), Integer.parseInt(maxSize), 10L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10),
                new ThreadFactoryBuilder().setNameFormat("c_t_%d").build(),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println("rejected!");
                    }
                });
 
        // 根据配置变更监听从而改变线程池参数，这里举例nacos，apollo思路也是一样只是api不同
        nacosConfigManager.getConfigService().addListener("order-service-dev.yml", nacosConfigProperties.getGroup(),
                new Listener() {
                    @Override
                    public Executor getExecutor() {
                        return null;
                    }
 
                    @Override
                    public void receiveConfigInfo(String configInfo) {
                        //配置变更，修改线程池配置
                        System.out.println(configInfo);
                        changeThreadPoolConfig(Integer.parseInt(coreSize);
                    }
                });
    }
   /**
     * 修改线程池核心参数
     *
     * @param coreSize
     * @param maxSize
     */
    private void changeThreadPoolConfig(int coreSize) {
        threadPoolExecutor.setCorePoolSize(coreSize);
    }
}

---

美团开源方案DynamicTp

架构

• 主要分为几个模块
  1.配置变更监听模块
  2.服务内部线程池管理模块
  3.三方组件线程池管理模块
  4.监控模块
  5.通知告警模块

  
  架构.png

监控模块

• 实现监控指标采集以及输出，默认提供以下三种方式，也可通过内部提供的 SPI 接口扩展其他实现

1. 默认实现 JsonLog 输出到磁盘，可以自己采集解析日志，存储展示，比如Flume采集日志到Kafka，然后Kafka分析
2. MicroMeter采集，引入 MicroMeter 相关依赖，暴露相关端点，采集指标数据，结合 Grafana 做监控大盘，默认这种模式
3. 暴露自定义 Endpoint 端点（dynamic-tp），可通过 http 方式实时访

• 一般采用第二点prometheus+grafana监控展示和采集

通知告警模块

• 对接办公平台，实现通知告警功能，已支持钉钉、企微、飞书、邮件，可通过内部提供的 SPI 接口扩展其他实现
• 服务启动后会开启一个定时监控任务，每隔一定时间（可配置）去计算线程池的活跃度，达到配置的 threshold 阈值后会触发一次告警

特性

• 相对自己实现，DynamicTp增加了很多特性

1. 代码零侵入
2. 通知告警
3. 运行监控
4. 任务增强, 支持上下文传递
5. 多配置中心支持，Nacos、Apollo、Zookeeper、Consul、Etcd
6. 中间件线程池管理

• 通知告警模块

  
  告警1.png
  告警2.png

• 监控

  
  监控.png
• 暴露 EndPoint 端点(dynamic-tp)，可以通过 http 方式请求

[
    {
        "dtp_name": "remoting-call",
        "core_pool_size": 6,
        "maximum_pool_size": 12,
        "queue_type": "SynchronousQueue",
        "queue_capacity": 0,
        "queue_size": 0,
        "fair": false,
        "queue_remaining_capacity": 0,
        "active_count": 0,
        "task_count": 21760,
        "completed_task_count": 21760,
        "largest_pool_size": 12,
        "pool_size": 6,
        "wait_task_count": 0,
        "reject_count": 124662,
        "reject_handler_name": "CallerRunsPolicy"
    }
]

---

参考文章

• 利用 Nacos 实现了一个动态化线程池，非常实用！
• Dynamic-tp gitee地址
• Dynamic-tp官方文档