0

在下面的代码行中

@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())" 修复它？

4

1 回答 1

0

您正在做的是创建 5 个不同的订阅者，但不能保证按顺序执行。如果你想保持执行顺序，你应该使用Observable.concatobservables 列表并将其组合到一个流中

    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 测试的内容。

于 2020-09-28T08:31:28.013 回答