Skip to main content
Version: Next

接入流程

Nacos、Apollo、Zookeeper、ETCD、Polaris、Consul 配置中心任选其一。

hippo4j 配置

<dependency>
<groupId>cn.hippo4j</groupId>
<artifactId>hippo4j-config-spring-boot-starter</artifactId>
<version>1.4.3-upgrade</version>
</dependency>

启动类上添加注解 @EnableDynamicThreadPool

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

SpringBoot 应用配置文件添加:

server:
port: 8090
servlet:
context-path: /example

spring:
profiles:
active: dev

dynamic:
thread-pool:
# 是否开启动态线程池
enable: true
# 是否打印 banner
banner: true
# 是否开启线程池数据采集,对接 Micrometer、ES、Log 等
collect: true
# 检查线程池状态,是否达到报警条件,单位毫秒
check-state-interval: 3000
# 通知报警平台,请替换为自己创建的群机器人
notify-platforms:
- platform: 'WECHAT'
token: xxx
- platform: 'DING'
token: xxx
secret: xxx # 加签专属
- platform: 'LARK'
token: xxx
# Nacos、Apollo、Zookeeper、ETCD、Polaris、Consul 任选其一
nacos:
data-id: xxx
group: xxx
apollo:
namespace: xxxx
# 配置中心文件格式
config-file-type: yml
# tomcat、undertow、jetty 三种容器线程池,任选其一
undertow:
core-pool-size: 100
maximum-pool-size: 200
keep-alive-time: 1000
# 全局通知配置-是否报警
alarm: true
# 活跃度报警阈值;假设线程池最大线程数 10,当线程数达到 8 发起报警
active-alarm: 80
# 容量报警阈值;假设阻塞队列容量 100,当容量达到 80 发起报警
capacity-alarm: 80
# 报警间隔,同一线程池下同一报警纬度,在 interval 时间内只会报警一次,单位秒
alarm-interval: 8
# 企业微信填写用户 ID(填写其它将无法达到 @ 效果)、钉钉填手机号、飞书填 ou_ 开头唯一 ID
receives: xxx
# 动态线程池列表
executors:
- thread-pool-id: 'message-consume'
# 核心线程数
core-pool-size: 1
# 最大线程数
maximum-pool-size: 1
# 阻塞队列名称,参考 BlockingQueueTypeEnum,支持 SPI
blocking-queue: 'LinkedBlockingQueue'
# 阻塞队列大小
queue-capacity: 1
# 执行超时时间,超过此时间发起报警,单位毫秒
execute-time-out: 1000
# 拒绝策略名称,参考 RejectedPolicyTypeEnum,支持 SPI
rejected-handler: 'AbortPolicy'
# 线程存活时间,单位秒
keep-alive-time: 1024
# 是否允许核心线程超时
allow-core-thread-time-out: true
# 线程工厂名称前缀
thread-name-prefix: 'message-consume'
# 是否报警
alarm: true
# 活跃度报警阈值;假设线程池最大线程数 10,当线程数达到 8 发起报警
active-alarm: 80
# 容量报警阈值;假设阻塞队列容量 100,当容量达到 80 发起报警
capacity-alarm: 80
# 通知配置,线程池中通知配置如果存在,则会覆盖全局通知配置
notify:
# 报警间隔,同一线程池下同一报警纬度,在 interval 时间内只会报警一次,单位分钟
interval: 8
# 企业微信填写用户 ID(填写其它将无法达到 @ 效果)、钉钉填手机号、飞书填 ou_ 开头唯一 ID
receives: xxx
- thread-pool-id: 'message-produce'
core-pool-size: 1
maximum-pool-size: 1
queue-capacity: 1
execute-time-out: 1000
blocking-queue: 'LinkedBlockingQueue'
rejected-handler: 'AbortPolicy'
keep-alive-time: 1024
allow-core-thread-time-out: true
thread-name-prefix: 'message-consume'
alarm: true
active-alarm: 80
capacity-alarm: 80
notify:
interval: 8
receives: xxx

ThreadPoolExecutor 适配

添加线程池配置类,通过 @DynamicThreadPool 注解修饰。threadPoolId 为服务端创建的线程池 ID。

package cn.hippo4j.example;

import cn.hippo4j.core.executor.DynamicThreadPool;
import cn.hippo4j.core.executor.support.ThreadPoolBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
public class ThreadPoolConfig {

@Bean
@DynamicThreadPool
public ThreadPoolExecutor messageConsumeDynamicExecutor() {
String threadPoolId = "message-consume";
ThreadPoolExecutor messageConsumeDynamicExecutor = ThreadPoolBuilder.builder()
.threadFactory(threadPoolId)
.threadPoolId(threadPoolId)
.dynamicPool()
.build();
return messageConsumeDynamicExecutor;
}

@Bean
@DynamicThreadPool
public ThreadPoolExecutor messageProduceDynamicExecutor() {
String threadPoolId = "message-produce";
ThreadPoolExecutor messageProduceDynamicExecutor = ThreadPoolBuilder.builder()
.threadFactory(threadPoolId)
.threadPoolId(threadPoolId)
.dynamicPool()
.build();
return messageProduceDynamicExecutor;
}

}

通过 ThreadPoolBuilder 构建动态线程池,只有 threadFactory、threadPoolId 为必填项,其它参数会从配置中心拉取。

项目中使用上述定义的动态线程池,如下所示:

@Resource
private ThreadPoolExecutor messageConsumeDynamicExecutor;

messageConsumeDynamicExecutor.execute(() -> xxx);

@Resource
private ThreadPoolExecutor messageProduceDynamicExecutor;

messageProduceDynamicExecutor.execute(() -> xxx);