美文网首页
RxJava2.0 操作符(5)—— Error Handle

RxJava2.0 操作符(5)—— Error Handle

作者: DoubleThunder | 来源:发表于2017-10-16 23:08 被阅读281次

很多操作符可用于对 Observable 发射的 onError 通知做出响应或者从错误中恢复,例如,你可以:

  1. 捕捉这个错误,切换到一个备用的 Observable 继续发射数据。
  2. 捕捉这个错误然后发射默认值。
  3. 捕捉这个错误并立即尝试重启这个 Observable。
  4. 捕捉这个错误,在一些回退间隔后重启这个 Observable。

Catch — 从 onError 通知中恢复发射数据。
Retry — 如果原始 Observable 遇到错误,重新订阅它期望它能正常终止。

5.1 Catch

从 onError 通知中恢复发射数据。

CatchCatch

Catch 操作符拦截原始 Observable 的 onError 通知,将它替换为其它的数据项或数据序列,让产生的 Observable 能够正常终止或者根本不终止。
RxJava 将 Catch 实现为三个不同的操作符:onErrorReturn ,onErrorResumeNext,onExceptionResumeNext。

5.1.1 onErrorReturn

让 Observable 遇到错误时发射一个特殊的项并且正常终止。

onErrorReturnonErrorReturn

示例代码:

Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
            for (int i = 0; i <= 5; i++) {
                emitter.onNext("" + i);
                if (i == 3) {
                    emitter.onError(new Throwable("nothing"));
                }
            }
            emitter.onComplete();
        }
    }).onErrorReturn(new Function<Throwable, String>() {
        @Override
        public String apply(@NonNull Throwable throwable) throws Exception {
            return "Error Return";
        }
    }).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {

        }

        @Override
        public void onNext(@NonNull String s) {
            Log.e(TAG, "onNext:" + s);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            Log.e(TAG, "onError" + e.getMessage());
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });

输出结果:

onNext:0
onNext:1
onNext:2
onNext:3
onNext:Error Return
onComplete

5.1.2 onErrorResumeNext

让 Observable 在遇到错误时,由另外一个 Observable 代替当前的 Observable 并继续发射数据。

onErrorResumeNextonErrorResumeNext

它有两种实现

  1. OnErrorResumeNext(Observable);出现错误时取到当前 Observable 并做处理。
  2. OnErrorResumeNext(Function(Throwable,Observable));出现错误时发射另一个 Observable。
    示例代码:
 Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
            for (int i = 0; i <= 5; i++) {
                emitter.onNext("" + i);
                if (i == 3) {
                    emitter.onError(new Throwable("nothing"));
                }
            }
            emitter.onComplete();
        }
    }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends String>>() {
        @Override
        public ObservableSource<? extends String> apply(@NonNull Throwable t) throws Exception {
            return Observable.just("hello","world");
        }
    }).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {

        }

        @Override
        public void onNext(@NonNull String s) {
            Log.e(TAG, "onNext:" + s);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            Log.e(TAG, "onError" + e.getMessage());
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });

输出结果:

onNext:0
onNext:1
onNext:2
onNext:3
onNext:hello
onNext:world
onComplete

5.1.3 onExceptionResumeNext

onExceptionResumeNext 类似于 OnErrorResume ,不同之处在于其会对 onError 抛出的数据类型做判断,如果是 Exception,也会使用另外一个 Observable 代替原 Observable 继续发射数据;否则会将错误分发给 Subscriber。

onExceptionResumeNextonExceptionResumeNext

示例代码:

 Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
            for (int i = 0; i <= 5; i++) {
                emitter.onNext("" + i);
                if (i == 3) {
                    emitter.onError(new Throwable("error throable"));//发送一个throwable
//                  emitter.onError(new Exception("error exception"));//发送一个exception
                }
            }
            emitter.onComplete();
        }
    }).onExceptionResumeNext(Observable.just("hello","world")).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {

        }

        @Override
        public void onNext(@NonNull String s) {
            Log.e(TAG, "onNext:" + s);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            Log.e(TAG, "onError:" + e.getMessage());
        }

        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });

当发送 throwable 时,错误分发给 Subscriber。
输出结果 1:

onNext:0
onNext:1
onNext:2
onNext:3
onError:error throable

当发送 exception 时,使用新的 Observable。
输出结果 2:

onNext:0
onNext:1
onNext:2
onNext:3
onNext:hello
onNext:world
onComplete

