RxJava----操作符：错误处理

这些操作符用于从错误通知中恢复

catch

Catch类似于java中的try/catch，当错误发生的时候，可以拦截对onError的调用，让Observable不会因为错误的产生而终止。在Rxjava中，将这个操作符实现为3个操作符，分别是：

onErrorReturn

当发生错误的时候，让Observable发射一个预先定义好的数据并正常地终止

Observable<String> values =Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("1");
subscriber.onNext("2");
subscriber.onError(new Exception("It's a exception"));
}});
values.onErrorReturn(new Func1<Throwable, String>() {
@Override
public String call(Throwable throwable) {
return "Error: " + throwable.getMessage();
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
log(s);
}
});

结果：

1
2
Error:It's a exception

onErrorResumeNext

当发生错误的时候，由另外一个Observable来代替当前的Observable并继续发射数据.在返回的 Observable 中是看不到错误信息的。

public final Observable<T> onErrorResumeNext(
Observable<? extends T> resumeSequence)
public final Observable<T> onErrorResumeNext(
Func1<java.lang.Throwable,? extends Observable<? extends T>> resumeFunction)

第二个重载的函数可以根据错误的信息来返回不同的 Observable。

Observable<String> values =Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("1");
subscriber.onNext("2");
subscriber.onError(new Exception("It's a exception"));
}});
values.onErrorResumeNext(Observable.just("7", "8", "9")).subscribe(new Action1<String>() {
@Override
public void call(String s) {
log(s);
}
});

结果：

1
2
7
8
9

onExceptionResumeNext

onExceptionResumeNext 和 onErrorResumeNext 的区别是只捕获 Exception；

Observable<String> values = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("1");
subscriber.onNext("2");
//                subscriber.onError(new Throwable() {}); // 这个为 error 不会捕获
subscriber.onError(new Exception("It's a exception"));// 这个为 Exception 会被捕获
}});

values.onExceptionResumeNext(Observable.just("hard")).subscribe(new Action1<String>() {
@Override
public void call(String s) {
log(s);
}
});

结果：

1
2
hard

retry

重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止

Retry操作符在发生错误的时候会重新进行订阅,而且可以重复多次，所以发射的数据可能会产生重复。如果重复指定次数还有错误的话就会将错误返回给观察者

public final Observable<T> retry()
public final Observable<T> retry(long count)

没有参数的 retry() 函数会一直重试，直到没有异常发生为止。而带有参数的 retry(n) 函数会重试 N 次， 如果 N 次后还是失败，则不再重试了，数据流发射一个异常信息并结束。

final Random random = new Random();
Observable<Integer> values = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
subscriber.onNext(random.nextInt() % 20);
subscriber.onNext(random.nextInt() % 20);
subscriber.onError(new Exception());
}});

values.retry(1).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
log(integer.toString());
}

结果：

14
18
-10
9
It's a exceptionn

上面的示例，发射了两个数字遇到异常信息，然后重试一次，又发射 两个数据遇到异常信息，然后抛出该异常并结束。

请注意：上面的示例中两次发射的数字不一样。说明 retry 并不像 replay 一样会缓存之前的数据。一般情况下，这样的情况都是不合理的。所以一般情况下，只有具有副作用的时候或者 Observable 是 hot 的时候 才应该使用 retry。

retryWhen

Rxjava还实现了RetryWhen操作符。当错误发生时，retryWhen会接收onError的throwable作为参数，并根据定义好的函数返回一个Observable，如果这个Observable发射一个数据，就会重新订阅。

public final Observable<T> retryWhen(
Func1<? super Observable<? extends java.lang.Throwable>,? extends Observable<?>> notificationHandler)

retryWhen 的参数是一个函数， 该函数的输入参数为一个异常 Observable，返回值为另外一个 Observable。 输入参数中包含了 retryWhen 发生时候遇到的异常信息；返回的 Observable 为一个信号，用来判别何时需要重试的：

如果返回的 Observable 发射了一个数据，retryWhen 将会执行重试操作

如果返回的 Observable 发射了一个错误信息，retryWhen 将会发射一个错误并不会重试

如果返回的 Observable 正常结束了，retryWhen 也正常结束。

参数返回的 Observable 发射的数据类型是无关紧要的。该 Observable 的数据只是用来当做是否重试的信号。数据本身是无用的。



下面一个示例，构造一个等待 100 毫秒再重试的机制：

Observable<Integer> source = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
subscriber.onNext(1);
subscriber.onNext(2);
subscriber.onError(new Exception("Failed"));
}});

source.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override
public Observable<?> call(Observable<? extends Throwable> observable) {
return observable.take(2)
.delay(100, TimeUnit.MILLISECONDS);
}
}) .timeInterval()
.subscribe(new Action1<TimeInterval<Integer>>() {
@Override
public void call(TimeInterval<Integer> integerTimeInterval) {
log(integerTimeInterval.toString());
}
});

结果：

TimeInterval [intervalInMilliseconds=0, value=1]
TimeInterval [intervalInMilliseconds=0, value=2]
TimeInterval [intervalInMilliseconds=101, value=1]
TimeInterval [intervalInMilliseconds=1, value=2]
TimeInterval [intervalInMilliseconds=100, value=1]
TimeInterval [intervalInMilliseconds=0, value=2]

源 Observable 发射两个数字 然后遇到异常；当异常发生的时候，retryWhen 返回的 判断条件 Observable 会获取到这个异常，这里等待 100毫秒然后把这个异常当做数据发射出去告诉 retryWhen 开始重试。take(2) 参数确保判断条件 Observable 只发射两个数据（源 Observable 出错两次）然后结束。所以当源 Observable 出现两次错误以后就不再重试了。

项目源码 GitHub求赞，谢谢！

引用：

RxJava操作符（五）Error Handling-云少嘎嘎嘎-ChinaUnix博客

RxJava 教程第三部分：驯服数据流之 高级错误处理 - 云在千峰