如何等待几个期货

假设我有几个未来，需要等到他们中的任何一个失败或者全部成功。

例如：假设有3个期货： f1 ， f2 ， f3 。

• 如果f1成功， f2失败，我不等f3 （并返回失败的客户端）。

• 如果f2在f1和f3仍在运行时失败，我不等它们（并返回失败 ）

• 如果f1成功，然后f2成功，我继续等待f3 。

你将如何实现它？

Answer 1

你可以使用如下所示的理解：

 val fut1 = Future{...} val fut2 = Future{...} val fut3 = Future{...} val aggFut = for{ f1Result <- fut1 f2Result <- fut2 f3Result <- fut3 } yield (f1Result, f2Result, f3Result) 

在这个例子中，期货1,2和3是并行的。 然后，在理解，我们等待，直到结果1，然后2，然后3可用。 如果1或2失败，我们不会再等3了。 如果所有3个都成功了，那么aggFut val将会持有一个3槽的元组，对应于3期货的结果。

现在，如果你需要你想停止等待的行为，如果说fut2先失败，事情会变得有点棘手。 在上面的例子中，你必须等待fut1完成才能实现fut2失败。 要解决这个问题，你可以尝试这样的事情：

  val fut1 = Future{Thread.sleep(3000);1} val fut2 = Promise.failed(new RuntimeException("boo")).future val fut3 = Future{Thread.sleep(1000);3} def processFutures(futures:Map[Int,Future[Int]], values:List[Any], prom:Promise[List[Any]]):Future[List[Any]] = { val fut = if (futures.size == 1) futures.head._2 else Future.firstCompletedOf(futures.values) fut onComplete{ case Success(value) if (futures.size == 1)=> prom.success(value :: values) case Success(value) => processFutures(futures - value, value :: values, prom) case Failure(ex) => prom.failure(ex) } prom.future } val aggFut = processFutures(Map(1 -> fut1, 2 -> fut2, 3 -> fut3), List(), Promise[List[Any]]()) aggFut onComplete{ case value => println(value) } 

现在这个工作是正确的，但是问题来自于知道哪一个Future从成功完成的Map移除。 只要你有一些方法来正确地将结果与生成结果的Future相关联，那么类似这样的结果就行得通。 它只是recursion地将完成的期货从地图上移走，然后在剩余的Futures上调用Future.firstCompletedOf ，直到没有剩下的Futures ，一路收集结果。 这不是很好，但如果你真的需要你正在谈论的行为，那么这个或类似的东西可以工作。

Answer 2

