不要在错误时停止 Observable答案

【问题标题】：Do not stop Observable on error不要在错误时停止 Observable
【发布时间】：2015-10-25 09:07:00
【问题描述】：

这是我目前拥有的 Scala 代码：

val b = Observable.interval(1 秒).map(n => if (n % 2 == 1) throw new Exception else n*n)

b.subscribe(n => println(n), e => println("error"), () => println("完成"))

这是我的输出：

0
错误

如何修改我的 Observable 以便在每个错误之后继续运行，并且我将得到如下输出：

0
error
2
error
4
...

【问题讨论】：

由于 Scala 标准库中没有名为 Observable 的内容，因此您应该说出您正在使用哪些库。
@RandallSchulz 我正在使用“RxJava”github.com/Netflix/RxJava
你能在地图函数中捕捉到异常吗？

标签： scala reactive-programming observable rx-java

【解决方案1】：

您可以使用各种错误处理程序之一。我认为在你的情况下 onErrorFlatMap 可能是正确的选择：

Wiki

JavaDoc

不幸的是，onErrorFlatMap 不是（从 0.19.0 版开始）scala api 的一部分。

【讨论】：

【解决方案2】：

我也有同样的问题，并且对 Scala Rx 没有 onErrorFlatMap 感到失望。所以我刺了自己实现这个功能。

我的解决方案如下所示（参见解决方案）。关键方法是这个：

  def recover[T](target: Observable[T]): Observable[Try[T]] = {
      target.map { Success(_) }.
        onErrorResumeNext(
          (err: Throwable) => Observable.just(Failure(err)) ++ recover(target)
        )
  }

'recover' 方法的详细信息

'recover' 的第一个参数是你想要继续挤压事件的 observable，即使在它抛出一个异常。我尝试了各种其他方法，但这是唯一有效的方法我。我最初期望 Scala Rx 的 onErrorReturn 将任何错误映射到我的恢复所指定的值函数，然后继续，但我错过了“可观察合同”的全部要点，那就是 Observable 需要在 onCompleted 或 OnError 之后停止发送任何进一步的事件。任何可观察到的错误后继续喷涌而出的事件将被贴上“病态”的标签（并被上流社会适当回避），正如这里所讨论的：https://github.com/ReactiveX/RxJava/wiki/Phantom-Operators#onerrorflatmap

成功的 onNext 调用的有效负载被包装在 Success() 中，而任何异常都将被处理通过 onErrorResumeNext，它将从 (1) 一个 Observable 包装创建串联的 Observable 流错误，以及 (2) 包含在recovery 递归调用中的目标实例。我最初担心无限递归..但一切都很好。

限制

我应该提到，在原始发帖人的问题的情况下——它使用 Observable.interval，这个不能很好地工作，因为恢复（目标）将是原始的 Observable.interval，它将开始从一开始就发射，所以你永远不会进步。对于间隔之类的东西，你会必须编写自己的基于计时器的间隔，可以重新启动。异常值有望为您提供足够的信息来告诉您需要从哪个值重新启动。

A SOLUTION




object RecoverFromExceptionsInObservable extends App {

  import rx.lang.scala._
  import scala.language.postfixOps
  import scala.util.{Try, Failure, Success}


  val MILLISECS = 500L
  var tickCount = 0

  /**
   * There is a bug in this code which we will ignore for the simple purpose of
   * this test. The bug is that timers are never stopped and cleaned up.
   */
  def getTickObservable(): Observable[Int] = {
    @volatile var subscribers: Set[Observer[Int]] = Set.empty

    val t = new java.util.Timer()
    val task = new java.util.TimerTask {
      def run() = {
        subscribers.foreach(s => s.onNext(tickCount))
        tickCount += 1
      }
    }
    t.schedule(task, 0L, MILLISECS)

    Observable.create[Int] {
      (obs: Observer[Int]) => {
        subscribers = subscribers + obs
        Subscription {
          subscribers = subscribers - obs
        }
      }
    }
  }

  def recover[T](target: Observable[T]): Observable[Try[T]] = {
      target.map { Success(_) }.
        onErrorResumeNext(
          (err: Throwable) => Observable.just(Failure(err)) ++ recover(target)
        )
  }


  val stream1 = getTickObservable() map { item =>
    if (item % 2 == 0) throw new RuntimeException(s"error on $item") else item
  }

  recover(stream1).subscribe(
    term => {
      println(s" ${Thread.currentThread().getName()}  onNext: $term")
    },
    t => {
      println("in error callback")
      println(s" ${Thread.currentThread().getName()}  onError: $t")
    },
    () => println(s" ${Thread.currentThread().getName()} subscriber complete")
  )
}

这是运行上述代码的部分输出：

 Timer-0  onNext: Success(1)
 Timer-0  onNext: Failure(java.lang.RuntimeException: error on 2)
 Timer-0  onNext: Success(3)
 Timer-0  onNext: Failure(java.lang.RuntimeException: error on 4)
 Timer-0  onNext: Success(5)
 Timer-0  onNext: Failure(java.lang.RuntimeException: error on 6)

我不希望这个答案永远持续下去，所以我跳过了一些关于我解决这个问题的替代方法的细节，如果你有兴趣，可以阅读here。

【讨论】：