为什么 subscribeOn 对 Rxjava 中的 PublishSubject 没有影响？

Question

这是我在Kotlin:中的测试代码

fun main() {
rxjava()
}
fun rxjava() {
val queuSubject = PublishSubject.create<String>()
queuSubject
.map { t ->
val a = t.toLong()
Thread.sleep(6000 / a)
println("map $a called ${Thread.currentThread().name} ")
a
}
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe({
println("thread in subscription ${Thread.currentThread().name}")
}, {
println("error ${it.message}")
})
for (i in 1..3) {
Thread {
queuSubject.onNext("$i")
}.start()
}
Thread.sleep(15000)
}

我正在尝试在不同的IO线程中运行map块和subscribe's onNext块。但输出是这样的：

map 3 called Thread-2 
thread in subscription RxCachedThreadScheduler-2
map 2 called Thread-1 
thread in subscription RxCachedThreadScheduler-2
map 1 called Thread-0 
thread in subscription RxCachedThreadScheduler-2

正如您所看到的，调用subscribeOn似乎对PublishSubject's流没有影响，而thread-0,thread-1 and thread-2指的是调用onNext方法的线程。

另外考虑以下代码：

fun main() {
rxjava()
}
fun rxjava() {
val queuSubject = PublishSubject.create<String>()
queuSubject
.map { t ->
val a = t.toLong()
Thread.sleep(6000 / a)
println("map $a called ${Thread.currentThread().name} ")
a
}
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe({
println("thread in subscription ${Thread.currentThread().name}")
}, {
println("error ${it.message}")
})
queuSubject.onNext("1")
queuSubject.onNext("2")
queuSubject.onNext("3")
Thread.sleep(15000)
}

我写了上面的代码，发现没有输出。但是，如果我从流中删除subscribeOn，消息将按如下顺序打印：

map 1 called main 
thread in subscription RxCachedThreadScheduler-1
map 2 called main 
thread in subscription RxCachedThreadScheduler-1
map 3 called main 
thread in subscription RxCachedThreadScheduler-1

这些代码有什么问题？谢谢

Answer 1

因为subscribeOn只影响源的订阅副作用。这样的副作用是，如果源在观察者订阅时立即开始发送事件：

Observable.just(1, 2, 3)
.subscribeOn(Schedulers.io())
.doOnNext(v -> System.out.println(Thread.currentThread() + " - " + v)
.blockingSubscribe();

PublishSubject没有订阅副作用，因为它只将信号从其onXXX方法中继到观察者的onXXX方法。

然而，subscribeOn具有时间效应，因为它延迟了对源的实际订阅，因此在PublishSubject的情况下，它可能无法及时看到注册的观察者，其他线程调用其onXXX方法。

如果要将处理从原始线程移开，请使用observeOn:

val queuSubject = PublishSubject.create<String>()
queuSubject
.observeOn(Schedulers.io()) // <----------------------------------------
.map { t ->
val a = t.toLong()
Thread.sleep(6000 / a)
println("map $a called ${Thread.currentThread().name} ")
a
}
.observeOn(Schedulers.io())
.subscribe({
println("thread in subscription ${Thread.currentThread().name}")
}, {
println("error ${it.message}")
})