如果原始Observable遇到错误,重新订阅它期望它能正常终止
Retry操作符不会将原始Observable的onError通知传递给观察者,它会订阅这个Observable,再给它一次机会无错误地完成它的数据序列。Retry总是传递onNext通知给观察者,由于重新订阅,可能会造成数据项重复,如上图所示。
Retry
onError
onNext
RxJava中的实现为retry和retryWhen。
retry
retryWhen
无论收到多少次onError通知,无参数版本的retry都会继续订阅并发射原始Observable。
接受单个count参数的retry会最多重新订阅指定的次数,如果次数超了,它不会尝试再次订阅,它会把最新的一个onError通知传递给它的观察者。
count
还有一个版本的retry接受一个谓词函数作为参数,这个函数的两个参数是:重试次数和导致发射onError通知的Throwable。这个函数返回一个布尔值,如果返回true,retry应该再次订阅和镜像原始的Observable,如果返回false,retry会将最新的一个onError通知传递给它的观察者。
Throwable
true
false
retry操作符默认在trampoline调度器上执行。
trampoline
retryWhen和retry类似,区别是,retryWhen将onError中的Throwable传递给一个函数,这个函数产生另一个Observable,retryWhen观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据,它就重新订阅,如果这个Observable发射的是onError通知,它就将这个通知传递给观察者然后终止。
retryWhen默认在trampoline调度器上执行,你可以通过参数指定其它的调度器。
示例代码
Observable.create((Subscriber<? super String> s) -> { System.out.println("subscribing"); 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);
输出
subscribing delay retry by 1 second(s) subscribing delay retry by 2 second(s) subscribing delay retry by 3 second(s) subscribing
Copyright© 2013-2020
All Rights Reserved 京ICP备2023019179号-8