Observable.interval()网络请求失败后如何恢复间隔轮询？

我使用Observable.interval()通过SDK每5秒轮询一次API接口。网络请求存在报错概率（通常前3次请求会出错），这会导致Observable.interval()仅执行一次后就停止运行。我该如何实现捕获错误并让间隔轮询持续执行？

原始代码示例 #

Single.fromCallable(() -> {
    return remoteRepository.performFirstQuery();
})
     .flatMapObservable(txHash -> {
         // 希望用txHash每5秒轮询一次，直到请求无错误
         return Observable.interval(0, 5, TimeUnit.SECONDS)
              .map(n -> {
                 return remoteRepository.performSecondQuery(txHash).execute();
               });
     })
     .subscribeOn(Schedulers.io())
     .observeOn(AndroidSchedulers.mainThread())
     .subscribe(txResult -> {

     }, err -> {

     });

编辑补充 #

我按照建议添加了.onErrorResumeNext()，但出现一次错误后，Observable.interval()的间隔触发就停止了。

修改后的代码：

.flatMapObservable(txHash -> {
      return Observable.interval(0, 5, TimeUnit.SECONDS)
       .map(n -> {
             return remoteRepository.performSecondQuery(txHash).execute();
       }).onErrorResumeNext(throwable -> {
            // 这个错误块会被调用，但只触发一次
            return Observable.empty();
      });
})

解决方案 #

问题出在错误处理的层级：你在Observable.interval()的整个流上添加了错误处理，一旦map里的请求报错，整个轮询流会直接终止并返回Observable.empty()。正确的做法是在单个网络请求的子流中处理错误，确保单个请求失败不会影响interval的定时发射逻辑。

修正后的代码如下：

Single.fromCallable(() -> {
    return remoteRepository.performFirstQuery();
})
     .flatMapObservable(txHash -> {
         return Observable.interval(0, 5, TimeUnit.SECONDS)
              // 将同步请求包装为Observable，在子流中隔离错误
              .flatMap(n -> Observable.fromCallable(() -> {
                  return remoteRepository.performSecondQuery(txHash).execute();
              })
              .onErrorResumeNext(throwable -> {
                  // 捕获单次请求的错误，可在这里记录日志或做错误处理
                  // 返回空Observable表示本次请求失败，但不终止轮询
                  return Observable.empty();
              }));
     })
     .subscribeOn(Schedulers.io())
     .observeOn(AndroidSchedulers.mainThread())
     .subscribe(txResult -> {
         // 请求成功时的处理逻辑
     }, err -> {
         // 这里只会捕获到interval流本身的致命错误，而非单次请求错误
     });

关键说明 #

1. 用flatMap替代map：将每个定时触发的请求包装成独立的Observable.fromCallable子流，这样单个请求的错误只会影响当前子流，不会向上传递终止整个轮询流。
2. 子流内处理错误：在fromCallable的子流上添加onErrorResumeNext，捕获单次请求的错误后返回Observable.empty()（或根据业务需求返回其他占位值），确保interval能继续发射下一个定时事件。

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