美文网首页
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

      延伸阅读

      深度阅读

      • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

      栏目导航

      热点阅读

      关于我们|服务条款|联系我们|RxJava2.0 操作符（5）—— Error Handle |投稿指南|网站地图|RSS订阅|排版工具|手机版

      提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

      Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

      备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

      本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

      所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！