rxjava-不按顺序发出的项目

问题描述

在下面的代码行中

@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（））”修复该问题？

解决方法

您正在做的是，您要创建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中的测试的更多信息。