RxJava2 PublishSubject 的意外行为

Question

我有以下代码使用PublishSubject。

val subject = PublishSubject.create<Int>()

val o1: Observable<String> =
        subject.observeOn(Schedulers.newThread()).map { i: Int ->
            println("${Thread.currentThread()} | ${Date()} | map => $i")
            i.toString()
        }

o1.subscribe {
    println("${Thread.currentThread()} | ${Date()} | direct subscription (1) => $it")
}

o1.subscribe {
    println("${Thread.currentThread()} | ${Date()} | direct subscription (2) => $it")
}

o1.subscribe {
    println("${Thread.currentThread()} | ${Date()} | direct subscription (3) => $it")
}

println("${Thread.currentThread()} | ${Date()} | submitting 1")

subject.onNext(1)

1) 我从它创建一个Observable 并对其进行映射（出于本示例的目的，我只是转换为String）=> o1。

2) 然后我订阅o1 3 次。

3) 最后我通过调用subject.onNext(1)“发布”一个事件。

令我惊讶的是，我得到了以下输出：

Thread[main,5,main] | Mon Jun 19 09:46:37 PDT 2017 | submitting 1
Thread[RxNewThreadScheduler-1,5,main] | Mon Jun 19 09:46:37 PDT 2017 | map => 1
Thread[RxNewThreadScheduler-2,5,main] | Mon Jun 19 09:46:37 PDT 2017 | map => 1
Thread[RxNewThreadScheduler-3,5,main] | Mon Jun 19 09:46:37 PDT 2017 | map => 1
Thread[RxNewThreadScheduler-1,5,main] | Mon Jun 19 09:46:37 PDT 2017 | direct subscription (1) => 1
Thread[RxNewThreadScheduler-2,5,main] | Mon Jun 19 09:46:37 PDT 2017 | direct subscription (2) => 1
Thread[RxNewThreadScheduler-3,5,main] | Mon Jun 19 09:46:37 PDT 2017 | direct subscription (3) => 1

map 最终被调用了 3 次，我不明白为什么，因为我订阅了 o1，这应该发生在 map 发生之后。我肯定错过了什么。任何帮助将不胜感激。

谢谢 颜

Answer 1

来自 cmets：

您订阅o1 三次，每次都创建一个独立的序列，直到PublishSubject 将onNext 发送到所有3 个链。

从所有 3 个订阅者的角度来看，PublishSubject 是一个多播源，通过subscribe() 调用建立的独立链向他们发送事件信号。

在Subject 上应用运算符通常不会使整个链变热，因为这些运算符元素仅在订阅时才附加到源Subject。因此，多个订阅将产生多个通道到同一上游Subject。

使用publish 得到一个ConnectableObservable（或在最后的另一个PublishSubject）以使序列从那时开始变得热门。