SpringBoot 自定义线程池 + 实时监控指标

1 核心知识点回顾

JDK原生线程池核心构造参数

new ThreadPoolExecutor(
corePoolSize, // 核心线程数
maximumPoolSize, // 最大线程数
keepAliveTime, // 非核心线程空闲存活时间
unit,
workQueue, // 阻塞队列
handler // 拒绝策略
);

监控要采集哪些核心指标

2、 Maven依赖

SpringBoot2.x/3.x内置Micrometer，配合Actuator暴露监控端点，Prometheus+Grafana可视化；

<!– web项目基础依赖 –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!– 监控端点暴露 –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!– prometheus格式指标输出 –>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

3、 yml配置

spring:
application:
name: custom–thread–pool–demo

# 暴露全部actuator端点
management:
endpoints:
web:
exposure:
include: '*'
metrics:
export:
prometheus:
enabled: true
endpoint:
health:
show-details: always

# 自定义线程池业务配置
thread-pool:
custom:
core-size: 5
max-size: 10
queue-capacity: 100
keep-alive-seconds: 60

4、 自定义线程池配置类 + 注册Micrometer监控指标

4.1 配置属性绑定类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Data
@Component
@ConfigurationProperties(prefix = "thread-pool.custom")
public class CustomThreadPoolProperties {
private Integer coreSize;
private Integer maxSize;
private Integer queueCapacity;
private Long keepAliveSeconds;
}

4.2 线程池配置 + 指标注册

import io.micrometer.core.instrument.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Configuration
public class ThreadPoolConfig {

private final CustomThreadPoolProperties poolProps;
private final MeterRegistry meterRegistry;

public ThreadPoolConfig(CustomThreadPoolProperties poolProps, MeterRegistry meterRegistry) {
this.poolProps = poolProps;
this.meterRegistry = meterRegistry;
}

/**
* 自定义业务线程池
*/
@Bean("businessThreadPool")
public Executor businessThreadPool() {
// 1. 构造原生JDK线程池
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(poolProps.getQueueCapacity());
RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();

ThreadPoolExecutor executor = new ThreadPoolExecutor(
poolProps.getCoreSize(),
poolProps.getMaxSize(),
poolProps.getKeepAliveSeconds(),
TimeUnit.SECONDS,
queue,
new CustomThreadFactory("business-pool-"),
handler
);

// 2. 注册所有监控指标到micrometer
registerThreadPoolMetrics(executor, "business_thread_pool");

return executor;
}

/**
* 把线程池所有运行指标注册到监控系统
*/
private void registerThreadPoolMetrics(ThreadPoolExecutor executor, String poolName) {
Tags tags = Tags.of("pool_name", poolName);

// 1. 静态配置指标
Gauge.builder("thread.pool.core.size", executor, ThreadPoolExecutor::getCorePoolSize)
.tags(tags)
.register(meterRegistry);
Gauge.builder("thread.pool.max.size", executor, ThreadPoolExecutor::getMaximumPoolSize)
.tags(tags)
.register(meterRegistry);
Gauge.builder("thread.pool.queue.capacity", () -> poolProps.getQueueCapacity())
.tags(tags)
.register(meterRegistry);

// 2. 动态实时运行指标
// 当前活跃线程数
Gauge.builder("thread.pool.active.threads", executor, ThreadPoolExecutor::getActiveCount)
.tags(tags)
.register(meterRegistry);
// 队列剩余容量
Gauge.builder("thread.pool.queue.remaining", queue -> queue.remainingCapacity(), executor.getQueue())
.tags(tags)
.register(meterRegistry);
// 队列积压任务数
Gauge.builder("thread.pool.queue.size", queue -> queue.size(), executor.getQueue())
.tags(tags)
.register(meterRegistry);
// 已完成任务总数
Gauge.builder("thread.pool.completed.task.count", executor, ThreadPoolExecutor::getCompletedTaskCount)
.tags(tags)
.register(meterRegistry);
// 总提交任务数
Gauge.builder("thread.pool.total.task.count", executor, ThreadPoolExecutor::getTaskCount)
.tags(tags)
.register(meterRegistry);

// 3. 拒绝任务计数器
AtomicLong rejectCount = new AtomicLong(0);
executor.setRejectedExecutionHandler((r, e) -> {
rejectCount.incrementAndGet();
handler(r, e);
});
FunctionCounter.builder("thread.pool.reject.count", rejectCount, AtomicLong::get)
.tags(tags)
.register(meterRegistry);
}

/**
* 自定义线程工厂：指定线程前缀，方便日志排查
*/
static class CustomThreadFactory implements ThreadFactory {
private final String prefix;
private final AtomicLong num = new AtomicLong(0);

public CustomThreadFactory(String prefix) {
this.prefix = prefix;
}

@Override
public Thread newThread(Runnable r) {
String threadName = prefix + num.getAndIncrement();
Thread t = new Thread(r, threadName);
t.setDaemon(false);
return t;
}
}
}

5、 测试线程池使用

5.1 注入线程池执行业务任务

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Random;

@Component
public class ThreadPoolTestTask {

@Resource(name = "businessThreadPool")
private Executor threadPool;

private final Random random = new Random();

// 定时不断提交任务，制造队列积压、线程活跃
@Scheduled(fixedRate = 200)
public void submitTask() {
threadPool.execute(() -> {
try {
// 模拟业务耗时 100~500ms
Thread.sleep(100 + random.nextInt(400));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}

6、 查看监控指标

6.1 Prometheus指标地址

启动项目后访问：

http://127.0.0.1:8080/actuator/prometheus

可以直接抓取所有线程池实时指标，示例指标片段：

# HELP thread_pool_active_threads Gauge measuring active threads of thread pool
# TYPE thread_pool_active_threads gauge
thread_pool_active_threads{pool_name="business_thread_pool",} 5.0
# HELP thread_pool_queue_size Gauge measuring pending task count in queue
thread_pool_queue_size{pool_name="business_thread_pool",} 23.0
# HELP thread_pool_reject_count Counter tracking rejected task number
thread_pool_reject_count{pool_name="business_thread_pool",} 0.0

6.2 指标字段说明