并行运行多个期货，超时返回默认值

Question

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

现在我要一个接一个地等待期货，如果有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，因为我认为合适。

Answer 1

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在下面回答了这个问题。

Answer 2

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

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) 

Answer 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") 
} 

Answer 4

为什么拿不到Future本身进行异常捕获和默认的回报？然后，您可以在每个未来依次简单地使用Await，并且您不必担心未来的异常处理问题。

Answer 5

这可能有点不方便，但您可以简单地测量已用时间并相应地修改超时。假设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。如果超时没有排除，它仍然有效，但某些任务可能会超过其指定的超时运行时间。

Answer 6

这是一个较长的（unakka）答案，它解决了可能的用例，也就是说，如果其中一个值“超时”，您希望使用该结果的默认值并对其执行某些操作（例如取消长时间计算或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)