1

这是我在 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

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

4

1 回答 1

3

因为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}")
        })
于 2021-07-02T08:42:52.710 回答