RxJava重试：恢复成功后重置重试计数器

我太懂这个痛点了！Observable.retry(n)默认的逻辑是统计整个流生命周期内的总错误次数，而不是连续错误次数——哪怕你中间成功恢复拿到了数据，之前的错误次数还是会累计，一旦总次数达到n，流就直接终止了，这完全不符合我们想要的「每次成功后重置重试计数器」的预期。

核心解决方案：用retryWhen自定义重试逻辑 #

retryWhen是RxJava里用来完全自定义重试规则的操作符，我们可以用它来维护一个连续错误计数器，并且在每次成功发射数据时重置这个计数器。

下面是具体的实现代码示例（以RxJava 2/3为例）：

// 定义允许的最大连续错误重试次数
int maxConsecutiveRetries = 3;

// 你的原始Observable流
Observable<String> yourDataStream = Observable.create(emitter -> {
    // 这里写你的业务逻辑，比如可能抛出异常的数据源读取
    // 示例：模拟偶尔出错的情况
    if (Math.random() > 0.7) {
        emitter.onError(new RuntimeException("临时错误"));
    } else {
        emitter.onNext("有效数据");
    }
});

yourDataStream
    // 每次成功获取数据时，重置连续错误计数器
    .doOnNext(data -> {
        System.out.println("成功获取数据: " + data);
        // 重置计数器（这里用原子类保证线程安全，因为可能在多线程环境下执行）
        consecutiveErrorCount.set(0);
    })
    .retryWhen(errors -> {
        // 用AtomicInteger维护连续错误次数
        AtomicInteger consecutiveErrorCount = new AtomicInteger(0);
        
        return errors.flatMap(error -> {
            int currentErrorCount = consecutiveErrorCount.incrementAndGet();
            
            if (currentErrorCount > maxConsecutiveRetries) {
                // 连续错误次数超过阈值，终止重试，传递错误
                System.err.println("连续错误次数超限，终止流: " + error.getMessage());
                return Observable.error(error);
            }
            
            // 可选：添加重试延迟，避免频繁重试
            long delaySeconds = currentErrorCount; // 指数退避，比如第1次等1秒，第2次等2秒
            System.out.printf("第%d次重试，延迟%d秒...%n", currentErrorCount, delaySeconds);
            return Observable.timer(delaySeconds, TimeUnit.SECONDS);
        });
    })
    .take(4) // 现在哪怕take(4)也不会因为总错误数超标而失败了
    .subscribe(
        data -> {}, // 这里可以处理正常数据
        error -> System.err.println("最终错误: " + error.getMessage())
    );

代码逻辑解释 #

1. 计数器重置：通过doOnNext操作符，每次流成功发射数据时，把连续错误计数器重置为0，确保下次出错时重新开始计数。
2. 连续错误统计：在retryWhen里用AtomicInteger维护连续错误次数（因为RxJava的操作符可能在多线程环境下执行，原子类能保证线程安全）。
3. 重试终止条件：当连续错误次数超过设定的阈值时，返回Observable.error来终止重试流程，否则返回一个延迟的Observable.timer来触发下一次重试。

这样改造后，你的流就会遵循「连续错误最多重试n次，成功后重置计数器」的规则，不管你take多少个数据，只要每次错误后能在连续重试次数内恢复，就不会因为总错误数超标而终止流了。

内容的提问来源于stack exchange，提问作者Denis Itskovich