代码之家 › 专栏 › 技术社区 › Denis Itskovich

RxJava-重试:成功恢复时重置重试计数器

rx-java2 rx-java java

Denis Itskovich · 技术社区 · 8 年前

目前我正在使用Observable。重试(n),以便在出现错误时重试。请查看此代码:

@Test
public void testObservableRetry() {
    Observable
            .<String>defer(() -> Observable
                    .just(Notification.createOnNext("valid value"), Notification.createOnError(new Error("Test")))
                    .dematerialize())
            .doOnNext(v -> System.out.println("onNext: " + v))
            .doOnError(e -> System.out.println("onError: " + e.getMessage()))
            .doOnSubscribe(d -> System.out.println("onSubscribe"))
            .doOnDispose(() -> System.out.println("onDispose"))
            .retry(2)
            .take(2) // .take(4) will fail the test
            .blockingLast();
}

上述代码可以在出错后恢复,但重试次数仅限于总错误数,而不是连续错误数。

E、 g.如果更换 .take(2) 具有 .take(4) -测试将失败,因为总错误数将超过4,尽管每次错误后都会恢复并能够获得下一个值:

---(v1)----(error)---(v2)----(error)----(v3)-----(error)---(v4)-

我想找到一种方法,在成功恢复后重置计数器。用例是例如网络连接-我希望在每次断开连接时提供相同的尝试次数,但在每次成功连接后,我希望重置计数器以允许无休止的流量,除非每次都能够在恒定的尝试次数内恢复。

编辑:

在此上下文中成功恢复意味着在重试后至少接收1个项目。所以我只想限制连续错误的数量,而不是总的错误。

2 回复 | 直到 8 年前

Denis Itskovich 8 年前

这只是对akarnokd的解决方案进行了一点修改(只是删除 nonEmpty 标志)@akarnokd-自由填写以编辑您的解决方案,我将删除此解决方案。

static <T> ObservableTransformer<T, T> retryEmpty(int count) {
    return o -> {
        AtomicInteger remaining = new AtomicInteger(count);
        return o.doOnNext(v -> remaining.lazySet(count))
                .retryWhen(err -> err.flatMap(e ->
                                (remaining.decrementAndGet() == 0)
                                        ? Observable.<T>error(e)
                                        : Observable.just(1)));
    };
}

akarnokd 8 年前

你需要 retryWhen 并在流程之前与it进行沟通:

@Test
public void emptyError() {
    AtomicInteger c = new AtomicInteger();
    Observable.fromCallable(() -> { throw new Exception("" + c.incrementAndGet()); })
    .compose(retryEmpty(2))
    .test()
    .assertFailureAndMessage(Exception.class, "2");
}

@Test
public void nonEmptyError() {
    Observable.just(1).concatWith(Observable.error(new Exception()))
    .compose(retryEmpty(2))
    .take(4)
    .test()
    .assertResult(1, 1, 1, 1);
}

static <T> ObservableTransformer<T, T> retryEmpty(int count) {
    return o ->
        Observable.defer(() -> {
            AtomicInteger remaining = new AtomicInteger(count);
            AtomicBoolean nonEmpty = new AtomicBoolean();

            return o.doOnNext(v -> nonEmpty.lazySet(true))
            .retryWhen(err -> 
                err.flatMap(e -> {
                    if (nonEmpty.get()) {
                        nonEmpty.lazySet(false);
                        remaining.lazySet(count);
                    } else
                    if (remaining.decrementAndGet() == 0) {
                        return Observable.error(e);
                    }
                    return Observable.just(1);
                })
             );
        });
}