kotlin - PublishSubject `subscribeOn`行为

Question

为什么subscribe在这里从不打印任何内容？只是出于好奇。无论如何，这是一个坏习惯，我会改用observeOn，但是我无法弄清楚为什么从来没有达到subscribe ...

fun main() {
val subject: PublishSubject<Int> = PublishSubject.create()
val countDownLatch = CountDownLatch(1)

subject
    .map { it + 1 }
    .subscribeOn(Schedulers.computation())
    .subscribe {
        println(Thread.currentThread().name)
        countDownLatch.countDown()
    }

subject.onNext(1)
countDownLatch.await()

}

Answer 1

为什么会这样
在订阅过程中，观察者会通过Subscribe通知向观察者发出准备接收商品的信号。有关详细信息，请参见 Observable contract。
此外， Subject 文档指出:

Note that a PublishSubject may begin emitting items immediately upon creation (unless you have taken steps to prevent this), and so there is a risk that one or more items may be lost between the time the Subject is created and the observer subscribes to it.

当您尝试通过 subject.onNext(_)订阅新线程后立即调用 .subscribeOn(Schedulers.computation())时，可观察对象(即 subject)可能仍在等待观察者的 Subscribe通知。但是，如果在发射第一个项目之前添加了一些时间延迟，则可观察对象更有可能在调用 Subscribe之前从观察者接收到 subject.onNext(_)通知。例如:

subject
    .subscribeOn(Schedulers.computation())
    .subscribe {...}

Thread.sleep(1000)

subject.onNext(1)

// prints "main"

该怎么办？
如果希望所有订阅都接收可观察对象发出的所有项目，则可以执行以下操作之一:

调用subject.onNext(_)之前，阻止主线程等待所有观察者被订阅。

创建一个新的可观察对象，该对象要等到所有可观察对象都被订阅后，才能在其内部调用subject.onNext(_)。

这些也可能有用:

ReplaySubject:这允许您存储所有先前项目的历史记录，并在每次订阅时重新发送它们。缺点:您需要在内存中存储任意数量的项目。

ConnectableObservable :这确保可观察对象仅在调用.connect()后发射项目。特别地，.autoConnect(n)运算符确保可观察对象仅在n观察者成功订阅之后才发出。

示例:阻塞主线程直到订阅

val subject: PublishSubject<Int> = PublishSubject.create()
val countDownLatch = CountDownLatch(1)
val isSubscribedLatch = CountDownLatch(1)

subject
    .subscribeOn(Schedulers.computation())
    .doOnSubscribe { isSubscribedLatch.countDown() }
    .map { it + 1 }
    .subscribe {
        countDownLatch.countDown()
        println(Thread.currentThread().name)
    }

isSubscribedLatch.await()
subject.onNext(1)
countDownLatch.await()