java - Rx Java mergeDelayError 未按预期工作

Question

我在 RxAndroid 和 Android 应用程序中使用 RxJava。我正在使用 mergeDelayError将两个 Retro fit 网络调用组合成一个 observable，如果其中一个发出一个，它将处理发出的项目，如果有一个则处理错误。这不起作用，它只会在遇到错误时触发 onError 操作。现在为了测试这一点，我转向了一个非常简单的示例，当我调用 onError 时，仍然不会调用successAction。请参见下面的示例。

Observable.mergeDelayError(
                Observable.error(new RuntimeException()),
                Observable.just("Hello")
            )
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .finallyDo(completeAction)
            .subscribe(successAction, errorAction);

只有当我使用两个成功的 observable 时才会调用成功操作。我是否遗漏了一些关于 mergeDelayError 应该如何工作的东西？

编辑:

我发现如果我删除 observeOn 和 subscribeOn 一切都会按预期工作。我需要指定线程并认为这就是使用 Rx 的全部意义所在。知道为什么指定这些 Scheduler 会破坏行为吗？

Answer 1

使用 .observeOn(AndroidSchedulers.mainThread(), true) 代替 .observeOn(AndroidSchedulers.mainThread()

public final Observable<T> observeOn(Scheduler scheduler, boolean delayError) {
        return observeOn(scheduler, delayError, RxRingBuffer.SIZE);
    }

上面是observeOn函数的签名。以下代码有效。

  Observable.mergeDelayError(
                Observable.error(new RuntimeException()),
                Observable.just("Hello")
        )
                .observeOn(AndroidSchedulers.mainThread(), true)
                .subscribeOn(Schedulers.io())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(String s) {

                    }
                });

从 ConcatDelayError 线程中得到这个技巧:https://github.com/ReactiveX/RxJava/issues/3908#issuecomment-217999009