rx-java - 如果某些可观察对象抛出异常，如何继续流？

Question

我必须在 couchbase 中插入文档列表。我正在使用 RxJava 和 asyncbucket 来使用下面提到的代码。 retryWhen 函数以指数方式尝试 3 次以保存文档。

我的问题是:如果失败会怎样？另一个列表会继续保存还是会抛出错误并且 observable 会停止？如果是，我如何让可观察对象继续尝试保存剩余列表？

Observable
    .from(docs)
    .subscribeOn(Schedulers.io())
    .flatMap(docToInsert->asyncBucket.insert(docToInsert).retryWhen(retryFunc()))
    .toBlocking()
    .last();

public static RetryWhenFunction retryFunc() {
    return RetryBuilder.anyOf(TemporaryFailureException.class,RequestCancelledException.class,
            TimeoutException.class).delay(Delay.exponential(TimeUnit.SECONDS, 10)).max(3).build();
}

Answer 1

您可以将错误更改为完成，这意味着获取一个文档的失败不会影响其他文档。

更改为:

...
.flatMap(docToInsert->asyncBucket.insert(docToInsert)
                        .retryWhen(retryFunc())
                        .onErrorResumeNext(Observable.empty())
...

使用 onErrorResumeNext() 您宣布此可观察链正在正常完成，因此它不会影响任何其他链。