从期货可观察到-来自多个线程的onNext

thund 发表于 Dev

thund

我想根据Observable列表的结果实时生成一个Futures。

在最简单的情况下，假设我有一份正在运行的期货清单Future.sequence，而我只是使用监视Observable每次交易完成时的进度来监视它们的进度。我基本上是这样的：

  def observeFuturesProgress(futs: List[Future[Int]]) : Observable[String] = {
    Observable[String](observer => {
      val loudFutures: List[Future[Int]] = futs.map(f => {
            f onComplete {
              case Success(a) => observer.onNext(s"just did $a more")
              case Failure(e) => observer.onError(e)
            }
            f
        })
      Future.sequence(loudFutures) onComplete {
          case Success(_) => observer.onCompleted()
          case Failure(e) => observer.onError(e)
      }
    })
  }

在我的测试环境中，这可以正常工作。但是我已经读过，onNext不应从不同的线程调用它，至少要注意没有重叠的调用。建议的解决方法是什么？似乎许多实际用途Observables都需要onNext从这样的异步代码中调用，但我在文档中找不到类似的示例。

我的狗汤姆

可观察合约

观察者必须按顺序（而不是并行）向观察者发出通知。他们可以从不同的线程发出这些通知，但是在通知之间必须存在正式的事前关系。

看看序列化

一个Observable可能异步地（可能从不同的线程）调用其观察者的方法。这可能使这样的Observable违反Observable约定，因为它可能尝试在其OnNext通知之一之前发送OnCompleted或OnError通知，或者可能同时从两个不同的线程发出OnNext通知。通过将Serialize运算符应用于此Observable，可以强制使其行为良好且同步。

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。