如何让Flux.interval的每个tick处理任务运行在独立新线程

问题根因 #

你当前的写法中doOnNext会在信号所在线程同步执行阻塞逻辑，直接占用了Flux.interval的发射调度线程，所以前一个something()执行完成前，后续的tick信号无法被处理，最终表现为调用频率被拉长到和处理耗时一致。

正确实现方案 #

将耗时的something()逻辑异步化，避免阻塞上游tick发射，同时通过flatMap的并发参数动态控制最大并行处理量，修改后代码如下：

// 可根据业务需求调整最大并行处理数，超过阈值的任务会触发背压丢弃
int MAX_PARALLEL_TASK = 10;
Flux.interval(Duration.ZERO, Duration.ofSeconds(1))
    .onBackpressureDrop()
    .flatMap(counter -> 
        // 封装阻塞逻辑为异步Publisher
        Mono.fromRunnable(() -> something())
            // 每个任务独立从有界弹性线程池获取线程运行
            .subscribeOn(Schedulers.boundedElastic())
            // 单个任务异常不中断主流程，可自行添加异常处理逻辑
            .onErrorResume(e -> {
                // 这里处理单个something()的执行异常
                return Mono.empty();
            }),
        // 配置最大并行度，动态限制同时运行的任务数量
        MAX_PARALLEL_TASK
    )
    .onErrorContinue((throwable, o) -> {
        // 全局异常兜底处理
    })
    .doOnComplete(() -> {
        // 流程结束处理逻辑
    })
    .subscribe();

核心说明 #

• doOnNext仅适用于日志打印、指标埋点等轻量操作，禁止在该操作符中执行耗时/阻塞逻辑
• 每个tick对应的任务都会独立申请线程执行，只要未达到最大并行度、线程池有空闲资源，每秒触发的tick都会立即提交任务运行，不会被之前未完成的任务阻塞
• 可随时调整MAX_PARALLEL_TASK的数值调整并行处理能力，相比固定多订阅者的方式灵活性更高

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