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

时间 2019-11-21

标签 rxjava2 rxjava 实战知识梳理如何使得错误发生自动中止订阅关系栏目 Java 繁體版

原文原文链接

1、发生错误时中止订阅的状况

在RxJava中，若是发生了错误，那么 订阅者会自动中止对上游的订阅关系 ，咱们将致使订阅取消的错误分为两种：java

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

在RxJava的设计中，若是发生了错误，那么订阅关系就取消了。可是在某些时候，咱们但愿在错误发生的时候不要取消订阅，由于这样订阅者只有从新经过subscribe方法才能收到消息，相似的场景如监测数据源变化、RxBus的实现等。git

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

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事件，以后它就再也没法收到消息了。 app

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

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

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 {

            }
        });
    }
复制代码

咱们在订阅者收到第四个数据的时候抛出一个异常，此时控制台的输出为以下，与上面相似，以后订阅者都没法接收到消息，由于订阅关系已经被解除了。 post

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

2.1 上游发生错误

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

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

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的回调，在其中咱们分为注释中的两部分进行处理，第四次的时候发起重订阅，而第八次则不发起，所以，第九个事件订阅者就收不到了，控制台的输出为： code

2.2 订阅者发生错误

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

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

解决办法就是， 把 LambdaObserver 代码拷贝出来，注释掉那句，而后继承于它去实现 Observer，代码在 RxSample 的十三章例子中。

从控制台能够看出，并无解除订阅关系，在发生错误以后，仍然能够继续收到数据。