16

我必须并行运行多个期货，并且程序不应该崩溃或挂起。

现在我一一等待期货，如果有 TimeoutException，则使用回退值。

val future1 = // start future1
val future2 = // start future2
val future3 = // start future3

// <- at this point all 3 futures are running

// waits for maximum of timeout1 seconds
val res1 = toFallback(future1, timeout1, Map[String, Int]())
// .. timeout2 seconds 
val res2 = toFallback(future2, timeout2, List[Int]())
// ... timeout3 seconds
val res3 = toFallback(future3, timeout3, Map[String, BigInt]()) 

def toFallback[T](f: Future[T], to: Int, default: T) = {
  Try(Await.result(f, to seconds))
    .recover { case to: TimeoutException => default }
}

正如我所看到的，这个片段的最大等待时间是timeout1 + timeout2 + timeout3

我的问题是：我怎样才能一次等待所有这些期货，所以我可以减少等待时间max(timeout1, timeout2, timeout3)？

编辑：最后我使用了@Jatin 和@senia 答案的修改：

private def composeWaitingFuture[T](fut: Future[T], 
                                    timeout: Int, default: T) =
  future { Await.result(fut, timeout seconds) } recover {
    case e: Exception => default
  }

后来它的使用如下：

// starts futures immediately and waits for maximum of timeoutX seconds
val res1 = composeWaitingFuture(future1, timeout1, Map[String, Int]())
val res2 = composeWaitingFuture(future2, timeout2, List[Int]())
val res3 = composeWaitingFuture(future3, timeout3, Map[String, BigInt]()) 

// takes the maximum of max(timeout1, timeout2, timeout3) to complete
val combinedFuture =
  for {
    r1 <- res1
    r2 <- res2
    r3 <- res3
  } yield (r1, r2, r3)

后来我combinedFuture按照我认为合适的方式使用。

4

7 回答 7

14

您可以使用或为理解创建future返回所有 3 个期货的结果：flatMap

val combinedFuture =
  for {
    r1 <- future1
    r2 <- future2
    r3 <- future3
  } yield (r1, r2, r3)

val (r1, r2, r3) = Await.result(combinedFuture , Seq(timeout1, timeout2, timeout3).max)

如果您正在使用akka，您可以在超时后使用默认值完成您的未来：

implicit class FutureHelper[T](f: Future[T]) extends AnyVal{
  import akka.pattern.after
  def orDefault(t: Timeout, default: => T)(implicit system: ActorSystem): Future[T] = {
    val delayed = after(t.duration, system.scheduler)(Future.successful(default))
    Future firstCompletedOf Seq(f, delayed)
  }
}

val combinedFuture =
  for {
    r1 <- future1.orDefault(timeout1, Map())
    r2 <- future2.orDefault(timeout2, List())
    r3 <- future3.orDefault(timeout3, Map())
  } yield (r1, r2, r3)

