scala - 有什么方法可以记录 akka 超时并提供有关错误位置的更多信息？

Question

10 月 16 日星期三更新)今天有一个 PullRequest，它提供了有关超时的目标信息。
https://github.com/akka/akka/pull/1780

Akka 的超时异常非常无用。

有什么方法可以获取有关超时中发生的位置/情况的有用消息？

像这样的异常(exception)没有帮助

java.util.concurrent.TimeoutException: Futures timed out after [5000] milliseconds
    at akka.dispatch.DefaultPromise.ready(Future.scala:834)
    at akka.dispatch.DefaultPromise.ready(Future.scala:811)
    at akka.dispatch.Await$.ready(Future.scala:64)
    at nl.cwi.crisp.examples.p2p.scala.Network.<init>(Node.scala:136)
    at nl.cwi.crisp.examples.p2p.scala.Main$$anonfun$11.apply(Node.scala:164)
    at nl.cwi.crisp.examples.p2p.scala.Main$$anonfun$11.apply(Node.scala:164)
    at akka.actor.ActorCell.newActor(ActorCell.scala:488)
    at akka.actor.ActorCell.create$1(ActorCell.scala:506)
    at akka.actor.ActorCell.systemInvoke(ActorCell.scala:591)
    at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:191)
    at akka.dispatch.Mailbox.run(Mailbox.scala:160)
    at akka.dispatch.ForkJoinExecutorConfigurator$MailboxExecutionTask.exec(AbstractDispatcher.scala:505)
    at akka.jsr166y.ForkJoinTask.doExec(ForkJoinTask.java:259)
    at akka.jsr166y.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:997)
    at akka.jsr166y.ForkJoinPool.runWorker(ForkJoinPool.java:1495)
    at akka.jsr166y.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:104)

Answer 1

使用当前的 akka 代码，它不会发生。让我们先来看看为什么。如果你看 PromiseActorRef你可以看到的对象:

def apply(provider: ActorRefProvider, timeout: Timeout): PromiseActorRef = {
  val result = Promise[Any]()
  val scheduler = provider.guardian.underlying.system.scheduler
  val a = new PromiseActorRef(provider, result)
  implicit val ec = a.internalCallingThreadExecutionContext
  val f = scheduler.scheduleOnce(timeout.duration) { result tryComplete Failure(new AskTimeoutException("Timed out")) }
  result.future onComplete { _ ‚áí try a.stop() finally f.cancel() }
  a
}

这是安排并行(与实际参与者调用并行)超时的地方。这个类没有关于它发送什么消息或它发送给哪个actor ref 的上下文。这可能就是为什么它只是说“超时”，这不是很有帮助。我有点希望类型安全人员稍微调整一下以提供更多信息，但如果他们不这样做或者如果您在此期间想要一些东西，您可以尝试这样的事情:

object NewAskPattern{
  implicit def ask(ref:ActorRef) = new BetterTimeoutMessageSupportAskableRef(ref)
}

class BetterTimeoutMessageSupportAskableRef(ref: ActorRef) {
  import akka.pattern.AskableActorRef
  val askRef = new AskableActorRef(ref)

  def ask(message: Any)(implicit timeout: Timeout, ec:ExecutionContext): Future[Any] = 
    (askRef ? message) recover{
      case to:TimeoutException => 
        val recip = askRef.actorRef.path
        val dur = timeout.duration
        throw new TimeoutException(s"Timed out sending message $message to recipient $recip using timeout of $dur") 
    }

  def ?(message: Any)(implicit timeout: Timeout, ec:ExecutionContext): Future[Any] = 
    ask(message)(timeout, ec)
} 

class MySlowActor extends Actor{
  def receive = {
    case any => 
      Thread.sleep(5000)
      sender ! "bar"
  }
}

object NewMessageTest{
  import NewAskPattern.ask

  def main(args: Array[String]) {
    implicit val timeout = Timeout(2 seconds)
    val sys = ActorSystem()
    import sys.dispatcher

    val slow = sys.actorOf(Props[MySlowActor])
    val fut = slow ? "foo"
    fut onComplete (println(_))
  }
}

这里的总体思路是包装 AskableActorRef来自 Akka lib 并稍微增强它。我要接 Future返回者 ask并添加一个 recover它的组合器允许我在超时时调整消息。由于此类具有发送消息的上下文以及发送给谁的上下文，因此它可以制定更有用的消息。然后是 NewAskPattern对象包含新的隐式为您提供 BetterTimeoutMessageSupportAskableRef这使您可以获得这种增强的行为。这是一个完美的解决方案吗？可能不是，但如果您真的想要这种行为，这对您来说可能是一个很好的起点。