Scala 等待 future 序列

Question

我希望像下面这样的代码会等待两个 future，但事实并非如此。

object Fiddle {
  val f1 = Future {
    throw new Throwable("baaa") // emulating a future that bumped into an exception
  }

  val f2 = Future {
    Thread.sleep(3000L) // emulating a future that takes a bit longer to complete
    2
  }

  val lf = List(f1, f2) // in the general case, this would be a dynamically sized list

  val seq = Future.sequence(lf) 

  seq.onComplete {
    _ => lf.foreach(f => println(f.isCompleted))
  }
}

val a = FuturesSequence

我假设 seq.onComplete 会等待它们全部完成后再自行完成，但事实并非如此；结果是:

true
false

.sequence 在 scala.concurrent.Future 的源代码中有点难以理解，我想知道如何实现等待(动态大小的)序列的所有原始 future 的并行，或者这里可能存在什么问题。

编辑:相关问题:https://worldbuilding.stackexchange.com/questions/12348/how-do-you-prove-youre-from-the-future :)

Answer 1

等待所有结果(失败与否)的一种常见方法是将失败“提升”到 future 的新表示中，以便所有 future 都以某种结果完成(尽管它们可能以代表失败的结果完成) 。实现这一目标的一种自然方法是提升为 Try。

Twitter's implementation of futures提供了一个 liftToTry 方法，使这变得微不足道，但您可以使用标准库的实现执行类似的操作:

import scala.util.{ Failure, Success, Try }

val lifted: List[Future[Try[Int]]] = List(f1, f2).map(
  _.map(Success(_)).recover { case t => Failure(t) }
)

现在，Future.sequence(lifted) 将在每个 future 完成时完成，并使用 Try 表示成功和失败。

因此，等待一系列 future 的所有原始 future 的通用解决方案可能如下所示，假设执行上下文当然是隐式可用的。

  import scala.util.{ Failure, Success, Try }

  private def lift[T](futures: Seq[Future[T]]) = 
    futures.map(_.map { Success(_) }.recover { case t => Failure(t) })

  def waitAll[T](futures: Seq[Future[T]]) =
    Future.sequence(lift(futures)) // having neutralized exception completions through the lifting, .sequence can now be used

  waitAll(SeqOfFutures).map { 
    // do whatever with the completed futures
  }