Scala future 序列和超时处理

Question

有一些很好的提示如何组合 future with timeouts .
但是我很好奇如何用 做到这一点 future 序列sequenceOfFutures

我的第一种方法看起来像这样

import scala.concurrent._
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits._

object FutureSequenceScala extends App {
  println("Creating futureList")

  val timeout = 2 seconds
  val futures = List(1000, 1500, 1200, 800, 2000) map { ms =>
    val f = future {
      Thread sleep ms
      ms toString
    }
    Future firstCompletedOf Seq(f, fallback(timeout))
  }

  println("Creating waitinglist")
  val waitingList = Future sequence futures
  println("Created")

  val results = Await result (waitingList, timeout * futures.size)
  println(results)

  def fallback(timeout: Duration) = future {
    Thread sleep (timeout toMillis)
    "-1"
  }
}

有没有更好的方法来处理一系列 future 中的超时，或者这是一个有效的解决方案？

Answer 1

您可能需要重新考虑您的代码中的一些内容。首先，我不太喜欢将任务提交到 ExecutionContext。其唯一目的是模拟超时并且还有 Thread.sleep用在他们身上。 sleep call 是阻塞的，您可能希望避免在执行上下文中有一个纯粹为了等待固定时间而阻塞的任务。我要从我的答案中窃取 here并建议对于纯超时处理，您应该使用我在该答案中概述的内容。 HashedWheelTimer是一种高效的计时器实现，与只是休眠的任务相比，它更适合超时处理。

现在，如果你走那条路，我建议的下一个变化是处理每个 future 的与超时相关的故障。如果您希望单个故障完全使聚合失败 Future从 sequence 返回打电话，然后什么都不做。如果您不希望这种情况发生，而是希望超时返回一些默认值，那么您可以使用 recover在 Future像这样:

withTimeout(someFuture).recover{
  case ex:TimeoutException => someDefaultValue
}

完成后，您可以利用非阻塞回调并执行以下操作:

waitingList onComplete{
  case Success(results) => //handle success
  case Failure(ex) => //handle fail
}

每个 future 都有一个超时，因此不会无限运行。无需 IMO 阻止并通过 atMost 提供额外的超时处理层。参数为 Await.result .但我想这假设您对非阻塞方法没问题。如果你真的需要阻止那里，那么你不应该等待 timeout * futures.size多少时间。这些 future 并行运行；那里的超时应该只需要与 future 本身的各个超时一样长(或者稍微长一点以解决 cpu/计时的任何延迟)。它当然不应该是超时时间 * future 总数。