val (r1, r2, r3) = Await.result(combinedFuture , allowance + Seq(timeout1, timeout2, timeout3).max)
于 2013-07-04T10:19:03.083 回答
10 
def toFallback[T](f: Future[T], to: Int, default: T) = {
  future{
  try{
        Await.result(f, to seconds)
   }catch{
        case e:TimeoutException => default
  }
 }

你甚至可以使这个块异步，每个请求等待它的最长时间。如果线程太多，则可能有一个线程使用 Akka 不断检查其他期货system scheduler。@Senia 在下面回答了这个问题。

于 2013-07-04T10:17:48.700 回答
3

我会避免使用Await.result，因为它使用线程只是为了阻塞。为期货实施超时的一种选择是：

val timer = new Timer()

def toFallback[T](f: Future[T], timeout: Int, default: T) = {
  val p = Promise[T]()
  f.onComplete(result => p.tryComplete(result))
  timer.schedule(new TimerTask {
    def run() {
      p.tryComplete(Success(default))
    }
  }, timeout)
  p.future
}

这将创建一个承诺，该承诺将由未来或指定超时后的默认结果完成 - 以先到者为准。

要同时运行查询，您可以这样做：

val future1 = // start future1
val future2 = // start future2
val future3 = // start future3

val res1 = toFallback(future1, timeout1, Map[String, Int]())
val res2 = toFallback(future2, timeout2, List[Int]())
val res3 = toFallback(future3, timeout3, Map[String, BigInt]())

val resultF = for {
  r1 <- res1
  r2 <- res2
  r3 <- res3
} yield (r1, r2, r3)

val (r1, r2, r3) = Await.result(resultF, Duration.Inf)
println(s"$r1, $r2, $r3")

//or
resultF.onSuccess {
  case (r1, r2, r3) => println(s"$r1, $r2, $r3")
}
于 2013-07-04T10:19:43.760 回答
2

这是一个更长（unakka）的答案，它解决了可能的用例，即，如果其中一个值“超时”，您想对该结果使用默认值并对其进行处理（例如取消 long-运行计算或 i/o 或其他）。

不用说，另一个故事是尽量减少阻塞。

基本思想是坐在一个循环中等待firstCompletedOf尚未完成的项目。上的超时ready是最小剩余超时。

此代码使用期限而不是持续时间，但使用持续时间作为“剩余时间”很容易。

import scala.language.postfixOps
import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits._
import scala.reflect._
import scala.util._
import java.lang.System.{ nanoTime => now }

import Test.time

class Test {

  type WorkUnit[A] = (Promise[A], Future[A], Deadline, A)
  type WorkQ[A] = Seq[WorkUnit[A]]

  def await[A: ClassTag](work: Seq[(Future[A], Deadline, A)]): Seq[A] = {
    // check for timeout; if using Duration instead of Deadline, decrement here
    def ticktock(w: WorkUnit[A]): WorkUnit[A] = w match {
      case (p, f, t, v) if !p.isCompleted && t.isOverdue => p trySuccess v ; w
      case _ => w
    }
    def await0(work: WorkQ[A]): WorkQ[A] = {
      val live = work filterNot (_._1.isCompleted)
      val t0 = (live map (_._3)).min
      Console println s"Next deadline in ${t0.timeLeft.toMillis}"
      val f0 = Future firstCompletedOf (live map (_._2))
      Try(Await ready (f0, t0.timeLeft))
      val next = work map (w => ticktock(w))
      if (next exists (!_._1.isCompleted)) {
        await0(next)
      } else {
        next
      }
    }
    val wq = work map (_ match {
      case (f, t, v) =>
        val p = Promise[A]
        p.future onComplete (x => Console println s"Value available: $x: $time")
        f onSuccess {
          case a: A => p trySuccess a  // doesn't match on primitive A
          case x => p trySuccess x.asInstanceOf[A]
        }
        f onFailure { case _ => p trySuccess v }
        (p, f, t, v)
    })
    await0(wq) map (_ match {
      case (p, f, t, v) => p.future.value.get.get
    })
  }
}

object Test {
  val start = now
  def time = s"The time is ${ Duration fromNanos (now - start) toMillis }"

  def main(args: Array[String]): Unit = {
    // #2 times out
    def calc(i: Int) = {
      val t = if (args.nonEmpty && i == 2) 6 else i
      Thread sleep t * 1000L
      Console println s"Calculate $i: $time"
      i
    }
    // futures to be completed by a timeout deadline
    // or else use default and let other work happen
    val work = List(
      (future(calc(1)), 3 seconds fromNow, 10),
      (future(calc(2)), 5 seconds fromNow, 20),
      (future(calc(3)), 7 seconds fromNow, 30)
    )
    Console println new Test().await(work)
  }
}

样品运行：

apm@mara:~/tmp$ skalac nextcompleted.scala ; skala nextcompleted.Test 
Next deadline in 2992
Calculate 1: The time is 1009
Value available: Success(1): The time is 1012
Next deadline in 4005
Calculate 2: The time is 2019
Value available: Success(2): The time is 2020
Next deadline in 4999
Calculate 3: The time is 3020
Value available: Success(3): The time is 3020
List(1, 2, 3)
apm@mara:~/tmp$ skala nextcompleted.Test arg
Next deadline in 2992
Calculate 1: The time is 1009
Value available: Success(1): The time is 1012
Next deadline in 4005
Calculate 3: The time is 3020
Value available: Success(3): The time is 3020
Next deadline in 1998
Value available: Success(20): The time is 5020
List(1, 20, 3)
于 2013-07-05T18:12:06.680 回答
0

这可能有点 hacky，但您可以简单地测量经过的时间并相应地修改超时。假设timeout1 <= timeout2 <= timeout3：

def now     = System.currentTimeMillis();
val start   = now;
def remains(timeout: Long): Long
            = math.max(0, timeout + start - now)

def toFallback[T](f: Future[T], to: Int, default: T) = {
  Try(Await.result(f, remains(to) seconds))
    .recover { case to: TimeoutException => default }
}

这样每次超时都是基于start = now被调用的时刻，所以整体运行时间最多为timeout3. 如果未对超时进行排序，它仍然可以工作，但某些任务的运行时间可能会超过其指定的超时时间。

于 2013-07-04T11:47:07.203 回答
0

为什么不让Future 自己执行默认的异常捕获和返回？然后你可以简单地Await依次对每个future进行处理，而不必担心future之外的异常处理。

于 2013-07-04T10:42:00.947 回答
0

使用Monix Task，它是类固醇的未来。

import monix.execution.Scheduler.Implicits.global
import monix.eval._
import scala.concurrent.duration._

val task1 = Task{Thread.sleep(1);"task1"}.timeoutTo(timeout1,Task.now("timeout1"))
val task2 = Task{Thread.sleep(2);"task2"}.timeoutTo(timeout2,Task.now("timeout2"))
Task.zipList(Seq(task1,task2)).runSyncUnsafe(Duration.Inf)
于 2018-05-16T08:20:23.583 回答