rxjava - 不按顺序发出的项目答案

【问题标题】：rxjava - items not emitted by sequentiallyrxjava - 不按顺序发出的项目
【发布时间】：2020-09-28 08:10:32
【问题描述】：

在下面的代码行中

@Test
fun rx() {
    val items = Observable.just(1, 2, 3, 4, 5)
            .observeOn(Schedulers.io()) //<---- if remove this line each item is emitted by sequentially (i.e 12345)

    items
            .filter { it == 1 }
            .doOnNext { print("1") }
            .subscribe()

    items
            .filter { it == 2 }
            .doOnNext { print("2") }
            .subscribe()

    items
            .filter { it == 3 }
            .doOnNext { print("3") }
            .subscribe()

    items
            .filter { it == 4 }
            .doOnNext { print("4") }
            .subscribe()

    items
            .filter { it == 5 }
            .doOnNext { print("5") }
            .subscribe()

    Thread.sleep(1000)
}

我希望日志应该说“12345”，但它说的是 23415、31245、53124 等。（即顺序看起来像随机的）

每一项 1、2、3、4、5 似乎不是按顺序发出的。

这有什么原因吗？以及如何使用“.observeOn(Schedulers.io())”修复它？

【问题讨论】：

不确定您的代码的目标是什么。如果您正在创建测试，请考虑到您有用于测试目的的特定调度程序，例如 TestScheduler，请在此处查看更多信息baeldung.com/rxjava-testing

标签： android rx-java rx-java2 rx-java3

【解决方案1】：

您正在做的是创建 5 个不同的订阅者，但不能保证按顺序执行。如果要保持执行顺序，则应使用Observable.concat 并将可观察对象列表合并为一个流

    val firstObservable = items
            .filter { it == 1 }
            .doOnNext { print("1") }
    val secondObservable = items
            .filter { it == 2 }
            .doOnNext { print("2") }

    val listOfObservables = arrayListOf(firstObservable, secondObservable...)
    
    Observable
        .concat(listOfObservables)
        .subscribe()
    Thread.sleep(1000);

编辑：

另外请查看关于 TestScheduler 的第一条评论，因为如果你想让你的测试更可预测，你应该阅读更多关于 TestScheduler 和 RxJava 测试的内容。

【讨论】：