【问题标题】：Concatenate many Future[Seq] into one Future[Seq]将多个 Future[Seq] 连接成一个 Future[Seq]
【发布时间】：2017-11-19 00:49:09
【问题描述】：

没有 Future，这就是我如何将所有较小的 Seq 组合成一个带有 flatmap

的大 Seq

category.getCategoryUrlKey(id: Int):Seq[Meta] // main method
val appDomains: Seq[Int]

val categories:Seq[Meta] = appDomains.flatMap(category.getCategoryUrlKey(_))

现在getCategoryUrlKey 方法可能会失败。我在前面放了一个断路器，以避免在 maxFailures 之后为下一个元素调用它。现在断路器不会返回Seq，而是返回Future[Seq]

lazy val breaker = new akka.pattern.CircuitBreaker(...)

private def getMeta(appDomainId: Int): Future[Seq[Meta]] = {
  breaker.withCircuitBreaker {
    category.getCategoryUrlKey(appDomainId)
  }
}

如何遍历 List appDomains 并将结果合并为一个 Future[Seq] ，可能合并为 Seq ？

如果函数式编程适用，有没有办法不用临时变量直接转换？

【问题讨论】：

标签： scala future seq circuit-breaker

【解决方案1】：

使用 Future.sequence 压缩期货序列

Future.sequenceconverts Seq[Future[T]] 到 Future[Seq[T]]

在您的情况下，T 是 Seq。在序列操作之后，您将得到 Seq[Seq[T]]。所以只需在序列操作后使用 flatten 将其展平即可。

def squashFutures[T](list: Seq[Future[Seq[T]]]): Future[Seq[T]] =
  Future.sequence(list).map(_.flatten)

你的代码变成了

Future.sequence(appDomains.map(getMeta)).map(_.flatten)

【讨论】：

好的，我必须先存储到 Seq[Future[Seq[T]]] ，然后我可以 squashFutures 。有没有办法在没有临时变量的情况下直接转换？
@RaymondChenon Future.sequence(appDomains.map(getMeta)).map(_.flatten) 是您必须做的。这里没有临时变量
你是英雄。

【解决方案2】：

从 TraversableOnce[Future[A]] 到 Future[TraversableOnce[A]]

val categories = Future.successful(appDomains).flatMap(seq => {
    val fs = seq.map(i => getMeta(i))
    val sequenced = Future.sequence(fs)
    sequenced.map(_.flatten)
})

Future.successful(appDomains) 将 appDomains 提升到 Future 的上下文中

希望这会有所帮助。

【讨论】：

【解决方案3】：

val metaSeqFutureSeq = appDomains.map(i => getMeta(i))
// Seq[Future[Seq[Meta]]]

val metaSeqSeqFuture = Future.sequence(metaSeqFutureSeq)
// Future[Seq[Seq[Meta]]]
// NOTE :: this future will fail if any of the futures in the sequence fails

val metaSeqFuture = metaSeqSeqFuture.map(seq => seq.flatten)
// Future[Seq[Meta]]

如果您想拒绝唯一失败的未来但保留成功的未来，那么我们将不得不有点创意并使用承诺来构建我们的未来。

import java.util.concurrent.locks.ReentrantLock

import scala.collection.mutable.ArrayBuffer
import scala.concurrent.{Future, Promise}
import scala.util.{Failure, Success}

def futureSeqToOptionSeqFuture[T](futureSeq: Seq[Future[T]]): Future[Seq[Option[T]]] = {
  val promise = Promise[Seq[Option[T]]]()

  var remaining = futureSeq.length

  val result = ArrayBuffer[Option[T]]()
  result ++ futureSeq.map(_ => None)

  val resultLock = new ReentrantLock()

  def handleFutureResult(option: Option[T], index: Int): Unit = {
    resultLock.lock()
    result(index) = option
    remaining = remaining - 1
    if (remaining == 0) {
      promise.success(result)
    }
    resultLock.unlock()
  }

  futureSeq.zipWithIndex.foreach({ case (future, index) => future.onComplete({
    case Success(t) => handleFutureResult(Some(t), index)
    case Failure(ex) => handleFutureResult(None, index)
  }) })

  promise.future
}

val metaSeqFutureSeq = appDomains.map(i => getMeta(i))
// Seq[Future[Seq[Meta]]]

val metaSeqOptionSeqFuture = futureSeqToOptionSeqFuture(metaSeqFutureSeq)
// Future[Seq[Option[Seq[Meta]]]]

val metaSeqFuture = metaSeqSeqFuture.map(seq => seq.flatten.flatten)
// Future[Seq[Meta]]

【讨论】：