rxjava:我可以使用retry()但有延迟吗? [英] rxjava: Can I use retry() but with delay?

问题描述

我在我的Android应用中使用rxjava异步处理网络请求.现在，我只想在经过一定时间后重试失败的网络请求.

I am using rxjava in my Android app to handle network requests asynchronously. Now I would like to retry a failed network request only after a certain time has passed.

有什么方法可以在Observable上使用retry()，但是只能在一定的延迟后重试吗?

Is there any way to use retry() on an Observable but to retry only after a certain delay?

有没有办法让Observable知道当前正在重试(而不是第一次尝试)?

Is there a way to let the Observable know that is is currently being retried (as opposed to tried for the first time)?

我看了看debounce()/throttleWithTimeout()，但是他们似乎在做些不同的事情.

I had a look at debounce()/throttleWithTimeout() but they seem to be doing something different.

我认为我找到了一种方法来做，但是我对确认这是正确的方法还是其他更好的方法感兴趣.

I think I found one way to do it, but I'd be interested in either confirmation that this is the correct way to do it or for other, better ways.

我正在做什么:在我的Observable.OnSubscribe的call()方法中，在调用Subscribers onError()方法之前，我只是让线程休眠了所需的时间.因此，要每1000毫秒重试一次，我会执行以下操作:

What I am doing is this: In the call() method of my Observable.OnSubscribe, before I call the Subscribers onError() method, I simply let the Thread sleep for the desired amount of time. So, to retry every 1000 milliseconds, I do something like this:

@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()，延迟也仍然存在.如果延迟不是在错误发生后 而是在重试发生前 (但显然不是在第一次尝试之前)，我会更好.

Since this method is running on an IO thread anyway it does not block the UI. The only problem I can see is that even the first error is reported with delay so the delay is there even if there's no retry(). I'd like it better if the delay wasn't applied after an error but instead before a retry (but not before the first try, obviously).

推荐答案

您可以使用 retryWhen()运算符将重试逻辑添加到任何Observable中.

You can use the retryWhen() operator to add retry logic to any Observable.

以下类包含重试逻辑:

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));

这篇关于rxjava:我可以使用retry()但有延迟吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！