ThreadPoolExecutor 不香了，轻量级动态线程池真香

一、不香前言

最初设计 Hippo4j 的轻量初衷是尽可能提高以及保障线程池对于线上应用的作用，所以加了很多个性化功能，态线这也间接导致强依赖 Hippo4j Server 项目。程池

自 Hippo4j 1.0.0 版本发布之后，真香不断有社区小伙伴提出相同的不香一个问题，如何能够轻量级使用动态线程池？轻量

这不，它来了。态线

GitHub：https://github.com/acmenlt/dynamic-threadpool

Gitee：https://gitee.com/acmenlt/dynamic-threadpool

随着 Hippo4j 1.1.0 版本的程池发布，除了在原有功能进行迭代输出外，真香额外添加了一种使用模式：依赖配置中心实现的不香轻量级动态线程池，将 Hippo4j 的轻量源代码从一种使用模式拆分为两种。

两种模式共用一套核心源代码，态线保留了基础且大家关注的程池功能，模块取名为：Hippo4j Core。真香

二、Hippo4j Core

所谓“一图胜千言”，小编画了一张图，来描述它的交互行为以及所支持功能。

只要你们项目中有配置中心，引用 hippo4j-core-spring-boot-starter 后，就可以使用以上功能啦。

1. 动态线程池参数更新

客户端项目启动时向配置中心请求动态线程池配置，获取配置后创建 DynamicThreadPool 线程池。

并向配置中心发起监听事件，源码下载当配置中心中配置发生变更时，监听事件实时修改项目中的线程池参数。

如在配置中心变更了动态线程池配置，会在日志中打印变更信息：

[MESSAGE-CONSUME] Changed thread pool. 

    coreSize :: [1 => 10]

    maxSize :: [1 => 20]

    queueType :: [ResizableCapacityLinkedBlockIngQueue => ResizableCapacityLinkedBlockIngQueue]

    capacity :: [1024 => 2048]

    keepAliveTime :: [1000 => 1000]

    executeTimeOut :: [600 => 600]

    rejectedType :: [DiscardOldestPolicy => DiscardOldestPolicy]

    allowCoreThreadTimeOut :: [false => false]

同时，通过消息推送通知相关负责人。目前通知平台已支持钉钉、企业微信以及飞书三种常用办公软件，以企业微信群聊机器人举例：

2. Web 线程池参数更新

SpringBoot 内置三种 Web 容器：Tomcat、Jetty、Undertow。

Hippo4j Core 已支持容器线程池的核心参数变更：corePoolSize、maximumPoolSize、keepAliveTime。

为什么要加 Web 线程池的动态更新？两个原因：

压测应用时，需要针对不同的压测流量来调整 Web 容器线程池的线程数。正常流程，调整后需要重新发布项目，无疑是比较费时费力；当 SpringBoot Java 应用响应时间变慢，并且服务器整体负载不高时，我们可以通过修改 Web 容器线程池来提高并行处理能力，以此提高响应时间。

当然，正常来说，线上的容器线程池配置是通过压测后得出的最优值。高防服务器所以，这个功能在线上应该谨慎使用，或者说尽量不在线上使用。

3. 动态线程池报警策略

为了让线程池运行出现问题，及时通知到相关负责人，Hippo4j 针对线程池做了四种定制化报警策略：

活跃度报警：假设设置线程池活跃度报警阈值为 80%，最大线程数 10。当线程数达到 8 发起报警；阻塞队列容量报警：假设设置容量报警阈值为 80%，阻塞队列容量 100。当容量达到 80 发起报警；拒绝任务报警：当线程池无法执行任务，开始执行拒绝策略时报警；执行时间报警：假设线程池超时时间设置 1000ms，任务执行时间超过 1000ms 发起报警。

问题比较多的小伙伴就问了，如果线程池 频繁拒绝任务或者执行时间频繁超时，那岂不是要被信息轰炸？

不会的。报警策略做了优化，当设置报警间隔时间内，线程池 + 报警类型 两个维度仅会发出一条通知报警消息。

