Akka 中的期货超时答案

【问题标题】：Timeout for Futures in AkkaAkka 中的期货超时
【发布时间】：2011-11-21 16:12:41
【问题描述】：

我们有一个服务器来处理不同参与者的投资组合和证券（在它内部）。对于证券数量较少（

akka.dispatch.FutureTimeoutException: Futures 在 [5000] 毫秒后超时

我可以通过增加akka 配置中的超时来绕过这个错误，这是正确的做法吗？在 1.2 之前的 akka 版本中，我可以在 actor 内部设置 self.timeout ，但这已被弃用。

我遇到的另一个问题（间歇性地）是整个服务器在加入我的投资组合参与者中的 futures.map 代码时挂起：

    //fork out for each security
    val listOfFutures = new ListBuffer[Future[Security]]()
    for (security <- portfolio.getSecurities.toList) {
      val securityProcessor = actorOf[SecurityProcessor].start()
      listOfFutures += (securityProcessor ? security) map {
        _.asInstanceOf[Security]
      }
    }
    EventHandler.info(this,"joining results from security processors")
    //join for each security
    val futures = Future.sequence(listOfFutures.toList)
    futures.map {
      listOfSecurities =>
        portfolioResponse = MergeHelper.merge(portfolio, listOfSecurities)
    }.get

【问题讨论】：

标签： timeout future akka

【解决方案1】：

您没有说明您使用的是哪个版本的 Akka，并且鉴于我使用水晶球的时间有限，我假设您使用的是 1.2。

您可以在调用 ask/ 时指定超时？

（另外，您的代码有点复杂，但我已经在您的其他问题中解决了。）

干杯， √

【讨论】：

正确。 Akka 1.2 服务器卡在下一行。所以要调试我不得不拆分： val futureResult = Future.sequence(portfolio.getSecurities.toList map { security => (actorOf[SecurityProcessor].start() ? security).mapTo[Security] } ) map { Securities => MergeHelper .merge（投资组合，证券）}
是否在调用序列时指定了更长的超时时间？帮助？您也可以尝试 1.3-RC1，它也修复了一些错误。
更长的超时肯定有助于解决超时错误。好的，将尝试 1.3。谢谢你的帮助，维克多！
可以肯定解决我的间歇性挂起的一件事是，如果我将以下内容添加到我的 Actor 中： self.dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("SecurityProcessor") .withNewThreadPoolWithLinkedBlockingQueueWithCapacity(100) .setCorePoolSize(1) 。 setMaxPoolSize(2) .setKeepAliveTimeInMillis(60000) .setRejectionPolicy(new CallerRunsPolicy) .build
如果你在每个actor中设置它，你会为每个actor创建一个新的调度器，其中maxPoolSize(2)根本没有意义，因为一个actor一次只能在一个线程上运行.