异常处理

在Rxjava订阅的Observable有时会抛出异常,在RxJava中有两大类策略,一个是准备备用的Observable,在发生异常时将subscriber订阅到新的Observable上,第二种是重试。

catch

第一种策略是捕获异常并使用新的Observable来替换出现异常的Observable

方法有

onErrorReturn:返回单个对象

onErrorResumeNext:当抛出Throwable时返回一个新的Observable

onExceptionResumeNext:同onErrorResumeNext但只有抛出Exception时才会触发

@Test
public void onException() {
Observable<Object> obs = Observable
.create(sub -> {
for (int i = 0; i < 10; i++) {
if (i == 1) {
//抛出异常
sub.onError(new RuntimeException("error"));
}
sub.onNext(i);
}
});
obs.onExceptionResumeNext(Observable.just("new resumed observable"))
.subscribe(obj -> System.out.println(obj));
}
-------输出---------
0
new resumed observable

从输出中看出当遇到异常后会调用注册的备用Observable

Retry

该策略当遇到exception时会进行重试

有两类方法

retry:可以根据重试次数和错误原因来判断是否需要重试

@Test
public void retry() {
Observable<Object> obs = Observable
.create(sub -> {
for (int i = 0; i < 10; i++) {
if (i == 1) {
sub.onError(new RuntimeException("error"));
}
sub.onNext(i);
}
});
obs.retry((time,ex)->{
if(time==2 && ex instanceof RuntimeException){
return false;
}
return true;
})
.subscribe(obj -> System.out.println(obj));
}
-------输出---------
0
0 rx.exceptions.OnErrorNotImplementedException: error

可以看到retry中判断次数和错误类型根据返回决定是否要重试

retryWhen:通过将异常组成Observable并通过返回的Observable的结果来判断是否重试

下面是官方的例子,第一次会1秒重试,第二次等待2秒,第三次等待3秒,第四次停止重试

@Test
public void retryWhen(){
Observable.create((s) -> {
s.onNext("subcribe");
s.onError(new RuntimeException("always fails"));
}).retryWhen(attempts -> {
return attempts.zipWith(Observable.range(1, 3), (n, i) -> i).flatMap(i -> {
System.out.println("delay retry by " + i + " second(s)");
return Observable.timer(i, TimeUnit.SECONDS);
});
}).toBlocking().forEach(System.out::println);
}
-------输出---------
subcribe
delay retry by 1 second(s)
subcribe
delay retry by 2 second(s)
subcribe
delay retry by 3 second(s)
subcribe
04-15 02:59