RxJava操作符（五）Error Handling
2016-09-01 00:31
417 查看
我们知道Rxjava对错误的处理很方便，就是当有错误出现的时候就会调用Subscriber的onError方法将错误分发出去，由Subscriber自己来处理错误。这种处理的错误有其优点，也有其缺点，就是每个Subscriber都要来定义如何处理错误，如果由100个Subscriber就要定义100遍，如何来统一地处理这些错误呢？这时我们就可以使用Error
handling相关的操作符来集中统一地处理错误。
一、Catch
Catch类似于java 中的try/catch，当错误发生的时候，可以拦截对onError的调用，让Observable不会因为错误的产生而终止。在Rxjava中，将这个操作符实现为3个操作符，分别是：
1、OnErrorReturn-当发生错误的时候，让Observable发射一个预先定义好的数据并正常地终止
2、OnErrorResume-当发生错误的时候，由另外一个Observable来代替当前的Observable并继续发射数据
3、OnExceptionResumeNext-类似于OnErrorResume,不同之处在于其会对onError抛出的数据类型做判断，如果是Exception，也会使用另外一个Observable代替
原Observable继续发射数据，否则会将错误分发给Subscriber。
下面使用OnErrorReturn和OnErrorResumeNext来分别处理两个Observable。
点击(此处)折叠或打开
private Observable<String> onErrorReturnObserver() {
return createObserver().onErrorReturn(throwable -> "onErrorReturn");
}
private Observable<String> onErrorResumeNextObserver() {
return createObserver().onErrorResumeNext(Observable.just("7", "8", "9"));
}
private Observable<String> createObserver() {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
for (int i = 1; i <= 6; i++) {
if (i < 3) {
subscriber.onNext("onNext:" + i);
} else {
subscriber.onError(new Throwable("Throw
error"));
}
}
}
});
}
分别对其进行订阅
点击(此处)折叠或打开
mLButton.setText("onErrorReturn");
mLButton.setOnClickListener(e -> onErrorReturnObserver().subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onErrorReturn-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onErrorReturn-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onErrorReturn-onNext:" + s);
}
}));
mRButton.setText("onErrorResume");
mRButton.setOnClickListener(e -> onErrorResumeNextObserver().subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onErrorResume-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onErrorResume-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onErrorResume-onNext:" + s);
}
}));
}
运行结果如下。可以看到onErrorReturn在错误发生的时候继续发射了提前定义好的数据并正常结束Observable；
而onErrorResume在错误发生后继续发射另外的数据7/8/9，然后正常结束了Observable。
下面我们使用OnExceptionResumeNext操作符来处理两个Observable
点击(此处)折叠或打开
private Observable<String> onExceptionResumeObserver(boolean isException) {
return createObserver(isException).onExceptionResumeNext(Observable.just("7", "8", "9"));
}
private Observable<String> createObserver(Boolean createExcetion) {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
for (int i = 1; i <= 6; i++) {
if (i < 3) {
subscriber.onNext("onNext:" + i);
} else if (createExcetion) {
subscriber.onError(new Exception("Exception"));
} else {
subscriber.onError(new Throwable("Throw
error"));
}
}
}
});
}
分别对其进行订阅
点击(此处)折叠或打开
mLButton.setText("onException-true");
mLButton.setOnClickListener(e -> onExceptionResumeObserver(true).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onException-true-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onException-true-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onException-true-onNext:" + s);
}
}));
mRButton.setText("onException-false");
mRButton.setOnClickListener(e -> onExceptionResumeObserver(false).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onException-false-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onException-false-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onException-false-onNext:" + s);
}
}));
运行结果如下。可以看到当发生错误时抛出的时Exception，则继续发射另外一个Observable的数据并正常结束；而抛出Throwable类型数据时，错误就分发到了Subscriber。
二、Retry
Retry操作符在发生错误的时候会重新进行订阅,而且可以重复多次，所以发射的数据可能会产生重复。如果重复指定次数还有错误的话就会将错误返回给观察者
Rxjava还实现了RetryWhen操作符。当错误发生时，retryWhen会接收onError的throwable作为参数，并根据定义好的函数返回一个Observable，如果这个Observable发射一个数据，就会重新订阅。
下面我们使用retry和retryWhen来处理两个Observable产生的错误，需要注意的是使用retryWhen的时候,因为每次重新订阅都会产生错误，所以作为参数的obserbvable会不断地发射数据，使用zipWith操作符可以限制重新订阅的次数，否则会无限制地重新订阅。
点击(此处)折叠或打开
private Observable<Integer> retryObserver() {
return createObserver().retry(2);
}
private Observable<Integer> retryWhenObserver() {
return createObserver().retryWhen(observable -> observable.zipWith(Observable.just(1, 2, 3),
(throwable, integer) -> throwable.getMessage() + integer)
.flatMap(throwable -> {
log(throwable);
return Observable.timer(1, TimeUnit.SECONDS);
}));
}
private Observable<Integer> createObserver() {
return Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
log("subscribe");
for (int i = 0; i < 3; i++) {
if (i == 2) {
subscriber.onError(new Exception("Exception-"));
} else {
subscriber.onNext(i);
}
}
}
});
}
分别进行订阅
点击(此处)折叠或打开
mLButton.setText("retry");
mLButton.setOnClickListener(e -> retryObserver().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
log("retry-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("retry-onError:" + e.getMessage());
}
@Override
public void onNext(Integer s) {
log("retry-onNext:" + s);
}
}));
mRButton.setText("retryWhen");
mRButton.setOnClickListener(e -> retryWhenObserver().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
log("retryWhen-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("retryWhen-onError:" + e.getMessage());
}
@Override
public void onNext(Integer s) {
log("retryWhen-onNext:" + s);
}
}));
}
运行结果如下。在尝试了几次还是产生错误后，retry会将错误分发给观察者，而retryWhen会正常结束，并不会讲错误分发出去。
关于错误的处理就到这里了，本文中的程序见：https://github.com/Chaoba/RxJavaDemo
handling相关的操作符来集中统一地处理错误。
一、Catch
Catch类似于java 中的try/catch，当错误发生的时候，可以拦截对onError的调用，让Observable不会因为错误的产生而终止。在Rxjava中，将这个操作符实现为3个操作符，分别是：
1、OnErrorReturn-当发生错误的时候，让Observable发射一个预先定义好的数据并正常地终止
2、OnErrorResume-当发生错误的时候，由另外一个Observable来代替当前的Observable并继续发射数据
3、OnExceptionResumeNext-类似于OnErrorResume,不同之处在于其会对onError抛出的数据类型做判断，如果是Exception，也会使用另外一个Observable代替
原Observable继续发射数据，否则会将错误分发给Subscriber。
下面使用OnErrorReturn和OnErrorResumeNext来分别处理两个Observable。
点击(此处)折叠或打开
private Observable<String> onErrorReturnObserver() {
return createObserver().onErrorReturn(throwable -> "onErrorReturn");
}
private Observable<String> onErrorResumeNextObserver() {
return createObserver().onErrorResumeNext(Observable.just("7", "8", "9"));
}
private Observable<String> createObserver() {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
for (int i = 1; i <= 6; i++) {
if (i < 3) {
subscriber.onNext("onNext:" + i);
} else {
subscriber.onError(new Throwable("Throw
error"));
}
}
}
});
}
分别对其进行订阅
点击(此处)折叠或打开
mLButton.setText("onErrorReturn");
mLButton.setOnClickListener(e -> onErrorReturnObserver().subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onErrorReturn-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onErrorReturn-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onErrorReturn-onNext:" + s);
}
}));
mRButton.setText("onErrorResume");
mRButton.setOnClickListener(e -> onErrorResumeNextObserver().subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onErrorResume-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onErrorResume-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onErrorResume-onNext:" + s);
}
}));
}
运行结果如下。可以看到onErrorReturn在错误发生的时候继续发射了提前定义好的数据并正常结束Observable；
而onErrorResume在错误发生后继续发射另外的数据7/8/9，然后正常结束了Observable。
下面我们使用OnExceptionResumeNext操作符来处理两个Observable
点击(此处)折叠或打开
private Observable<String> onExceptionResumeObserver(boolean isException) {
return createObserver(isException).onExceptionResumeNext(Observable.just("7", "8", "9"));
}
private Observable<String> createObserver(Boolean createExcetion) {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
for (int i = 1; i <= 6; i++) {
if (i < 3) {
subscriber.onNext("onNext:" + i);
} else if (createExcetion) {
subscriber.onError(new Exception("Exception"));
} else {
subscriber.onError(new Throwable("Throw
error"));
}
}
}
});
}
分别对其进行订阅
点击(此处)折叠或打开
mLButton.setText("onException-true");
mLButton.setOnClickListener(e -> onExceptionResumeObserver(true).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onException-true-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onException-true-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onException-true-onNext:" + s);
}
}));
mRButton.setText("onException-false");
mRButton.setOnClickListener(e -> onExceptionResumeObserver(false).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
log("onException-false-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("onException-false-onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
log("onException-false-onNext:" + s);
}
}));
运行结果如下。可以看到当发生错误时抛出的时Exception，则继续发射另外一个Observable的数据并正常结束；而抛出Throwable类型数据时，错误就分发到了Subscriber。
二、Retry
Retry操作符在发生错误的时候会重新进行订阅,而且可以重复多次，所以发射的数据可能会产生重复。如果重复指定次数还有错误的话就会将错误返回给观察者
Rxjava还实现了RetryWhen操作符。当错误发生时，retryWhen会接收onError的throwable作为参数，并根据定义好的函数返回一个Observable，如果这个Observable发射一个数据，就会重新订阅。
下面我们使用retry和retryWhen来处理两个Observable产生的错误，需要注意的是使用retryWhen的时候,因为每次重新订阅都会产生错误，所以作为参数的obserbvable会不断地发射数据，使用zipWith操作符可以限制重新订阅的次数，否则会无限制地重新订阅。
点击(此处)折叠或打开
private Observable<Integer> retryObserver() {
return createObserver().retry(2);
}
private Observable<Integer> retryWhenObserver() {
return createObserver().retryWhen(observable -> observable.zipWith(Observable.just(1, 2, 3),
(throwable, integer) -> throwable.getMessage() + integer)
.flatMap(throwable -> {
log(throwable);
return Observable.timer(1, TimeUnit.SECONDS);
}));
}
private Observable<Integer> createObserver() {
return Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
log("subscribe");
for (int i = 0; i < 3; i++) {
if (i == 2) {
subscriber.onError(new Exception("Exception-"));
} else {
subscriber.onNext(i);
}
}
}
});
}
分别进行订阅
点击(此处)折叠或打开
mLButton.setText("retry");
mLButton.setOnClickListener(e -> retryObserver().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
log("retry-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("retry-onError:" + e.getMessage());
}
@Override
public void onNext(Integer s) {
log("retry-onNext:" + s);
}
}));
mRButton.setText("retryWhen");
mRButton.setOnClickListener(e -> retryWhenObserver().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
log("retryWhen-onCompleted\n");
}
@Override
public void onError(Throwable e) {
log("retryWhen-onError:" + e.getMessage());
}
@Override
public void onNext(Integer s) {
log("retryWhen-onNext:" + s);
}
}));
}
运行结果如下。在尝试了几次还是产生错误后，retry会将错误分发给观察者，而retryWhen会正常结束，并不会讲错误分发出去。
关于错误的处理就到这里了，本文中的程序见：https://github.com/Chaoba/RxJavaDemo
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- RxJava操作符（四）Combining
- RxJava操作符（三）Filtering
- RxJava操作符（二）Transforming Observables
- RxJava操作符（一）Creating Observables
- jdk1.7实现ftp上传文件
- java 利用集合实现斗地主算法
- java.util.Vector
- Java垃圾回收机制
- 【spring】AOP注解实现
- java打包遇到问题java.io.IOException: invalid header field
- "Calendars" and "DateFormats" should not be static
- eclipse中Java连接sql server数据库
- java基础十［包、Jar存档文件和部署］（阅读Head First Java记录）
- Spring源码阅读（2）- bean解析初体验
- 单向链表的插入排序
- Java Socket实现文件传输示例代码
- Java设计模式之享元模式
- Java 容器相关知识全面总结
- 从一个简单的Java单例示例谈谈并发
- 你真的了解一段Java程序的生命史吗