使用akka - io TCP初始化akka参与者

Question

使用Akka-IO TCP，在actor中建立连接的过程如下:

class MyActor(remote: InetSocketAddress) extends Actor {
  IO(Tcp) ! Connect(remote)    //this is the first step, remote is the address to connect to
  def receive = {
    case CommandFailed(_: Connect) => context stop self // failed to connect
    case Connected(remote, local) =>
      val connection = sender()
      connection ! Register(self)
      // do cool things...  
  }
}

您向IO(Tcp)发送Connect消息，并期望收到CommandFailed或Connected消息。

现在，我的目标是创建一个包装TCP连接的actor，但是我希望我的actor只在连接建立后才开始接受消息-否则，在等待Connected消息时，它将开始接受查询，但没有人可以发送它们。

What I tried:

class MyActor(address: InetSocketAddress) extends Actor {
  def receive = {
    case Initialize =>
      IO(Tcp) ! Connect(address)
      context.become(waitForConnection(sender()))
    case other => sender ! Status.Failure(new Exception(s"Connection to $address not established yet."))
  }
  private def waitForConnection(initializer: ActorRef): Receive = {
    case Connected(_, _) =>
      val connection = sender()
      connection ! Register(self)
      initializer ! Status.Success(Unit)
      // do cool things
    case CommandFailed(_: Connect) =>
      initializer ! Status.Failure(new Exception("Failed to connect to " + host))
      context stop self
  }
}

我的第一个receive期待一个组成的Initialize消息，它将触发整个连接过程，一旦完成，Initialize的发送方收到成功消息，并知道它可以知道开始发送查询。

我不太喜欢它，它迫使我用

创建actor

val actor = system.actorOf(MyActor.props(remote))
Await.ready(actor ? Initialize, timeout)

它将不是很"重启"友好。

有没有办法保证我的actor在Tcp层回复Connected之前不会开始从邮箱接收消息?

Answer 1

使用Stash特性来存储您现在无法处理的消息。当每个过早的消息到达时，使用stash()来延迟它。打开连接后，使用unstashAll()将这些消息返回到邮箱进行处理。然后，您可以使用become()切换到消息处理状态。

Answer 2

为了使您的actor更适合重启，您可以重写与actor生命周期相关的方法，例如preStart或postStop。Akka文档对actor的start, stop和restart钩子有很好的解释。

class MyActor(remote: InetSocketAddress) extends Actor {
  override def preStart() {
    IO(Tcp) ! Connect(remote) 
  }
  ...
}

现在你可以用val actor = system.actorOf(MyActor.props(remote))开始你的演员。它在启动时建立连接，重新启动时重新建立连接。