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 

但是如果我想让观察者（o2）跳过异常呢？

我看着RxJava的文档error handling，没有办法跳过这个错误。 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()); 

它的工作，但它很麻烦。我不知道是否操纵Observable（如map）

以上是关于Observable跳过异常时，有什么办法可以轻松地跳过任何RuntimeException。以下是有关在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") 
); 

它打印出：

当例外onNext发生时，它会触发onError和NOT响应Observable的任何数据。如果我希望订户吞下异常，我必须尝试捕捉onNext()中的ArithmeticException。有没有更清洁的解决方案？

看来当Subscriber在onNext()面临错误，在（onNext）内无法处理时，它应该停止，对吧？这是一个好的设计吗？

Answer 1

试试这个方法。

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

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

Answer 2

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

它是一种方法，但请记住，RxJava假定错误是真正没有得到解决的（您可以预期值为0）。另一方面，如果你想忽略除以0的异常，也许你应该在分割之前过滤/映射你的值。

Answer 3

import rx.lang.scala._ 

import scala.concurrent.duration.DurationDouble 
import scala.util.{Failure, Success, Try} 

val o1 = List(1, 2, 3, 0, 4) toObservable 
val o2 = List(1, 2, 3, 4, 5) toObservable 

val o3 = o1 zip o2 map {case (i1, i2) => i2/i1 } // this will trigger a division by 0 error 

val o4 = o3 lift { 
     subscriber: Subscriber[Try[Int]] => new Subscriber[Int](subscriber) { 
     override def onNext(v: Int) { 
      if (!subscriber.isUnsubscribed) 
      subscriber.onNext(Success(v)) 
     } 

     override def onError(e: Throwable) { 
      if (!subscriber.isUnsubscribed) 
      subscriber.onNext(Failure(e)) 
     } 

     override def onCompleted() { 
      if (!subscriber.isUnsubscribed) 
      subscriber.onCompleted() 
     } 
     } 
    } 

o4 subscribe(println(_))