期货/成功竞赛

Question

我正在学习期货,我正在尝试创建一种方法,将两个期货作为参数(f和g)并返回成功完成的第一个未来,否则它返回f或g.

一些用例来说明我的方法的行为是:

Future 1        | Future 2         | Result
Success First     Success Second     Future 1
Success First     Failure Second     Future 1
Success Second    Success First      Future 2
Success Second    Failure First      Future 1
Failure First     Failure Second     Future 2 (because we had a failure on Future 1, so try to see what is the result Future 2)

所以我创建了这个方法:

def successRace(f: Future[T], g: Future[T]): Future[T] = {
        val p1 = Promise[T]()
        val p2 = Promise[T]()
        val p3 = Promise[T]()
        p1.completeWith(f)
        p2.completeWith(g)
        p3. ????
        p3.future
}

而现在,我怎么知道哪一个先完成了？

Answer 1

用例是第一次成功完成：

scala> :pa
// Entering paste mode (ctrl-D to finish)

def firstSuccessOf[T](fs: Future[T]*)(implicit x: ExecutionContext): Future[T] = {
  val p = Promise[T]()
  val count = new java.util.concurrent.atomic.AtomicInteger(fs.size)
  def bad() = if (count.decrementAndGet == 0) { p tryComplete new Failure(new RuntimeException("All bad")) }
  val completeFirst: Try[T] => Unit = p tryComplete _
  fs foreach { _ onComplete { case v @ Success(_) => completeFirst(v) case _ => bad() }}
  p.future
}

// Exiting paste mode, now interpreting.

firstSuccessOf: [T](fs: scala.concurrent.Future[T]*)(implicit x: scala.concurrent.ExecutionContext)scala.concurrent.Future[T]

所以

scala> def f = Future { Thread sleep 5000L ; println("Failing") ; throw new NullPointerException }
f: scala.concurrent.Future[Nothing]

scala> def g = Future { Thread sleep 10000L ; println("OK") ; 7 }
g: scala.concurrent.Future[Int]

scala> firstSuccessOf(f,g)
res3: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@5ed53f6b

scala> res0Failing
          3.value
res4: Option[scala.util.Try[Int]] = None

scala> res3.valueOK

res5: Option[scala.util.Try[Int]] = Some(Success(7))

或者

scala> def h = Future { Thread sleep 7000L ; println("Failing too") ; throw new NullPointerException }
h: scala.concurrent.Future[Nothing]


scala> firstSuccessOf(f,h)
res10: scala.concurrent.Future[Nothing] = scala.concurrent.impl.Promise$DefaultPromise@318d30be

scala> 

scala> res10.Failing
value
res11: Option[scala.util.Try[Nothing]] = None

scala> Failing too


scala> res10.value
res12: Option[scala.util.Try[Nothing]] = Some(Failure(java.lang.RuntimeException: All bad))

@ ysusuk 的答案是Future.firstCompletedOf引擎盖下的作用。

Answer 2

你想使用该tryCompleteWith方法。它可以被多次调用，并且只有第一个完成未来的人获胜。

def successRace(f: Future[T], g: Future[T]): Future[T] = {
  val p = Promise[T]()
  p.tryCompleteWith(f)
  p.tryCompleteWith(g)
  p.future
}