举个例子，有一个线程池 ID：message-consum 的线程池，设置了报警间隔为 5 分钟。

也就是说，活跃度、阻塞队列容量、拒绝任务、执行时间几个报警纬度，message-consum 线程池在 5分钟内最多每个类型发送一条报警通知。

目前已支持了钉钉、云服务器提供商企业微信以及飞书的群机器人报警。企业微信机器人示例如下：

上图中的链路信息只会在超时报警时存在，这样可以通过链路信息，更方便定位到线程池任务执行缓慢的原因。

三、代码示例

Nacos 或 Apollo 配置中心任选其一。

SpringBoot Pom 文件引入 Hippo4j Core Maven 坐标。

    cn.hippo4j

    hippo4j-core-spring-boot-starter

    1.1.0

</dependency>

启动类上添加 @EnableDynamicThreadPool 注解。

@SpringBootApplication

@EnableDynamicThreadPool

public class ExampleApplication {

    public static void main(String[] args) {

        SpringApplication.run(ExampleApplication.class, args);

    }

}

在配置中心中添加 spring.dynamic.thread-pool 前缀的配置。如下：

server:

  port: 8090

  servlet:

    context-path: /example

spring:

  profiles:

    active: dev

  dynamic:

    thread-pool:

      enable: true  # 是否开启动态线程池

      banner: true  # 是否打印 banner

      collect: true  # 是否开启线程池数据采集，对接 Prometheus

      check-state-interval: 3  # 检查线程池状态，是否达到报警条件，单位秒

      notify-platforms:  # 通知报警平台，支持多个，或者任选其一

        - platform: WECHAT  # 企业微信

          secret-key: 1d307bfa-815f-4662-a2e5-99415e947bb8

        - platform: DING  # 钉钉

          secret-key: 56417ebba6a27ca352f0de77a2ae9da66d01f39610b5ee8a6033c60ef9071c55

        - platform: LARK  # 飞书

          secret-key: 2cbf2808-3839-4c26-a04d-fd201dd51f9e

      nacos:  # nacos apollo 任选其一

        data-id: xxx

        group: xxx

      apollo:

        namespace: xxxx

      config-file-type: yml  # 配置中心文件格式

      executors:

        - thread-pool-id: message-consume  # 线程池标识

          core-pool-size: 1  # 核心线程数

          maximum-pool-size: 1  # 最大线程数

          queue-capacity: 1  # 阻塞队列大小

          execute-time-out: 1000  # 执行超时时间，执行任务时间超过此时间发起报警

          blocking-queue: LinkedBlockingQueue  # 阻塞队列名称，参考 QueueTypeEnum，支持 SPI

          rejected-handler: AbortPolicy  # 拒绝策略名称，参考 RejectedPolicies，支持 SPI

          keep-alive-time: 1024  # 线程存活时间，单位秒  

          allow-core-thread-time-out: true  # 是否允许核心线程超时

          thread-name-prefix: message-consume  # 线程名称前缀

          notify:  # 通知配置

            is-alarm: true  # 是否报警

            active-alarm: 80  # 活跃度报警阈值；假设线程池最大线程数 10，当线程数达到 8 发起报警

            capacity-alarm: 80  # 容量报警阈值；假设阻塞队列容量 100，当容量达到 80 发起报警

            interval: 8  # 报警间隔，同一线程池下同一报警纬度，在 interval 时间内只会报警一次，单位分钟

            receives: # 任选其一

              DING: xxx  # 手机号

              WECHAT: xxx  # 填写企业微信用户 ID（填写其它将无法达到 @ 效果）

              LARK: xxx # 填写 ou_开头的用户唯一标识，否则只能普通 @

使用 Hippo4j ThreadPoolBuilder 构建动态线程池。

import cn.hippo4j.core.executor.DynamicThreadPool;

import cn.hippo4j.core.executor.support.ThreadPoolBuilder;

@Bean

@DynamicThreadPool

