题
我在 Android 应用程序中使用 rxjava 来异步处理网络请求。现在我想仅在经过一定时间后重试失败的网络请求。
有没有办法在 Observable 上使用 retry() 但仅在一定延迟后重试?
有没有办法让 Observable 知道当前正在重试(而不是第一次尝试)?
我查看了 debounce()/throttleWithTimeout() 但他们似乎在做一些不同的事情。
编辑:
我想我找到了一种方法来做到这一点,但我对确认这是正确的方法或其他更好的方法感兴趣。
我正在做的是这样的:在 Observable.OnSubscribe 的 call() 方法中,在调用订阅者的 onError() 方法之前,我只需让线程休眠所需的时间。因此,要每 1000 毫秒重试一次,我会执行以下操作:
@Override
public void call(Subscriber<? super List<ProductNode>> subscriber) {
try {
Log.d(TAG, "trying to load all products with pid: " + pid);
subscriber.onNext(productClient.getProductNodesForParentId(pid));
subscriber.onCompleted();
} catch (Exception e) {
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e.printStackTrace();
}
subscriber.onError(e);
}
}
由于该方法无论如何都在 IO 线程上运行,因此它不会阻塞 UI。我看到的唯一问题是,即使是第一个错误也会延迟报告,因此即使没有 retry(),延迟也会存在。如果不应用延迟我会更好 后 一个错误,而是 前 重试(显然不是在第一次尝试之前)。
解决方案
您可以使用retryWhen()
运算符将重试逻辑添加到任何可观察到的。
以下类包含重试逻辑:
rxjava 2.x
public class RetryWithDelay implements Function<Observable<? extends Throwable>, Observable<?>> {
private final int maxRetries;
private final int retryDelayMillis;
private int retryCount;
public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
this.maxRetries = maxRetries;
this.retryDelayMillis = retryDelayMillis;
this.retryCount = 0;
}
@Override
public Observable<?> apply(final Observable<? extends Throwable> attempts) {
return attempts
.flatMap(new Function<Throwable, Observable<?>>() {
@Override
public Observable<?> apply(final Throwable throwable) {
if (++retryCount < maxRetries) {
// When this Observable calls onNext, the original
// Observable will be retried (i.e. re-subscribed).
return Observable.timer(retryDelayMillis,
TimeUnit.MILLISECONDS);
}
// Max retries hit. Just pass the error along.
return Observable.error(throwable);
}
});
}
}
.
rxjava 1.x
public class RetryWithDelay implements
Func1<Observable<? extends Throwable>, Observable<?>> {
private final int maxRetries;
private final int retryDelayMillis;
private int retryCount;
public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
this.maxRetries = maxRetries;
this.retryDelayMillis = retryDelayMillis;
this.retryCount = 0;
}
@Override
public Observable<?> call(Observable<? extends Throwable> attempts) {
return attempts
.flatMap(new Func1<Throwable, Observable<?>>() {
@Override
public Observable<?> call(Throwable throwable) {
if (++retryCount < maxRetries) {
// When this Observable calls onNext, the original
// Observable will be retried (i.e. re-subscribed).
return Observable.timer(retryDelayMillis,
TimeUnit.MILLISECONDS);
}
// Max retries hit. Just pass the error along.
return Observable.error(throwable);
}
});
}
}
.
用法:
// Add retry logic to existing observable.
// Retry max of 3 times with a delay of 2 seconds.
observable
.retryWhen(new RetryWithDelay(3, 2000));
. 其他提示
灵感来自保罗的答案,如果你不关心abhijit sarkar ,延迟重新提交的最简单方法是:
source.retryWhen(throwables -> throwables.delay(1, TimeUnit.SECONDS))
.
您可能希望查看更多关于重试和重复
此示例适用于JXJava 2.2.2:
毫延迟重试:
Single.just(somePaylodData)
.map(data -> someConnection.send(data))
.retry(5)
.doOnSuccess(status -> log.info("Yay! {}", status);
.
用延迟重试:
Single.just(somePaylodData)
.map(data -> someConnection.send(data))
.retryWhen((Flowable<Throwable> f) -> f.take(5).delay(300, TimeUnit.MILLISECONDS))
.doOnSuccess(status -> log.info("Yay! {}", status)
.doOnError((Throwable error)
-> log.error("I tried five times with a 300ms break"
+ " delay in between. But it was in vain."));
.
如果someConnection.send()失败,我们的源单个失败。 发生这种情况时,在重试内失败的可观察到发出错误。 我们延迟发射300ms并将其发送回信号重试。 拿(5)保证我们在获得五个错误后终止我们的信令可观察。 重试,当看到终止并且在第五失败后没有重试。
这是一个基于我看到的 Ben Christensen 片段的解决方案, 重试示例, , 和 有条件的测试时重试 (我必须改变 n.getThrowable()
到 n
使其发挥作用)。我用了 evant/gradle-retrolambda 让 lambda 表示法在 Android 上工作,但您不必使用 lambda(尽管强烈推荐)。对于延迟,我实现了指数退避,但您可以在那里插入任何您想要的退避逻辑。为了完整起见,我添加了 subscribeOn
和 observeOn
运营商。我在用着 ReactiveX/RxAndroid 为了 AndroidSchedulers.mainThread()
.
int ATTEMPT_COUNT = 10;
public class Tuple<X, Y> {
public final X x;
public final Y y;
public Tuple(X x, Y y) {
this.x = x;
this.y = y;
}
}
observable
.subscribeOn(Schedulers.io())
.retryWhen(
attempts -> {
return attempts.zipWith(Observable.range(1, ATTEMPT_COUNT + 1), (n, i) -> new Tuple<Throwable, Integer>(n, i))
.flatMap(
ni -> {
if (ni.y > ATTEMPT_COUNT)
return Observable.error(ni.x);
return Observable.timer((long) Math.pow(2, ni.y), TimeUnit.SECONDS);
});
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);
而不是使用myRequestobservable.retry我使用包装器函数retryogservable(myRequestobservable,Retrycount,秒),它返回一个新的可观察到处理延迟的间接,所以我可以做
retryObservable(restApi.getObservableStuff(), 3, 30)
.subscribe(new Action1<BonusIndividualList>(){
@Override
public void call(BonusIndividualList arg0)
{
//success!
}
},
new Action1<Throwable>(){
@Override
public void call(Throwable arg0) {
// failed after the 3 retries !
}});
// wrapper code
private static <T> Observable<T> retryObservable(
final Observable<T> requestObservable, final int nbRetry,
final long seconds) {
return Observable.create(new Observable.OnSubscribe<T>() {
@Override
public void call(final Subscriber<? super T> subscriber) {
requestObservable.subscribe(new Action1<T>() {
@Override
public void call(T arg0) {
subscriber.onNext(arg0);
subscriber.onCompleted();
}
},
new Action1<Throwable>() {
@Override
public void call(Throwable error) {
if (nbRetry > 0) {
Observable.just(requestObservable)
.delay(seconds, TimeUnit.SECONDS)
.observeOn(mainThread())
.subscribe(new Action1<Observable<T>>(){
@Override
public void call(Observable<T> observable){
retryObservable(observable,
nbRetry - 1, seconds)
.subscribe(subscriber);
}
});
} else {
// still fail after retries
subscriber.onError(error);
}
}
});
}
});
}
. retryWhen
是一个复杂的,甚至可能是有缺陷的操作符。官方文档和至少一个答案在这里使用 range
运算符,如果没有重试,该运算符将会失败。看我的 讨论 与 ReactiveX 成员 David Karnok 一起。
我通过改变改进了 kjones 的答案 flatMap
到 concatMap
并通过添加 RetryDelayStrategy
班级。 flatMap
不保留发射顺序 concatMap
确实如此,这对于退避延迟很重要。这 RetryDelayStrategy
, 顾名思义,让用户从生成重试延迟的各种模式中进行选择,包括回退。该代码可以在我的 GitHub 完成以下测试用例:
- 第一次尝试成功(无需重试)
- 重试 1 次后失败
- 尝试重试 3 次,但第二次成功,因此不会重试第三次
- 第三次重试成功
看 setRandomJokes
方法。
将修改添加到 kjones 答案。
修改
public class RetryWithDelay implements
Func1<Observable<? extends Throwable>, Observable<?>> {
private final int MAX_RETRIES;
private final int DELAY_DURATION;
private final int START_RETRY;
/**
* Provide number of retries and seconds to be delayed between retry.
*
* @param maxRetries Number of retries.
* @param delayDurationInSeconds Seconds to be delays in each retry.
*/
public RetryWithDelay(int maxRetries, int delayDurationInSeconds) {
MAX_RETRIES = maxRetries;
DELAY_DURATION = delayDurationInSeconds;
START_RETRY = 1;
}
@Override
public Observable<?> call(Observable<? extends Throwable> observable) {
return observable
.delay(DELAY_DURATION, TimeUnit.SECONDS)
.zipWith(Observable.range(START_RETRY, MAX_RETRIES),
new Func2<Throwable, Integer, Integer>() {
@Override
public Integer call(Throwable throwable, Integer attempt) {
return attempt;
}
});
}
}
. 与 kjones 但更新为最新版本 对于 rxjava 2.x 版本:('io.reacctivex.rxjava2:rxjava:2.1.3')
public class RetryWithDelay implements Function<Flowable<Throwable>, Publisher<?>> {
private final int maxRetries;
private final long retryDelayMillis;
private int retryCount;
public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
this.maxRetries = maxRetries;
this.retryDelayMillis = retryDelayMillis;
this.retryCount = 0;
}
@Override
public Publisher<?> apply(Flowable<Throwable> throwableFlowable) throws Exception {
return throwableFlowable.flatMap(new Function<Throwable, Publisher<?>>() {
@Override
public Publisher<?> apply(Throwable throwable) throws Exception {
if (++retryCount < maxRetries) {
// When this Observable calls onNext, the original
// Observable will be retried (i.e. re-subscribed).
return Flowable.timer(retryDelayMillis,
TimeUnit.MILLISECONDS);
}
// Max retries hit. Just pass the error along.
return Flowable.error(throwable);
}
});
}
}
.
用法:
//将重试逻辑添加到现有可观察到。 //用2秒的延迟重试max 3次。
observable
.retryWhen(new RetryWithDelay(3, 2000));
. 您可以在重试运算符中返回的可观察者返回的延迟
/**
* Here we can see how onErrorResumeNext works and emit an item in case that an error occur in the pipeline and an exception is propagated
*/
@Test
public void observableOnErrorResumeNext() {
Subscription subscription = Observable.just(null)
.map(Object::toString)
.doOnError(failure -> System.out.println("Error:" + failure.getCause()))
.retryWhen(errors -> errors.doOnNext(o -> count++)
.flatMap(t -> count > 3 ? Observable.error(t) : Observable.just(null).delay(100, TimeUnit.MILLISECONDS)),
Schedulers.newThread())
.onErrorResumeNext(t -> {
System.out.println("Error after all retries:" + t.getCause());
return Observable.just("I save the world for extinction!");
})
.subscribe(s -> System.out.println(s));
new TestSubscriber((Observer) subscription).awaitTerminalEvent(500, TimeUnit.MILLISECONDS);
}
.
您可以在此处查看更多示例。 https://github.com/politrons/reactive
基于 kjones 答案在这里是rxjava 2.x重试延迟作为扩展的kotlin版本。替换生成的Observable
以创建生成的Flowable
的相同扩展名。
fun <T> Observable<T>.retryWithDelay(maxRetries: Int, retryDelayMillis: Int): Observable<T> {
var retryCount = 0
return retryWhen { thObservable ->
thObservable.flatMap { throwable ->
if (++retryCount < maxRetries) {
Observable.timer(retryDelayMillis.toLong(), TimeUnit.MILLISECONDS)
} else {
Observable.error(throwable)
}
}
}
}
.
然后只使用它在可观察到的observable.retryWithDelay(3, 1000)
只是这样做:
Observable.just("")
.delay(2, TimeUnit.SECONDS) //delay
.flatMap(new Func1<String, Observable<File>>() {
@Override
public Observable<File> call(String s) {
L.from(TAG).d("postAvatar=");
File file = PhotoPickUtil.getTempFile();
if (file.length() <= 0) {
throw new NullPointerException();
}
return Observable.just(file);
}
})
.retry(6)
.subscribe(new Action1<File>() {
@Override
public void call(File file) {
postAvatar(file);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
}
});
. 对于kotlin&rxjava1版本
class RetryWithDelay(private val MAX_RETRIES: Int, private val DELAY_DURATION_IN_SECONDS: Long)
: Function1<Observable<out Throwable>, Observable<*>> {
private val START_RETRY: Int = 1
override fun invoke(observable: Observable<out Throwable>): Observable<*> {
return observable.delay(DELAY_DURATION_IN_SECONDS, TimeUnit.SECONDS)
.zipWith(Observable.range(START_RETRY, MAX_RETRIES),
object : Function2<Throwable, Int, Int> {
override fun invoke(throwable: Throwable, attempt: Int): Int {
return attempt
}
})
}
}
. (kotlin)我一点改进了具有指数退避和应用防御的代码和应用的防御发射.range():
fun testOnRetryWithDelayExponentialBackoff() {
val interval = 1
val maxCount = 3
val ai = AtomicInteger(1);
val source = Observable.create<Unit> { emitter ->
val attempt = ai.getAndIncrement()
println("Subscribe ${attempt}")
if (attempt >= maxCount) {
emitter.onNext(Unit)
emitter.onComplete()
}
emitter.onError(RuntimeException("Test $attempt"))
}
// Below implementation of "retryWhen" function, remove all "println()" for real code.
val sourceWithRetry: Observable<Unit> = source.retryWhen { throwableRx ->
throwableRx.doOnNext({ println("Error: $it") })
.zipWith(Observable.range(1, maxCount)
.concatMap { Observable.just(it).delay(0, TimeUnit.MILLISECONDS) },
BiFunction { t1: Throwable, t2: Int -> t1 to t2 }
)
.flatMap { pair ->
if (pair.second >= maxCount) {
Observable.error(pair.first)
} else {
val delay = interval * 2F.pow(pair.second)
println("retry delay: $delay")
Observable.timer(delay.toLong(), TimeUnit.SECONDS)
}
}
}
//Code to print the result in terminal.
sourceWithRetry
.doOnComplete { println("Complete") }
.doOnError({ println("Final Error: $it") })
.blockingForEach { println("$it") }
}
. 如果需要打印重试计数, 您可以使用rxjava的wiki页面 https://github.com/Reactivex / Rxjava / Wiki /错误处理运算符
observable.retryWhen(errors ->
// Count and increment the number of errors.
errors.map(error -> 1).scan((i, j) -> i + j)
.doOnNext(errorCount -> System.out.println(" -> query errors #: " + errorCount))
// Limit the maximum number of retries.
.takeWhile(errorCount -> errorCount < retryCounts)
// Signal resubscribe event after some delay.
.flatMapSingle(errorCount -> Single.timer(errorCount, TimeUnit.SECONDS));
.