RxJava 合并保留顺序没有意义答案

【问题标题】：RxJava merge preserve order without senseRxJava 合并保留顺序没有意义
【发布时间】：2021-10-17 19:47:59
【问题描述】：

我读到了合并和连接之间的区别。 Concat 保留顺序，不合并。

但在我的情况下，合并总是保持顺序。有什么问题？为什么我的合并等待以前的任务？我想获得随机结果，但在这种情况下，我总是得到 A1 A2 A3... B1 B2 B3...

fun doSomething() {
        compositeDisposable.add(
            Observable.merge(getNumber1(), getNumber2())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                    { item -> println(item) },
                    { error -> println(error.message) }
                ))
    }

    private fun getNumber1(): Observable<String> {
        return Observable.create { emitter ->
            for (i in 1..10) {
                emitter.onNext("ObservableA: " + i)
            }
            emitter.onComplete()
        }
    }

    private fun getNumber2(): Observable<String> {
        return Observable.create { emitter ->
            for (i in 1..10) {
                emitter.onNext("ObservableB: " + i)
            }
            emitter.onComplete()
        }
    }

【问题讨论】：

标签： android kotlin observable rx-java

【解决方案1】：

为什么我的合并等待之前的任务？

这是因为它们都在同一个线程上执行。要让这两个任务并行工作，subscribeOn 应该在每个 Observable 被合并之前调用：

Observable.merge(
    getNumber1().subscribeOn(Schedulers.io()),
    getNumber2().subscribeOn(Schedulers.io())
).observeOn(AndroidSchedulers.mainThread())
    ...

您也可以在emitter.onNext 之后添加Thread.sleep(100) 以进行更好的测试

【讨论】：

【解决方案2】：

要了解合并和合并之间的区别，您可以参考以下示例，其中每 10 毫秒发生一次发射。

val observableFirst = Observable.interval(10, TimeUnit.MILLISECONDS).takeWhile { it < 10 }.map { "A$it" }
val observableSecond = Observable.interval(10, TimeUnit.MILLISECONDS).takeWhile { it < 10 }.map { "B$it" }

val result = arrayListOf<String>()
Observable.merge(observableFirst, observableSecond)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnComplete { println(result) }
    .subscribe(
        { item -> result.add(item) },
        { error -> println(error.message) }
    )

输出（每次可能不同）：[A0, B0, A1, B1, A2, B2, A3, B3, A4, B4, A5, B5, A6, B6, B7, A7, B8, A8, B9, A9]

在您的示例中，for 循环会导致非常快的发射间隔。因此，您没有发现关于其顺序的任何差异。

在我的示例中，如您所见，发射顺序没有得到维护。此外，您可以使用 concat 而不是 merge 来检查这一点。

输出（无差异）：[A0, A1, A2, A3, A4, A5, A6, A7, A8, A9, B0, B1, B2, B3, B4, B5, B6, B7, B8, B9]

【讨论】：