如果第二个立即失败，ReactiveX concat 不会从第一个 observable 产生 onNext

Question

我连接两个可观察对象以首先显示缓存中的数据，然后开始从网络加载数据并显示更新的数据。

Observable.concat(
    getContentFromCache.subscribeOn(dbScheduler),
    getContentFromNetwork.subscibeOn(networkScheduler)
).observeOn(AndroidSchedulers.mainThread())
 .subscribe(subscriber);

如果没有网络连接，第二个 observable 在调用 OnSubscribe 后立即失败。

如果第二个 observable 立即失败，来自第一个 observable 的数据将丢失。 onNext 方法永远不会在订阅者中调用。

我认为，这可能是由于 OperatorConcat.ConcatSubscriber 中的以下代码

    @Override
    public void onNext(Observable<? extends T> t) {
        queue.add(nl.next(t));
        if (WIP_UPDATER.getAndIncrement(this) == 0) {
            subscribeNext();
        }
    }

    @Override
    public void onError(Throwable e) {
        child.onError(e);
        unsubscribe();
    }

看起来收到错误后它取消订阅，并且所有待处理的 onNext 都丢失了。

解决我的问题的最佳方法是什么？

更新

看起来我已经找到了解决方案，我没有为串联的可观察对象设置 observOn，而是为每个可观察对象设置了 observOn。

Observable.concat(
    getContentFromCache.subscribeOn(dbScheduler).observeOn(AndroidSchedulers.mainThread()),
    getContentFromNetwork.subscibeOn(networkScheduler).observeOn(AndroidSchedulers.mainThread())
)
 .subscribe(subscriber);

Answer 1

observeOn 的默认行为是onError 事件可以跳到队列前面，这是来自文档的引用：

请注意，onError 通知将提前于onNext 通知 如果Scheduler 是真正异步的，则在发射线程上。

这里有一个小测试来说明这件事：

Scheduler newThreadScheduler = Schedulers.newThread();

Observable<Integer> stream = Observable.create(integerEmitter -> {
    integerEmitter.onNext(1);
    integerEmitter.onNext(2);
    integerEmitter.onNext(3);
    integerEmitter.onNext(4);
    integerEmitter.onNext(5);
    integerEmitter.onError(new RuntimeException());
}, Emitter.BackpressureMode.NONE);

TestSubscriber<Integer> subscriber = new TestSubscriber<>();
stream.subscribeOn(Schedulers.computation())
        .observeOn(newThreadScheduler).subscribe(subscriber);

subscriber.awaitTerminalEvent();

subscriber.assertValues(1, 2, 3, 4, 5);
subscriber.assertError(RuntimeException.class);

通常消费者会期望以下序列：1 > 2 > 3 > 4 > 5 > Error。但是只使用observeOn 可能会提前出错并且测试会失败。

这个行为是很久以前在https://github.com/ReactiveX/RxJava/issues/1680 实现的，检查这样做的动机。为了避免这种行为，可以使用重载的observeOn 和delayError 参数：

表示onError 通知是否不能提前onNext 调度边界另一侧的通知。如果true 以onError 结尾的序列将按照收到的顺序重播 来自上游

这是您通常所期望的，因此将 observeOn(newThreadScheduler) 更改为 observeOn(newThreadScheduler, true) 将修复测试。

然后是@Neil 的问题：为什么@Rostyslav 提出的解决方案有效？它正在工作，因为最终序列没有线程切换。

在建议的解决方案中，最终序列是由同一线程上的两个序列构成的：第一个序列是来自缓存的数据，第二个序列只是来自网络的错误。它们是在同一个线程上一起制作的，并且在没有线程切换之后 - 订阅者在 AndroidSchedulers.mainThread() 上观察到。如果您尝试将最终的Scheduler 更改为其他值，它将再次失败。

Answer 2

RxJava 中的操作符通常被设计为短路 onError 通知。因为被连接的 observables 是异步源，所以你正在经历短路。如果您不想要短路，那么您可以对具体化的 observables 进行 concat，然后执行您想要的处理：

Observable.concat(
    getContentFromCache.materialize().subscribeOn(dbScheduler),
    getContentFromNetwork.materialize().subscribeOn(networkScheduler)
)

另一种方法是使用onErrorResumeNext:

Observable.concat(
    getContentFromCache.subscribeOn(dbScheduler),
    getContentFromNetwork.onErrorResumeNext(something)
        .subscibeOn(networkScheduler)
)

Answer 3

This is the small Example for concat both Observable

// First Observable //   
 private Observable<Book> getAutobiography()
        {
            String [] booknames= new String[]{"book1","book2"};
            final ArrayList<Book> arrayList = new ArrayList<>();
            for(String name:booknames){
                Book book = new Book();
                book.setBooktype("Autobiograph");
                book.setBookname(name);
                arrayList.add(book);
            }

            return Observable.create(new ObservableOnSubscribe<Book>() {
                @Override
                public void subscribe(ObservableEmitter<Book> emitter) throws Exception {

                    for(Book book:arrayList)
                    {
                        emitter.onNext(book);

                    }
                    if (!emitter.isDisposed()) {
                        emitter.onComplete();
                    }
                }
            }).subscribeOn(Schedulers.io());
        }

// Second Observable 



    private Observable<Book> getProgrammingBooks()
        {
            String [] booknames= new String[]{"book3","book4"};
            final ArrayList<Book> arrayList = new ArrayList<>();
            for(String name:booknames){
                Book book = new Book();
                book.setBooktype("Programming");
                book.setBookname(name);
                arrayList.add(book);
            }

            return Observable.create(new ObservableOnSubscribe<Book>() {
                @Override
                public void subscribe(ObservableEmitter<Book> emitter) throws Exception {

                    for(Book book:arrayList)
                    {
                        emitter.onNext(book);

                    }
                    if (!emitter.isDisposed()) {
                        emitter.onComplete();
                    }
                }
            }).subscribeOn(Schedulers.io());
        }

// Concat them 

    Observable.concat(getProgrammingBooks(),getAutobiography()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Book>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(Book book) {


            Log.d("bookname",""+book.getBookname());
            Log.d("booktype",""+book.getBooktype());
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    });
// it print both values //