很多操作符可用于对 Observable 发射的 onError 通知做出响应或者从错误中恢复，例如，你可以：
- 捕捉这个错误，切换到一个备用的 Observable 继续发射数据。
- 捕捉这个错误然后发射默认值。
- 捕捉这个错误并立即尝试重启这个 Observable。
- 捕捉这个错误，在一些回退间隔后重启这个 Observable。
Catch — 从 onError 通知中恢复发射数据。
Retry — 如果原始 Observable 遇到错误，重新订阅它期望它能正常终止。
5.1 Catch
从 onError 通知中恢复发射数据。
Catch
Catch 操作符拦截原始 Observable 的 onError 通知，将它替换为其它的数据项或数据序列，让产生的 Observable 能够正常终止或者根本不终止。
RxJava 将 Catch 实现为三个不同的操作符：onErrorReturn ，onErrorResumeNext，onExceptionResumeNext。
5.1.1 onErrorReturn
让 Observable 遇到错误时发射一个特殊的项并且正常终止。
onErrorReturn
示例代码：
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 并继续发射数据。
onErrorResumeNext
它有两种实现
- OnErrorResumeNext(Observable);出现错误时取到当前 Observable 并做处理。
- 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。
onExceptionResumeNext
示例代码：
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 遇到错误，重新订阅它期望它能正常终止。
Retry
Retry 操作符不会将原始 Observable 的 onError 通知传递给观察者，它会订阅这个 Observable，再给它一次机会无错误地完成它的数据序列。Retry 总是传递 onNext 通知给观察者，由于重新订阅，可能会造成数据项重复，如上图所示。
RxJava 中的实现为 Retry 和 RetryWhen。
5.2.1 Retry
无论收到多少次 onError 通知，无参数版本的 retry 都会继续订阅并发射原始 Observable。
Retry 的三种实现方法：
- retry(long) ：会最多重新订阅指定的次数，如果次数超了，它不会尝试再次订阅，它会把最新的一个 onError 通知传递给它的观察者。
- retry()： 等价于 retry(Long.MAX_VALUE) ;
- 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
RetryWhen
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 会正常结束，并不会讲错误分发出去。