public ThreadPoolExecutor dynamicThreadPoolExecutor() {

    String consumeThreadPoolId = "message-consume";

    return ThreadPoolBuilder.builder()

            .threadFactory(consumeThreadPoolId)

            .dynamicPool()

            .build();

}

按照 Spring Bean 注入的方式使用动态线程池即可。

@Resource

private ThreadPoolExecutor dynamicThreadPoolExecutor;

dynamicThreadPoolExecutor.execute(() -> xxx);

没了，是不是很 easy？我大致试了下，不到两分钟的时间，就能让你的 SpringBoot 项目快速接入动态线程池。

总结下接入步骤：

Pom 中引入 Hippo4j Core 包依赖；启动类上添加动态线程池启用注解；配置中心（Nacos 或 Apollo）添加动态线程池配置；项目中以 Spring Bean 的形式创建动态线程池 。

四、常见问题

1. 项目关闭时，如何保障线程池中任务全部完成

答：借鉴了 Spring 封装的线程池框架。构建动态线程池时，指定 waitForTasksToCompleteOnShutdown 和 awaitTerminationMillis

import cn.hippo4j.core.executor.DynamicThreadPool;

import cn.hippo4j.core.executor.support.ThreadPoolBuilder;

@Bean

@DynamicThreadPool

public ThreadPoolExecutor dynamicThreadPoolExecutor() {

    String consumeThreadPoolId = "message-consume";

    return ThreadPoolBuilder.builder()

            .threadFactory(consumeThreadPoolId)

            .waitForTasksToCompleteOnShutdown(true)

            .awaitTerminationMillis(5000L)

            .dynamicPool()

            .build();

}

这两个参数什么意思呢？

waitForTasksToCompleteOnShutdown：是否在关闭线程池时等待任务完成，这里我们设置 true；awaitTerminationMillis：等待任务完成的时间，单位毫秒。

问题很多的小伙伴可能就问了：为啥要有 awaitTerminationMillis  这个参数？直接等待全部任务完成不就行了。

线程池中都是执行很快的任务可能是没问题。但是，如果线程池里面都是耗时的任务呢？

停止项目时等个几分钟甚至更长时间是无法忍受的。这个需要根据大家项目的实际情况评估。

2. 动态线程池是否可以传递上下文参数

可以的，同样是借鉴 Spring 线程池框架。实现 TaskDecorator 接口，并在构建动态线程池时指定。

import cn.hippo4j.core.executor.DynamicThreadPool;

import cn.hippo4j.core.executor.support.ThreadPoolBuilder;

@Bean

@DynamicThreadPool

public ThreadPoolExecutor dynamicThreadPoolExecutor() {

    String consumeThreadPoolId = "message-consume";

    return ThreadPoolBuilder.builder()

            .threadFactory(consumeThreadPoolId)

            .waitForTasksToCompleteOnShutdown(true)

            .awaitTerminationMillis(5000L)

            .taskDecorator(new TaskDecoratorTest.ContextCopyingDecorator())

            .dynamicPool()

            .build();

}3. Hippo4j Core 和 Hippo4j Server 转换麻烦么

有些小伙伴最初用的 Hippo4j Core，觉得功能并不满足使用，想要使用 Hippo4j Server，问我如何转换？

其实这里非常简单。从项目上来说：代码无需任何改变，把 Pom 文件中的依赖坐标改下就可以。

其次就是将配置中心里的配置迁移到 Hippo4j 的控制台，将线程池记录创建才出来即可。

五、文末总结

文章介绍了 Hippo4j 新增的一种使用模式：依赖配置中心的轻量动态线程池的实现。

不太好评价 Hippo4j Server 和 Hippo4j Core 的好坏。一个是功能更强大，一个是引入更加轻量，使用上具体如何，交给使用者来评价。

如果项目中使用了 配置中心以及线程池 的话，强烈推荐大家引入到项目中试一试，为项目线上的稳定性多了一份保障。

很赞哦!（5）