如何管理一组 Akka 期货答案

【问题标题】：How to manage a set of Akka futures如何管理一组 Akka 期货
【发布时间】：2016-09-06 02:50:49
【问题描述】：

我有一组 Future[T]，我想将其管理为我正在编写的库的单个对象。在我当前的实现中，我使用 Future.sequence 来收集它们并等待它们解决，这样我就可以对它们做未来的事情（地图、收集、过滤）。但是，这只使我能够匹配成功或失败，而我正在处理的期货集合不一定是这种情况。有些会失败，有些会成功，我希望能够从成功的那些中提取我可以的值并收集其他的异常和错误，以便我可以适当地升级它们。在伪代码中，它会类似于

Future.sequence(Set[Future[T]]) andThen {
  case FullSuccess => "woot"
  case SomeErrors  => "well, that's still ok."
  case FullErrors  => "Ok, who's the wise guy."
}

我真正想要的是在有数据的地方有数据，如果序列中只有 1 个期货失败，则不必返回完全失败。

感谢您的帮助。

【问题讨论】：

标签： scala functional-programming future

【解决方案1】：

很遗憾，您的案例没有内置帮助器，但创建自己的帮助器很容易：

import scala.concurrent.{Await, Future}
import scala.util.{Failure, Success, Try}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.DurationInt

def sequenceOfTries[T](futures: Seq[Future[T]]): Future[Seq[Try[T]]] =
  futures.foldLeft(Future.successful(List[Try[T]]())) {
    case (accF, f) => accF.flatMap {
      acc => f.map(v => Success(v) :: acc).recover { case ex => Failure(ex) :: acc }
    }
  }.map(_.reverse)


val v = Seq(
  Future.successful(1),
  Future.failed(new IllegalStateException("2")),
  Future.successful(3),
  Future.failed(new IllegalStateException("4"))
)

Await.result(sequenceOfTries(v), 1.second)

结果：

v: Seq[scala.concurrent.Future[Int]] = List(scala.concurrent.impl.Promise$KeptPromise@2416f7e5, scala.concurrent.impl.Promise$KeptPromise@2aaf675d, scala.concurrent.impl.Promise$KeptPromise@360d48f, scala.concurrent.impl.Promise$KeptPromise@230f8be2)    

res0: Seq[scala.util.Try[Int]] = List(Success(1), Failure(java.lang.IllegalStateException: 2), Success(3), Failure(java.lang.IllegalStateException: 4))

UPD。或者，您可以像这样使用Future.sequence（结果相同）：

def sequenceOfTries[T](futures: Seq[Future[T]]): Future[Seq[Try[T]]] =
  Future.sequence(futures.map(_.map(x => Success(x)).recover { case ex => Failure(ex) }))

【讨论】：

将可能失败的 Future[T] 转换为保证成功的 Future[Try[T]] 很常见，足以保证它自己的辅助方法，恕我直言。