java - 为什么 RxJava Observable.doOnDispose 会触发两次？

Question

这段代码

@RunWith(JUnit4::class)
class Playground {

    @Test
    fun test() {
        val subject1 = BehaviorSubject.create<Int>()
        val subject2 = BehaviorSubject.create<Int>()
        val observable = subject1.doOnDispose { println("subject1 observable disposed") }
        val disposable = Observable.combineLatest(
            observable.takeUntil { it < 1 },
            subject2.doOnDispose { println("subject2 observable disposed") },
            BiFunction { t1: Int, t2: Int ->
                println("$t1 $t2")
            })
            .subscribe()
        subject1.onNext(1)
        subject2.onNext(0)
        subject1.onNext(0)
        Thread.sleep(100)
        disposable.dispose()
    }
}

有这样的输出

1 0
0 0
subject1 observable disposed
subject1 observable disposed
subject2 observable disposed

这是错误的想法，因为奇怪的是 observable 可以被处理两次。谁能解释一下为什么会这样？

implementation("io.reactivex.rxjava2:rxjava:2.2.2")

Answer 1

因为它适用于 OP，所以让它为 future 的读者提供答案不会有什么坏处，我想。基本上升级 RxJava 版本到 2.2.6 解决了这个问题:

implementation 'io.reactivex.rxjava2:rxjava:2.2.6'

任何对此行为背后的问题感兴趣的人都可以在此处查看 RxJava 2.2.2 中问题的 PR https://github.com/ReactiveX/RxJava/pull/6269