如何用Micrometer监控Reactor调度器中BoundedElasticScheduler线程池

内置指标补充说明 #

Reactor 3.5+针对BoundedElasticScheduler暴露了一组更贴合需求的指标，你可能没注意到：

• reactor.scheduler.boundedelastic.pool.size：当前线程池的实际大小
• reactor.scheduler.boundedelastic.pool.max：配置的线程池最大容量
• reactor.scheduler.boundedelastic.tasks.pending：等待队列中的任务数量

另外，针对你提到的3.4.x版本executor.active指标轮询精度低的问题，可以通过Micrometer的TimeWindowMax对该指标做二次处理——主动以短间隔（如100ms）采样并记录时间窗口内的峰值，弥补监控工具轮询间隔过大的缺陷，不用依赖工具本身的聚合能力。

自定义精准监控实现 #

如果内置指标仍无法满足需求，直接基于BoundedElasticScheduler的内部API实现自定义监控，能精准捕获真实线程使用情况：

1. 获取线程池核心状态 #

BoundedElasticScheduler自带metrics()方法，返回的SchedulerMetrics对象能直接拿到核心数据：

• activeWorkers()：当前活跃线程数（真实反映线程池负载，包含嵌套Mono/Flux的线程占用）
• pendingTasks()：等待队列中的任务数
• maxWorkers()：线程池的最大配置值

2. 用Micrometer实现峰值监控 #

通过定时采样记录峰值，示例代码如下：

// 初始化目标调度器
BoundedElasticScheduler scheduler = BoundedElasticScheduler.create("my-bounded-scheduler");
MeterRegistry meterRegistry = new SimpleMeterRegistry();

// 注册时间窗口峰值指标（比如记录5分钟内的活跃线程数峰值，100ms采样一次）
TimeWindowMax activeWorkersPeak = TimeWindowMax.builder("bounded_elastic.active_workers.peak")
        .description("Peak number of active workers in the BoundedElasticScheduler")
        .timeWindow(Duration.ofMinutes(5))
        .register(meterRegistry);

// 注册当前活跃线程数、等待任务数的Gauge指标
Gauge.builder("bounded_elastic.active_workers", scheduler.metrics(), SchedulerMetrics::activeWorkers)
        .register(meterRegistry);
Gauge.builder("bounded_elastic.pending_tasks", scheduler.metrics(), SchedulerMetrics::pendingTasks)
        .register(meterRegistry);

// 启动定时采样任务
ScheduledExecutorService sampler = Executors.newSingleThreadScheduledExecutor(r -> {
    Thread t = new Thread(r);
    t.setDaemon(true);
    t.setName("scheduler-metrics-sampler");
    return t;
});
sampler.scheduleAtFixedRate(() -> {
    int currentActive = scheduler.metrics().activeWorkers();
    activeWorkersPeak.record(currentActive);
}, 0, 100, TimeUnit.MILLISECONDS);

3. 跟踪线程池上限触发次数 #

如果需要统计线程池达到最大容量的频率，可以添加一个计数器指标：

Counter poolMaxHitCounter = Counter.builder("bounded_elastic.pool_max_hits")
        .description("Number of times the thread pool reached its maximum capacity")
        .register(meterRegistry);

// 在采样逻辑中加入判断
if (scheduler.metrics().activeWorkers() == scheduler.metrics().maxWorkers()) {
    poolMaxHitCounter.increment();
}

针对已尝试方案的优化建议 #

• 对于Reactor 3.4.x的executor.active精度问题：放弃依赖监控工具的轮询聚合，改为在应用内部用TimeWindowMax主动短间隔采样，能精准捕获短任务的峰值。
• 对于自定义DistributedSummary的不足：不要仅跟踪调度操作，直接使用SchedulerMetrics提供的activeWorkers()数据，它能反映线程池的真实负载，包括嵌套Mono/Flux执行时的线程占用。

内容的提问来源于stack exchange，提问作者Jens