使用 SingleScheduler 从多个线程调用时，RxJava2 PublishSubject 订阅者无法接收项目

Question

我有以下单元测试，我尝试从不同的线程发送 10 个Strings，并测试我从单个线程接收到这些Strings。我的问题是这个测试失败了。有时它会成功，但有时我只收到8 或9 项目，然后测试挂起直到闩锁超时。我是否以错误的方式使用SingleScheduler？我错过了什么吗？

val consumerCallerThreadNames = mutableSetOf<String>()
val messageCount = AtomicInteger(0)

val latch = CountDownLatch(MESSAGE_COUNT)

@Test
fun someTest() {
    val msg = "foo"

    val subject = PublishSubject.create<String>()
    subject
            .observeOn(SingleScheduler())
            .subscribe({ message ->
                consumerCallerThreadNames.add(Thread.currentThread().name)
                messageCount.incrementAndGet()
                latch.countDown()
            }, Throwable::printStackTrace)

    1.rangeTo(MESSAGE_COUNT).forEach {
        Thread({
            try {
                subject.onNext(msg)
            } catch (t: Throwable) {
                t.printStackTrace()
            }
        }).start()
    }
    latch.await(10, SECONDS)

    assertThat(consumerCallerThreadNames).hasSize(1)
    assertThat(messageCount.get()).isEqualTo(MESSAGE_COUNT)
}

companion object {
    val MESSAGE_COUNT = 10
}

如果我将其重写为使用单线程 ExecutorService，则测试不再出现问题，因此问题出在 Rx 或我对 Rx 缺乏了解。

Answer 1

RxJava 要求不能同时调用on*。这意味着您的代码不是线程安全的。

由于只有主题本身以并发方式使用，它应该可以通过使用 Subject<T>.toSerialized() 方法序列化（本质上是 Java 的“同步”）主题本身来修复。

val subject = PublishSubject.create<String>() 变为 val subject = PublishSubject.create<String>().toSerialized()。