Reactor 3中ParallelFlux<T>下.publish(Function<T,R>)的等效方法是什么

核心原因 #

普通 Flux 提供的publish(Function)重载是针对普通序列的多播自定义处理API，ParallelFlux 没有提供同名同功能的重载，且你现有的FluxFunction入参类型为Flux<OtherThing>，与ParallelFlux类型不匹配，因此调用报错。

两种适配方案 #

可根据业务是否需要保留并行处理特性选择对应方案：

方案1：保留并行特性，适配Function实现 #

ParallelFlux 提供的as()方法与普通Flux的publish(Function)功能等效，可将当前ParallelFlux实例传入自定义函数处理。
需要做两处调整：

1. 修改自定义Function的入参类型为ParallelFlux：

public class ParallelFluxFunction implements Function<ParallelFlux<OtherThing>, ParallelFlux<DiffOtherThing>> {
  @Override
  ParallelFlux<DiffOtherThing> apply(ParallelFlux<OtherThing> parallelFlux) {
    // 原有逻辑调整为操作ParallelFlux即可，ParallelFlux同样支持map/filter等常用操作符
  }
}

2. 替换原调用的publish()为as()：

SomeFlux.parallel()
        // 注意修正原拼写错误：parallel 不是 parrallel
        .runOn(Schedulers.parallel())
        .map(thing -> makeOtherThing(thing))
        .as(new ParallelFluxFunction())
        .subscribe();

如果原有publish(Function)是为了利用多播特性避免上游重复订阅，可以配合多播API实现完全等效效果：

SomeFlux.parallel()
        .runOn(Schedulers.parallel())
        .map(thing -> makeOtherThing(thing))
        .publish() // 开启并行流多播
        .autoConnect() // 自动连接上游
        .as(new ParallelFluxFunction())
        .subscribe();

方案2：合并并行流为普通Flux，复用原有逻辑 #

如果Function处理阶段不需要保持并行特性，可以先调用sequential()将ParallelFlux转换回普通Flux，无需修改原有FluxFunction实现：

SomeFlux.parallel()
        .runOn(Schedulers.parallel())
        .map(thing -> makeOtherThing(thing))
        .sequential() // 合并并行轨道为普通单流
        .publish(fluxFunction) // 直接复用原有Function实现
        .subscribe();

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