unit-testing - rxjava testScheduler竞争条件

我偶然发现了一个奇怪的testScheduler行为，无法解决问题。下面的代码已大大简化，但起源于现实生活中的问题。
考虑以下测试:

@Test
fun testSchedulerFun(){

    val testScheduler = TestScheduler()

    val stringsProcessor = PublishProcessor.create<String>()

    val completable = Completable.complete()

    completable
        .doOnComplete { stringsProcessor.onNext("onComplete") }
        .subscribeOn(testScheduler)
        .subscribe()

    val testSubscriber = stringsProcessor
        .subscribeOn(testScheduler) //this line of code messes the test
        .test()

    testScheduler.triggerActions()

    testSubscriber
        .assertValues("onComplete")

}

**当我在stringsProcessor上订阅了经过测试的testScheduler时，测试将失败。当我删除该行时，它会成功。 **
我看到的事件流是:

triggerActions

completable和stringsProcessor被预订并向下游传播其事件。

很显然，在testSubscriber完成之后，对stringsProcessor.onNext("onComplete")进行了评估。

我想知道为什么

最佳答案

测试失败的原因是，您在其上调用stringProcessor的时间onNext没有订阅者。该订阅者仅次于此，因为您添加了“此行混乱” subscribeOn。
不涉及竞争条件，因为一切都以确定的顺序在同一线程上运行:

当代码执行completable ... subscribe()部分时，一个任务与testScheduler排队，将执行doOnComplete调用。

当代码执行test部分时，另一个任务与testScheduler一起排队，它将观察处理器。

triggerActions执行任务1，该任务不向任何订阅者发出该值，然后执行任务2，现在准备观察处理器，但是什么也没发生。