在用Akka处理下一个消息之前等待异步将来调用

Question

当接收事件时，Akka参与者将一次处理一条消息，直到请求完成后才移动到下一条消息。

例如：

 def doThing():Future[Unit] = /* Non blocking request here */

 def receive = {
     case DoThing => doThing() pipeTo sender
 } 

这将调用doThing（）并开始处理未来，但不会在处理下一条消息之前等待它完成--它将简单地以尽可能快的速度执行队列中的下一条消息。

实质上，Akka似乎认为“返回未来”是“完成处理”，并进入下一个消息。

def receive = {
    case DoThing => sender ! blocking(Await.result(doThing()))
}

Answer 1

正如注释中所建议的，您可以使用stash(http://doc.akka.io/docs/akka/current/scala/actors.html#stash)特性来存储传入消息，同时等待future解析。

需要保存当前发件人，这样您就不会在发件人执行元引用上不正确地关闭。您可以通过一个简单的case类来实现这一点，比如下面定义的case类。

class MyActor extends Actor with Stash {

  import context.dispatcher

  // Save the correct sender ref by using something like
  // case class WrappedFuture(senderRef: ActorRef, result: Any)
  def doThing(): Future[WrappedFuture] = ???

  override def receive: Receive = {
    case msg: DoThing =>
      doThing() pipeTo self

      context.become({
        case WrappedFuture(senderRef, result) =>
          senderRef ! result
          unstashAll()
          context.unbecome()
        case newMsg: DoThing =>
          stash()
      }, discardOld = false)
  }
}