java - RxJava Observable 和 Subscriber 用于跳过异常？

Question

如果我有一个 Observalbe:

List<Integer> ints = Lists.newArrayList(1, 2, 0, 3, 4);
Observable<Integer> o1 = Observable.from(ints);

我想生成另一个 observable，除以 12:

Observable<Integer> o2 = o1.map(i -> 12/i);
o2.subscribe(
  v -> logger.info ("Subscriber value {}", v) ,
  t -> logger.error("Subscriber onError {} : {}", t.getClass() , t.getMessage())
);

很明显会报错，遇到'0'就停止了:

RxTest - Subscriber value 12
RxTest - Subscriber value 6
RxTest - Subscriber onError class java.lang.ArithmeticException : / by zero

但是如果我想让 Observer(o2) 跳过异常怎么办？

我查看了 RxJava 关于 error handling 的文档, 没有办法跳过错误。 onErrorResumeNext() 和 onExceptionResumeNext() 需要一个 backup/fallback Observable ，这不是我想要的。 onErrorReturn需要指定返回值。

这三种错误处理方法都不能恢复原来的观察者。例如:

Observable<Integer> o2 = o1.map(i -> 12/i)
      .onErrorReturn(t -> 0);

它打印:

RxTest - Subscriber value 12
RxTest - Subscriber value 6
RxTest - Subscriber value 0

不打印其余的 12/3 和 12/4

唯一的解决方案似乎是在 map 函数中中继:

Observable<Integer> o2 = o1.map(i -> {
  try {
    return Optional.of(12/i);
  } catch (ArithmeticException e) {
    return Optional.empty();
  }
}).filter(Optional::isPresent)
  .map(o -> (Integer) o.get());

它可以工作，但是很麻烦。我想知道在操作Observable(例如map)

时是否有任何方法可以轻松跳过任何 RuntimeException

上面是关于在 Observable 中跳过异常。以下是关于在 Subscriber 中跳过异常:

情况是一样的:

List<Integer> ints = Lists.newArrayList(1, 2, 0 , 3 , 4);
Observable<Integer> o1 = Observable.from(ints);
o1.subscribe(
  i -> logger.info("12 / {} = {}", i, 12 / i),
  t -> logger.error("{} : {}", t.getClass() , t.getMessage()),
  () -> logger.info("onCompleted")
);

打印出来:

RxTest - 12 / 1 = 12
RxTest - 12 / 2 = 6
RxTest - class java.lang.ArithmeticException : / by zero

当 onNext 发生异常时，它会触发 onError ，并且不会响应来自 Observable 的任何数据。如果我希望订阅者吞下异常，我必须 try catch onNext() 中的 ArithmeticException。有没有更清洁的解决方案？

似乎当 Subscriber 在 onNext() 中遇到无法在 (onNext) 中处理的错误时，它应该停止，正确的 ？这是一个好的设计吗？

Answer 1

Observable.just(1, 2, 3, 0, 4)
            .flatMap(i -> Observable.defer(() -> Observable.just(12 / i))
                    .onErrorResumeNext(Observable.just(0)));

这是一种方法，但请记住，RxJava 假定错误是真正无法预料的(您可以期望值为 0)。另一方面，如果您想忽略除以 0 的异常，也许您应该在除法之前过滤/映射您的值。