RxJava Observable和Subscriber用于跳过异常？

Question

如果我有一个Observalbe:

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

我想生成另一个可观察的,除以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关于错误处理的文档,没有办法跳过错误.在onErrorResumeNext()和onExceptionResumeNext()需要备份/回退 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());

它有效,但很麻烦.我想知道RuntimeException在操作时是否有任何方法可以轻松跳过Observable(例如map)

以上是关于跳过异常的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.如果我想用户吞下例外,我不得不尝试,赶上ArithmeticException的onNext().有没有更清洁的解决方案？

看来,当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异常,你可能应该在除法之前过滤/映射你的值.

Answer 2

试试这个方法。

Observable<Integer> o2 = o1.map(i -> 12/i).onErrorFlatMap(e -> Observable.empty());

在上面的代码中，错误事件被空流替换，并且原始流正在恢复。因此，错误将被跳过。