Scala等待未来的序列

Question

我希望像下面这样的代码可以等待这两种未来，但是没有。

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

我以为< code>seq.onComplete会在完成自身之前等待它们全部完成，但事实并非如此；它会导致:

true
false

在scala.concurrent.Future的源代码中有点难以遵循，我想知道如何实现等待（动态大小的）序列的所有原始未来的并行，或者这里可能有什么问题。

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

Answer 1

这是一个支持前面答案的示例。有一种简单的方法可以做到这一点，只需使用标准的Scala API。

在这个例子中，我正在创建3个期货。这些操作将分别在 5 秒、7 秒和 9 秒完成。对 Await.result 的调用将被阻止，直到所有期货都得到解决。一旦所有3个期货完成，a将被设置为List（5，7，9），执行将继续。

此外，如果在任何期货中引发异常，等待。结果将立即取消阻止并引发异常。取消注释＜code＞异常（…）

  try {
    val a = Await.result(Future.sequence(Seq(
      Future({
        blocking {
          Thread.sleep(5000)
        }
        System.err.println("A")
        5
      }),
      Future({
        blocking {
          Thread.sleep(7000)
        }
        System.err.println("B")
        7
        //throw new Exception("Ha!")
      }),
      Future({
        blocking {
          Thread.sleep(9000)
        }
        System.err.println("C")
        9
      }))),
      Duration("100 sec"))

    System.err.println(a)
  } catch {
    case e: Exception ⇒
      e.printStackTrace()
  }

Answer 2

由< code>Future.sequence生成的< code>Future在以下任一情况下完成:

• 所有期货均已成功完成，或
• 其中一个期货失败了

第二点是在您的案例中发生的情况，在包装的未来中的一个失败后立即完成是有意义的，因为在失败的案例中，包装的未来只能保存一个可丢弃的。等待其他期货是没有意义的，因为结果将是同样的失败。

Answer 3

等待所有结果（失败与否）的一种常见方法是在未来将失败“提升”到一个新的表示中，这样所有的未来都会以某种结果完成（尽管它们可能以表示失败的结果完成）。一种自然的方法是提升到＜code＞尝试＜code＞。

Twitter的 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）将在每个未来完成时完成，并将使用Try表示成功和失败。

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

  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
  }