Spring Integration+Reactor：Flux+split+FluxMessageChannel无法并行执行处理器

我来帮你拆解下这个问题——你遇到的核心卡点是第一个subscribeOn的作用范围没覆盖到后续的消息处理环节，导致所有日志都挤在同一个线程里输出。

先看你的代码逻辑：

.handle(message -> Flux.range(0, 10)
        .doOnNext(i -> LOG.info("> " + i))
        .subscribeOn(Schedulers.parallel()))
.split()
.channel(new FluxMessageChannel())
.handle(message -> LOG.info(" -> " + message.getPayload()))

这里的subscribeOn(Schedulers.parallel())只对Flux.range(0,10)的元素生成过程生效，也就是生成0-9这些数值时会用到并行线程，但split()把这些元素拆成独立消息后，FluxMessageChannel默认是同步传递消息的，所以第二个handle会直接复用第一个handle的线程，自然看不到并行执行的效果。

给你两个可行的解决方案：

方案一：给FluxMessageChannel指定并行调度器 #

直接在创建FluxMessageChannel时传入并行调度器，让通道异步把消息分发到不同线程：

.handle(message -> Flux.range(0, 10)
        .doOnNext(i -> LOG.info("> " + i))
        .subscribeOn(Schedulers.parallel()))
.split()
.channel(new FluxMessageChannel(Schedulers.parallel())) // 这里指定并行调度器
.handle(message -> LOG.info(" -> " + message.getPayload()))

方案二：在第二个handle前切换调度上下文 #

如果不想修改通道配置，可以在第二个handle前用publishOn切换到并行线程池，确保后续处理逻辑跑在不同线程：

.handle(message -> Flux.range(0, 10)
        .doOnNext(i -> LOG.info("> " + i))
        .subscribeOn(Schedulers.parallel()))
.split()
.channel(new FluxMessageChannel())
.publishOn(Schedulers.parallel()) // 切换到并行线程上下文
.handle(message -> LOG.info(" -> " + message.getPayload()))

两种方案都能让第二个handle的日志输出在不同的并行线程里，你可以根据自己的业务场景选择合适的方式。

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