RxJava2 实战知识梳理(13) - 如何使得错误发生时不自动停止订阅关系

RxJava2 实战系列文章

一、发生错误时停止订阅的情况

在RxJava中，如果发生了错误，那么 订阅者会自动停止对上游的订阅关系 ，我们将导致订阅取消的错误分为两种：

上游：上游发生错误，并发送onError事件给订阅者。
下游：订阅者在onNext中处理时发生了异常。

在RxJava的设计中，如果发生了错误，那么订阅关系就取消了。但是在某些时候，我们希望在错误发生的时候不要取消订阅，因为这样订阅者只有重新通过subscribe方法才能收到消息，类似的场景如监测数据源变化、RxBus的实现等。

我们先用两个简单的例子来演示一下上面提到的两种情况：

1.1 上游传递消息时发生错误

订阅者在初始时候订阅到mPublishObject，当该PublishObject发送到第四个事件时，主动抛出一个异常，以模拟上游发生异常的情况。

     private void upError() {
        mPublishSubject.map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                if (integer == 4) {
                    throw new RuntimeException();
                }
                return integer;
            }
        }).observeOn(AndroidSchedulers.mainThread()).subscribe(getNormalObserver());
    }

    private Observer<Integer> getNormalObserver() {
        return new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "onNext=" + value);
            }
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError=" + e);
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        };
    }

从控制台的输出可以看到，当第四次发送事件后，由于上游发生了异常，因此订阅者收到了onError事件，之后它就再也无法收到消息了。

1.2 订阅者处理消息时发生错误

下面，我们再来看订阅处理消息时发生错误的场景：

    private void downError() {
        mPublishSubject.observeOn(AndroidSchedulers.mainThread()).subscribe(getErrorObserver());
    }

    private LambdaObserver<Integer> getErrorObserver() {
        return new LambdaObserver<>(new Consumer<Integer>() {
            @Override
            public void accept(Integer value) throws Exception {
                Log.d(TAG, "onNext=" + value);
                if (value == 4) {
                    throw new RuntimeException();
                }
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "onError=" + throwable);
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.d(TAG, "onComplete");
            }
        }, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {

            }
        });
    }

我们在订阅者收到第四个数据的时候抛出一个异常，此时控制台的输出为如下，与上面类似，之后订阅者都无法接收到消息，因为订阅关系已经被解除了。

二、发生异常时的处理办法

2.1 上游发生错误

在上游发生错误的时候，一般通过重订阅的方式来解决。我们可以根据错误的类型判断是否需要重订阅，重订阅的时候使用retryWhen操作符，这个我们在 RxJava2 实战知识梳理(6) - 基于错误类型的重试请求已经介绍过了。

下面，我们演示一下在上面的错误当中如何恢复：

    private void upErrorIgnore() {
        mPublishSubject.map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                if (integer == 4) {
                    throw new RuntimeException("retry");
                } else if (integer == 8) {
                    throw new RuntimeException("don't retry");
                }
                return integer;
            }
        }).observeOn(AndroidSchedulers.mainThread()).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                //第一步，通过flatMap对错误进行响应。
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Exception {
                        //第二步：根据错误的类型判断是否需要重订阅。
                        return "retry".equals(throwable.getMessage()) ? Observable.just(0) : Observable.empty();
                    }
                });
            }
        }).subscribe(getNormalObserver());
    }

在第四次/第八次点击的是否，我们分别在上游抛出一个异常，这样就会触发retryWhen的回调，在其中我们分为注释中的两部分进行处理，第四次的时候发起重订阅，而第八次则不发起，因此，第九个事件订阅者就收不到了，控制台的输出为：

2.2 订阅者发生错误

但是retryWhen只能处理上游发生错误的情况，对于上面说的第二种情况并不能处理，因此假如是上面介绍的第二种情况：订阅者在onNext处理中发生错误的情况，仍然会解除订阅关系。

这里首先要感谢 Johnny Shieh 提供的解决方法，在 RxJava 2 版本的 Rxbus 一文中，他分析了这一问题的原因，这是因为在LambdaObserver的源码中，如果在onNext中发生了异常，那么首先会调用onError方法，而onError中会执行取消订阅的操作。