如何在 RxJava 出错后继续流式传输项目？

Question

我是 RxJava 新手，遇到了以下问题。假设我有一个项目序列和一个项目传播错误，我想忽略它并继续处理其他项目。

我有以下片段：

    Observable.from(Arrays.asList("1", "2", "3"))
            .map(x -> {
                if (x.equals("2")) {
                    throw new NullPointerException();
                }
                return x + "-";
            })
            .onExceptionResumeNext(Observable.empty())
            .subscribe(System.out::println);

我得到：1-

但我想得到：1- , 3-

我该怎么做？

Answer 1

诀窍是将值以某种方式转换为新的可观察对象和平面图，如下例所示。 flatMap 中的每个值现在都可以抛出异常并逐个处理。因为 flatMap 中的子流只包含一个元素，所以在 onError 之后是否关闭 observable 并不重要。我使用 RxJava2 作为测试环境。

@Test
public void name() throws Exception {
    Observable<String> stringObservable = Observable.fromArray("1", "2", "3")
            .flatMap(x -> {
                return Observable.defer(() -> {
                    try {
                        if (x.equals("2")) {
                            throw new NullPointerException();
                        }
                        return Observable.just(x + "-");
                    } catch (Exception ex) {
                        return Observable.error(ex);
                    }
                }).map(s -> {
                    if (s.equals("3-")) {
                        throw new IllegalArgumentException();
                    }
                    return s + s;
                }).take(1)
                        .zipWith(Observable.just("X"), (s, s2) -> s + s2)
                        .onErrorResumeNext(Observable.empty());
            });

    TestObserver<String> test = stringObservable.test();

    test.assertResult("1-1-X");
}

如何在 RxJava 出错后继续流式传输项目？

How to continue streaming items after error in RxJava?

java

reactive-programming

rx-java

rx-android