ParallelFlux<T> Reactor 3 的 .publish(Function<T, R>) 等效项答案

【问题标题】：Equivalent of .publish(Function<T, R>) for ParallelFlux<T> Reactor 3ParallelFlux<T> Reactor 3 的 .publish(Function<T, R>) 等效项
【发布时间】：2021-09-25 10:09:53
【问题描述】：

我有一些这样的 Reactor Flux：

SomeFlux.map(thing -> makeOtherThing(thing))
        .publish(fluxFunction)
        .subscribe();

.publish() 的参数是 fluxFunction，它是 Java 8
Function<T, R> 功能接口的实现对象，如下所示：

public class FluxFunction implements Function<Flux<OtherThing>, Flux<DiffOtherThing>> {
  @Override
  Flux<DiffOtherThing> apply(Flux<OtherThing>) {
    // some code which "consumes" incoming Flux and outputs another
  }
}

我的问题是，如果我将上面的代码更改为：

SomeFlux.parallel()
        .runOn(Schedulers.parrallel())
        .map(thing -> makeOtherThing(thing))
        .publish(fluxFunction)
        .subscribe();

方法 .publish(fluxFunction) 无效，因为我现在正在处理 ParallelFlux，是否有一些等效于 .publish(Function) 的方法可以在 parallelFlux 上使用？如果是这样，我将如何编辑我的 Function 实现以适应这一点？

【问题讨论】：

这里有更多的上下文会很好，为什么是并行通量，makeOtherThing 做什么以及为什么在ParallelFlux 上需要publish。用户案例？

标签： java spring-boot reactive-programming spring-webflux project-reactor

【解决方案1】：

所以看起来是等价的（至少在我的情况下）是：

SomeFlux.parallel()
        .runOn(Schedulers.parallel())
        .map(thing -> makeOtherThing(thing))
        .as(fluxFunction) // changed ".publish()" to ".as()" 
        .subscribe()

我还将fluxFunction 重新定义为：

// Now operates on ParallelFlux<> rather than Flux<>
public class FluxFunction implements Function<ParallelFlux<OtherThing>, ParallelFlux<DiffOtherThing>> {
  @Override
  ParallelFlux<DiffOtherThing> apply(ParallelFlux<OtherThing>) {
    // some code which "consumes" incoming ParallelFlux and outputs another
  }
}

这似乎对我来说非常有效。

我希望这可以帮助任何发现自己陷入类似困境的人。

【讨论】：