Spring Reactor如何避免对相似订阅者重复映射操作？

Question

我有一个发出字符串的发布者和许多可能使用相同映射函数来创建具有不同过滤器的模型的订阅者。

出版商：

val publisher: Flux<String> = ...

订阅者#1

val sub1 = publisher.map{veryExpensiveConverter.convert(it)}
                    .filter(it.metric<10)

订阅者#2

val sub2 = publisher.map{veryExpensiveConverter.convert(it)}
                    .filter(it.metric>5)

订阅者#3

val sub3 = sub2.map{cheapConverter.convert(it)}
                    .filter(it.metric>8)

订阅者#4

val sub4 = sub3.map{yetAnotherConverter.convert(it)}
                    .filter(it.metric>80)

最后我订阅了所有fluxes

Flux.merge(sub1, sub2, sub3, ..., subn)
     .map{//some logic for following data of subscribers}
     .subscribe()

问题：veryExpensiveConverter 对每个订阅者的相同发布记录执行多次。执行流程看起来

Input1 -> veryExpensiveConverter -> filter1 -> output1
       -> veryExpensiveConverter -> filter2 -> output2
       -> veryExpensiveConverter -> cheapConverter -> filter3 -> output3

我也想拥有

Input1 -> veryExpensiveConverter -> filter1 -> output1  
                                 -> filter2 -> output2
                                 -> cheapConverter -> filter3 -> output3

什么模式最适合避免为每个订阅者执行相同的映射？

Answer 1

您可以.share()在某种程度上确保对该共享部分的每个订阅只触发其上方的单个订阅。

您还可以查看 .publish().xxx() 方法以获得更高级的自动触发（.share() 将在第一个订阅到来时立即启动其来源）。

像这样：

val expensiveDoneOnce = publisher
    .map{veryExpensiveConverter.convert(it)}
    .publish()
    .refCount(2)
val sub1 = expensiveDoneOnce.filter(it.metric < 10)
val sub2 = expensiveDoneOnce.filter(it.metric > 5)

Spring Reactor如何避免对相似订阅者重复映射操作？

How to avoid repeating maping operations for similar subscribers in Sping Reactor?

spring

reactive-programming

kotlin

project-reactor