rx-java - 错误后重试同一项目

Question

我正在尝试使用 rx-java 构建一个强大的处理管道，但遇到了问题。

这是一个例子:

public static void main(String[] args) {
    AtomicInteger div = new AtomicInteger(-1);
    Observable.just(1, 1, 1).map(item -> 1 / div.getAndIncrement())
            .retry().subscribe(item -> System.out.println(item));
}

本例中的输出是 4 个项目，因为非流可观察值被重播，但这并不相关，因此为了简单起见，请忽略它。我添加了注释，显示了达到结果的计算以及重新订阅的点:

-1 // 1 / -1
// 1/0 (error) - resubscribes to observable
1 // 1 / 1
0 // 1 / 2
0 // 1 / 3

发生这种情况是因为一旦错误通知被传递，retry 运算符(与所有重试运算符一样)会导致重新订阅。

我的预期输出是:

-1 // 1 / -1
// 1/0 (error) - resubscribe but resume erroneous item (1)
1 // 1 / 1
0 // 1 / 0

当错误通知被传递时，重新订阅过程应该包括流中的错误项目(在同一项目上重试) - 因为错误是外部的并且没有嵌入到已处理的项目中(因此重新处理将使感)。

这是出现一些外部错误(例如数据库连接)的情况，我希望未处理的项目在一定延迟后重新处理。我知道可以使用标准重试运算符重新订阅，但它们都会放弃错误的项目。

我还考虑过将所有处理包装在 try-catch 中，我怀疑可能会出现错误，但将样板代码添加到我的处理代码中，但我不喜欢这样做。

所以我的问题是:是否有一种标准方法可以重试失败的项目？

我已经考虑过做类似的事情(未经测试):

someSubject.flatMap(
    item-> Observable.just(item)
        .doOnError(err -> someSubject.onNext(item))).onErrorX...

并抑制错误...

但这似乎不自然，并且在我的用例中成本高昂(为每个项目创建一个可观察对象)。

是否有一个运算符或运算符组合可以导致重试将错误的项目传递回可观察的开头，而不会“破坏”或将项目包装在不同的可观察中？

这也是我习惯使用 async-retry 的重试方式.

Answer 1

这对于 RxJava 通常是不可能的。如果元素的处理失败，则没有内置方法可以从该位置恢复。您能做的最好的事情就是 try catch 有问题的函数回调并手动重试。第二好的方法是使用 flatMap，其中可能有问题的计算是可以单独重试的内部 Observable:

source.flatMap(v ->
    Observable.just(v).map(v -> v / counter.getAndIncrement()).retry()
)