如何添加超时以检测 Observable 有一段时间没有发出

Question

我的用例如下：

我创建了一个 Observable 来访问远程服务器以获取一些数据。但是，由于服务器设计不善，因此永远不会有任何响应和例外。为了解决这个问题，我想要一些超时重试机制。

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

对于 RXJAVA 中的任何操作符，我应该如何处理这种情况？

我当前的代码如下所示：

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()。