您可以使用承诺，并发送给第一个失败或最终完成的汇总成功：

 def sequenceOrBailOut[A, M[_] <: TraversableOnce[_]](in: M[Future[A]] with TraversableOnce[Future[A]])(implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], executor: ExecutionContext): Future[M[A]] = { val p = Promise[M[A]]() // the first Future to fail completes the promise in.foreach(_.onFailure{case i => p.tryFailure(i)}) // if the whole sequence succeeds (ie no failures) // then the promise is completed with the aggregated success Future.sequence(in).foreach(p trySuccess _) p.future } 

那么，如果你想阻止，或者只是把它map到别的东西，那么你可以Await那个结果的Future 。

与理解的区别在于，在这里你得到了第一个失败的错误，而在理解的时候，你得到了input集合的遍历顺序的第一个错误（即使另一个失败了）。 例如：

 val f1 = Future { Thread.sleep(1000) ; 5 / 0 } val f2 = Future { 5 } val f3 = Future { None.get } Future.sequence(List(f1,f2,f3)).onFailure{case i => println(i)} // this waits one second, then prints "java.lang.ArithmeticException: / by zero" // the first to fail in traversal order 

和：

 val f1 = Future { Thread.sleep(1000) ; 5 / 0 } val f2 = Future { 5 } val f3 = Future { None.get } sequenceOrBailOut(List(f1,f2,f3)).onFailure{case i => println(i)} // this immediately prints "java.util.NoSuchElementException: None.get" // the 'actual' first to fail (usually...) // and it returns early (it does not wait 1 sec) 

Answer 3

这是一个不使用演员的解决scheme。

 import scala.util._ import scala.concurrent._ import java.util.concurrent.atomic.AtomicInteger // Nondeterministic. // If any failure, return it immediately, else return the final success. def allSucceed[T](fs: Future[T]*): Future[T] = { val remaining = new AtomicInteger(fs.length) val p = promise[T] fs foreach { _ onComplete { case s @ Success(_) => { if (remaining.decrementAndGet() == 0) { // Arbitrarily return the final success p tryComplete s } } case f @ Failure(_) => { p tryComplete f } } } p.future } 

Answer 4

你可以单独用期货来做到这一点。 这是一个实现。 请注意，它不会提前终止执行！ 在这种情况下，你需要做一些更复杂的事情（也许自己实现中断）。 但是如果你只是不想继续等待某些不可行的事情，那么关键是要等待第一件事情的完成，当没有任何事情发生或者你遇到exception时停止：

 import scala.annotation.tailrec import scala.util.{Try, Success, Failure} import scala.concurrent._ import scala.concurrent.duration.Duration import ExecutionContext.Implicits.global @tailrec def awaitSuccess[A](fs: Seq[Future[A]], done: Seq[A] = Seq()): Either[Throwable, Seq[A]] = { val first = Future.firstCompletedOf(fs) Await.ready(first, Duration.Inf).value match { case None => awaitSuccess(fs, done) // Shouldn't happen! case Some(Failure(e)) => Left(e) case Some(Success(_)) => val (complete, running) = fs.partition(_.isCompleted) val answers = complete.flatMap(_.value) answers.find(_.isFailure) match { case Some(Failure(e)) => Left(e) case _ => if (running.length > 0) awaitSuccess(running, answers.map(_.get) ++: done) else Right( answers.map(_.get) ++: done ) } } } 

下面是一切正常的时候的一个例子：

 scala> awaitSuccess(Seq(Future{ println("Hi!") }, Future{ Thread.sleep(1000); println("Fancy meeting you here!") }, Future{ Thread.sleep(2000); println("Bye!") } )) Hi! Fancy meeting you here! Bye! res1: Either[Throwable,Seq[Unit]] = Right(List((), (), ())) 

但是当出现问题时：

 scala> awaitSuccess(Seq(Future{ println("Hi!") }, Future{ Thread.sleep(1000); throw new Exception("boo"); () }, Future{ Thread.sleep(2000); println("Bye!") } )) Hi! res2: Either[Throwable,Seq[Unit]] = Left(java.lang.Exception: boo) scala> Bye! 

Answer 5

这个问题已经回答了，但我发布我的价值类解决scheme（价值类添加2.10），因为这里没有一个。 请随意批评。

  implicit class Sugar_PimpMyFuture[T](val self: Future[T]) extends AnyVal { def concurrently = ConcurrentFuture(self) } case class ConcurrentFuture[A](future: Future[A]) extends AnyVal { def map[B](f: Future[A] => Future[B]) : ConcurrentFuture[B] = ConcurrentFuture(f(future)) def flatMap[B](f: Future[A] => ConcurrentFuture[B]) : ConcurrentFuture[B] = concurrentFutureFlatMap(this, f) // work around no nested class in value class } def concurrentFutureFlatMap[A,B](outer: ConcurrentFuture[A], f: Future[A] => ConcurrentFuture[B]) : ConcurrentFuture[B] = { val p = Promise[B]() val inner = f(outer.future) inner.future onFailure { case t => p.tryFailure(t) } outer.future onFailure { case t => p.tryFailure(t) } inner.future onSuccess { case b => p.trySuccess(b) } ConcurrentFuture(p.future) } 

ConcurrentFuture是一个没有开销的Future包装器，它将默认的Future map / flatMap从do-this-then-then改为combine-all-and-fail-if-any-fail。 用法：

 def func1 : Future[Int] = Future { println("f1!");throw new RuntimeException; 1 } def func2 : Future[String] = Future { Thread.sleep(2000);println("f2!");"f2" } def func3 : Future[Double] = Future { Thread.sleep(2000);println("f3!");42.0 } val f : Future[(Int,String,Double)] = { for { f1 <- func1.concurrently f2 <- func2.concurrently f3 <- func3.concurrently } yield for { v1 <- f1 v2 <- f2 v3 <- f3 } yield (v1,v2,v3) }.future f.onFailure { case t => println("future failed $t") } 

在上面的例子中，f1，f2和f3将同时运行，如果任何顺序失败，元组的未来将立即失败。

Answer 6

你可以使用这个：

 val l = List(1, 6, 8) val f = l.map{ i => future { println("future " +i) Thread.sleep(i* 1000) if (i == 12) throw new Exception("6 is not legal.") i } } val f1 = Future.sequence(f) f1 onSuccess{ case l => { logInfo("onSuccess") l.foreach(i => { logInfo("h : " + i) }) } } f1 onFailure{ case l => { logInfo("onFailure") } 

Answer 7

为此，我会使用阿卡演员。 与理解不同的是，一旦任何一个期货失败，它就会失败，所以从这个意义上来说效率更高一些。

 class ResultCombiner(futs: Future[_]*) extends Actor { var origSender: ActorRef = null var futsRemaining: Set[Future[_]] = futs.toSet override def receive = { case () => origSender = sender for(f <- futs) f.onComplete(result => self ! if(result.isSuccess) f else false) case false => origSender ! SomethingFailed case f: Future[_] => futsRemaining -= f if(futsRemaining.isEmpty) origSender ! EverythingSucceeded } } sealed trait Result case object SomethingFailed extends Result case object EverythingSucceeded extends Result 

然后，创build演员，发送一条消息给它（这样它就会知道将答复发送到哪里）并等待答复。

 val actor = actorSystem.actorOf(Props(new ResultCombiner(f1, f2, f3))) try { val f4: Future[Result] = actor ? () implicit val timeout = new Timeout(30 seconds) // or whatever Await.result(f4, timeout.duration).asInstanceOf[Result] match { case SomethingFailed => println("Oh noes!") case EverythingSucceeded => println("It all worked!") } } finally { // Avoid memory leaks: destroy the actor actor ! PoisonPill } 

Answer 8

你可能想结账Twitter的未来API。 值得注意的是Future.collect方法。 它正是你想要的： https ： //twitter.github.io/scala_school/finagle.html

源代码Future.scala可以在这里find： https ： //github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Future.scala