java - 可观察以重用运算符执行

Question

给定以下示例(kotlin 代码)

val subject = PublishSubject.create<Int>()

val stream = subject.map {
    println("mapping")
    it * 2
}

stream.forEach { println("A: $it") }
stream.forEach { println("B: $it") }

subject.onNext(1)
subject.onCompleted()

输出将是

mapping
A: 2
mapping
B: 2

我想要实现的是，source observable 被映射一次，所有订阅者都得到结果，但不为每个订阅者执行映射操作......

像这样

mapping
A: 2
B: 2

在我的例子中，我在延迟和性能至关重要的地方进行非常昂贵的计算，我有一个热可观察作为源和大量订阅者......

我们如何重用运算符执行？以及通常不同的映射操作？

Answer 1

您可以使用 cache缓存任何 future 订阅者可观察到的源结果:

val stream = subject.map {
    println("mapping")
    it * 2
}.cache()

如果你想更精细地控制缓存的方式 replay值得研究。

如果您不想缓存源 Observable 的每个项目，而只想重新发布 新项目，您可以使用 publish与 autoConnect :

val stream = subject.map {
    println("mapping")
    it * 2
}.publish()
 .autoConnect()

给定以下事件序列:

stream.forEach { println("A: $it") }
stream.forEach { println("B: $it") }

subject.onNext(1)

stream.forEach { println("C: $it") }
subject.onNext(2)
subject.onCompleted()

将打印:

mapping
A: 2
B: 2
mapping
A: 4
B: 4
C: 4