未来/成功竞赛

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
}

Answer 3

我完全同意前面的答案，但我希望我的例子能进一步澄清它，所以：

def successRace[T](f: Future[T], g: Future[T]): Future[T] = {
  val promise = Promise[T]()

  f onComplete(promise.tryComplete(_))
  g onComplete(promise.tryComplete(_))

  promise.future
}

因此，第一个完成的Future将设置包装在Try中的值(so、Success或Failure)。