Scala未来序列和超时处理

Question

如何将期货与超时结合起来有一些很好的提示.但是我很好奇如何用Future sequence 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"
  }
}

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

Answer 1

您的代码中有一些内容可能需要重新考虑.对于初学者来说,我并不是将任务提交到其中的忠实粉丝,ExecutionContext其唯一目的是模拟超时并且也在其中Thread.sleep使用.该sleep呼叫被阻止,你可能想避免在纯粹阻断的等待时间的固定金额的缘故执行上下文的任务.我将从这里的答案中窃取,并建议对于纯超时处理,你应该使用我在答案中概述的内容.这HashedWheelTimer是一个高效的计时器实现,比刚睡觉的任务更适合超时处理.

现在,如果你走这条路线,下一次改变我会建议处理每个未来的个别超时相关故障.如果您希望单个故障完全失败,Future则从sequence呼叫中返回聚合,则不执行任何额外操作.如果你不希望这样的事情发生,而是想将超时返回一些默认值,而不是,那么你可以使用recover在Future类似这样的:

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

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

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

每个未来都有一个超时,因此不会无限运行.没有必要IMO阻止那里并通过atMostparam 提供额外的超时处理层Await.result.但我想这可以假设你对非阻塞方法没问题.如果你真的需要阻止,那么你不应该等待timeout * futures.size一段时间.这些期货并行运行; 那里的超时应该只需要与期货本身的个别超时一样长(或者只是稍微长一点来解释cpu /时间的任何延迟).它当然不应该是超时*期货的总数.