rx-java - 如何添加超时来检测 Observable 有一段时间没有发出

Question

我的用例如下:

我创建了一个 Observable 来访问远程服务器以获取一些数据。但由于服务器设计不完善，可能永远没有响应，也不会出现异常。为了解决这个问题，我想要一些超时重试机制。

目前，我尝试启动一个计时器来停止请求并在其中抛出异常，然后重试，直到一定次数的尝试或我的实际超时。我尝试使用 mergeWith 运算符将请求与映射的 Observable.interval 合并，以使用 Observable.error() 生成错误，但是我可以' t 在订阅者中捕获错误，并且看起来 Observable.interval 永远不会结束。

我应该如何处理RXJAVA中任何运算符(operator)的这种情况？

我当前的代码如下所示:

Observable.fromEmitter(fetchNetwork->...)
.mergeWith(Observable.interval(...)
           .flatmap(n->(observable.error)))
.retryWhen(error->(checkTimeExceed))
.subscribe(handleResult)

Answer 1

您可以将timeout()运算符与retryWhen()一起使用:

Observable.fromEmitter(fetchNetwork->...)
     .timeout(TIMEOUT_VALUE, TimeUnit.SECONDS)
     .retryWhen(observable -> observable.flatMap(error -> {
                if (error instanceof TimeoutException) {
                    return Observable.just(new Object());
                } else {
                    return Observable.error(error);
                }
            }))
     .subscribe(handleResult)

这将使请求在 TIMEOUT_VALUE 秒后超时，并且只要请求超时就会重试，其他错误将照常传播到订阅者 onError()。