輕量級動態線程池才是“王道”？

作者：龍臺 2022-03-14 08:02:08

開發前端

如果項目中使用了配置中心以及線程池的話，強烈推薦大家引入到項目中試一試，為項目線上的穩定性多了一份保障。

大家好，我是龍臺。

一、前言

最初設計 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 坐標。

<dependency>
    <groupId>cn.hippo4j</groupId>
    <artifactId>hippo4j-core-spring-boot-starter</artifactId>
    <version>1.1.0</version>
</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 的好壞。一個是功能更強大，一個是引入更加輕量，使用上具體如何，交給使用者來評價。