5.2 Retry

如果原始 Observable 遇到错误,重新订阅它期望它能正常终止。

RetryRetry

Retry 操作符不会将原始 Observable 的 onError 通知传递给观察者,它会订阅这个 Observable,再给它一次机会无错误地完成它的数据序列。Retry 总是传递 onNext 通知给观察者,由于重新订阅,可能会造成数据项重复,如上图所示。

RxJava 中的实现为 Retry 和 RetryWhen。

5.2.1 Retry

无论收到多少次 onError 通知,无参数版本的 retry 都会继续订阅并发射原始 Observable。

Retry 的三种实现方法:

  1. retry(long) :会最多重新订阅指定的次数,如果次数超了,它不会尝试再次订阅,它会把最新的一个 onError 通知传递给它的观察者。
  2. retry(): 等价于 retry(Long.MAX_VALUE) ;
  3. retry(BiPredicate) :这个函数返回一个 boolean,如果返回 true,retry 再次订阅原 Observable;如果返回 false,retry 会将最新的一个 onError 通知传递给它的观察者。

示例代码 1:

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
        for (int i = 3; i <= 5; i++) {
            emitter.onNext("" + i);
            if (i == 3) {
//                        emitter.onError(new Throwable("error throable"));
                emitter.onError(new Exception("error exception"));
            }
        }
        emitter.onComplete();
    }
}).retry(2)
        .subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {

    }

    @Override
    public void onNext(@NonNull String s) {
        Log.e(TAG, "onNext:" + s);
    }

    @Override
    public void onError(@NonNull Throwable e) {
        Log.e(TAG, "onError:" + e.getMessage());
    }

    @Override
    public void onComplete() {
        Log.e(TAG, "onComplete");
    }
});

输出结果:

onNext:3
onNext:3
onNext:3
onError:error exception

示例代码 2:

//将前一段代码 retry(2) 替换成。两者等价
retry(new BiPredicate<Integer, Throwable>() {
    @Override
    public boolean test(@NonNull Integer integer, @NonNull Throwable throwable) throws Exception {
//          Log.e(TAG, integer + "");
        if (integer >= 2) {
            return false;
        }
        return true;
    }
})

输出结果:

onNext:3
onNext:3
onNext:3
onError:error exception

5.2.1 RetryWhen

RetryWhenRetryWhen

retryWhen 和 retry 类似,区别是,retryWhen 将 onError 中的 Throwable 传递给一个函数,这个函数产生另一个 Observable,retryWhen 观察它的结果再决定是不是要重新订阅原始的 Observable。如果这个 Observable 发射了一项数据,它就重新订阅,如果这个 Observable 发射的是 onError 通知,它就将这个通知传递给观察者然后终止。

需要注意的是使用 retryWhen 的时候,因为每次重新订阅都会产生错误,所以作为参数的 obserbvable 会不断地发射数据,使用 zipWith 操作符可以限制重新订阅的次数,否则会无限制地重新订阅。

示例代码 2:

Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
            emitter.onNext("3");
            emitter.onError(new Exception("error exception"));
            emitter.onComplete();
        }
    }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(@NonNull Observable<Throwable> to) throws Exception {
    
            return to.zipWith(Observable.range(1, 4)
            , new BiFunction<Throwable, Integer, String>() {
                @Override
                public String apply(@NonNull Throwable t, @NonNull Integer i) throws Exception {
                    Log.e(TAG, "retryWhen count :" + integer);
                    return "integer - "+integer;
                }
            });
        }
    }).subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {
    
        }
    
        @Override
        public void onNext(@NonNull String s) {
            Log.e(TAG, "onNext:" + s);
        }
    
        @Override
        public void onError(@NonNull Throwable e) {
            Log.e(TAG, "onError:" + e.getMessage());
        }
    
        @Override
        public void onComplete() {
            Log.e(TAG, "onComplete");
        }
    });

输出结果:

onNext:3
retryWhen count :1
onNext:3
retryWhen count :2
onNext:3
retryWhen count :3
onNext:3
retryWhen count :4
onComplete

retry 会将错误分发给观察者,而 retryWhen 会正常结束,并不会讲错误分发出去。

相关文章

网友评论

      本文标题:RxJava2.0 操作符(5)—— Error Handle

      本文链接:https://www.haomeiwen.com/subject/bafruxtx.html