Spring Integration 5.0 + Project Reactor：控制线程

Question

https://stackoverflow.com/a/47136941/1776585 的后续问题

在使用 Flux + split() + FluxMessageChannel 时，我无法让我的集成处理程序在并行线程中运行。

考虑以下 sn-p：

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

所有日志都在一个线程中输出：

[     parallel-1] d.a.Application    : > 0
[     parallel-1] d.a.Application    :  -> 0
[     parallel-1] d.a.Application    : > 1
[     parallel-1] d.a.Application    :  -> 1
[     parallel-1] d.a.Application    : > 2
[     parallel-1] d.a.Application    :  -> 2
[     parallel-1] d.a.Application    : > 3
[     parallel-1] d.a.Application    :  -> 3
[     parallel-1] d.a.Application    : > 4
[     parallel-1] d.a.Application    :  -> 4
[     parallel-1] d.a.Application    : > 5
[     parallel-1] d.a.Application    :  -> 5
[     parallel-1] d.a.Application    : > 6
[     parallel-1] d.a.Application    :  -> 6
[     parallel-1] d.a.Application    : > 7
[     parallel-1] d.a.Application    :  -> 7
[     parallel-1] d.a.Application    : > 8
[     parallel-1] d.a.Application    :  -> 8
[     parallel-1] d.a.Application    : > 9
[     parallel-1] d.a.Application    :  -> 9

如何强制多线程处理？

我已经尝试在Flux 上使用.parallel().runOn()，但这只会使获取数据并行，但实际处理仍然在一个线程上运行。

我也在Flux 上尝试了.publishOn(Schedulers.parallel())，但没有效果。

并且还将ExecutorChannel 或Poller 与执行程序添加到处理程序没有帮助。

Answer 1

这有一些技巧：

.channel(new FluxMessageChannel())
.channel(MessageChannels.executor(Executors.newCachedThreadPool()))
.handle(message -> LOG.info(" -> " + message.getPayload())))

FluxMessageChannel 使用的那些消息将与额外的ExecutorChannel 并行。

我认为您的要求类似于功能请求，以使上述FluxMessageChannel 可配置。而这样的subscribeOn/publishOn等可以在那里配置。

请随时就此事提出JIRA！