Reactor项目：我需要处理器吗？

Question

我试图在Reactor顶部设计一个管道框架。

在每个阶段（不考虑第一个和最后一个阶段），我们都有转换对象的任务（即字符串到其长度或url到其HTML内容等）。举个例子：

您可以看到中间层有3个任务，每个任务将一个X对象转换为一个Y对象（顺便说一句，它始终是一个完全连接的层）

我的问题/困境：我的第一个想法是，我所需要的是通量。merge（），然后将其连接到每个订阅者。例如：

Flux<X> source = Flux.merge(x1Flux, x2Flux)  
source.subscribe(y1Subscriber)
source.subscribe(y2Subscriber)

另一种选择是放置处理器（TopicProcessor？）这将充当中间件（如pub-sub模式中的中间件）

我不知道哪种解决方案最适合我的问题。逻辑上是一样的，但每种架构的实际含义是什么？

谢谢

Answer 1

我在这里的一般方法是使用ConnectableFlux来延迟发布，直到您完成整个管道设置，然后在设置管道后对每个通量调用connect（）。

你可以使用处理器，但我建议尽可能避免使用。

一般要点（未检查语法）类似于：

ConnectableFlux<String> x1 = Flux.just("x1").publish();
ConnectableFlux<String> x2 = Flux.just("x2").publish();

ConnectableFlux<String> y1 = Flux.<String>from(Flux.merge(x1, x2)).publish();
ConnectableFlux<String> y2 = Flux.<String>from(Flux.merge(x1, x2)).publish();
ConnectableFlux<String> y3 = Flux.<String>from(Flux.merge(x1, x2)).publish();

ConnectableFlux<String> z3 = Flux.<String>from(Flux.merge(y1, y2, y3)).publish();

x1.connect();
x2.connect();
y1.connect();
//...etc.

另请注意，您可能希望使用concat（）或mergeSequential（）而不是合并（），具体取决于您的用例（合并（）将急切地订阅发布者，concat（）不会，mergeSequential（）将按收到的顺序合并，可能会交错值。）