Akka scala-io 询问并等待响应

Question

我在我的 akka 演员中使用 scala-io，在我的情况下，我需要在官方文档 (http://doc.akka.io/docs/akka/snapshot/scala/io-tcp.html) 中发送请求并等待响应，我可以看到答案是异步的。

如何等待响应，我可以使用某种方式吗？ （问）模式

class SocketClient(remoteAddress: InetSocketAddress, listener: ActorRef) extends Actor {

  import Tcp._
  import context.system

  IO(Tcp) ! Connect(remoteAddress)

  def receive = {

    case CommandFailed(_: Connect) =>
      listener ! ConnectFailure
      context stop self

    case Connected(remote, local) =>
      listener ! ConnectSuccess

      val connection = sender
      connection ! Register(self)

      context become {

        case data: ByteString =>
          connection ! Write(data)

        case CommandFailed(w: Write) =>
          Logger.error(s"Error during writing")

        case Received(data) =>
          listener ! data

        case Disconnect =>
          connection ! Close

        case _: ConnectionClosed =>
          Logger.error(s"Connection has been closed ${remoteAddress.getAddress}")
          context stop self
      }
  }
}

我可以使用类似的东西吗：

connection ? Write(data)

Answer 1

是的，但您应该考虑到这样一个事实，即 ask-pattern 只允许您接收来自 actor 的第一个回复。

在您的情况下，它是connection，它可能会回复一些其他甚至不相关的对象（这取决于您选择的背压/确认模式。例如，如果您使用Write - 您可能会收到书面（到socket) 对象的确认而不是响应。

您可以通过以下方式避免它：

• 将 NoAck 用作 AckEvent（请参阅 http://doc.akka.io/docs/akka/snapshot/scala/io-tcp.html，限制读取和写入部分）。

• 使用原子请求/回复（没有多部分）

• 每个协议使用一个参与者（每个“乒乓”序列）

换句话说，ask 模式只是为每条消息创建自己的内部参与者并使其成为发送者，因此所有回复（针对此特定消息）都将发送给该​​微参与者。当它收到第一个回复时 - 未来（由? 返回）完成 - 内部参与者被销毁（因此其他回复将被忽略）。

另外，connection 会自动回复注册的（通过Register 消息）侦听器而不是发送者 - 所以您应该创建中介参与者：

class Asker(connection: ActorRef) extends Actor {
   import Tcp._
   connection ! Register(self); 
   def receive = {
        case x => 
          val parent = sender()
          connection ! x
          context become {case x => parent ! x; context.unbecome()} 
   }   
}

trait TcpAskSupport {
   self: Actor => 

   def asker(connection: ActorRef) = 
      context.child(connection.path.name + "Asker")
        .getOrElse(system.actorOf(Props(classOf[Asker], connection), 
           connection.path.name + "Asker"))
}

使用示例：

class Client extends Actor with TcpAskSupport {

  import Tcp._
  import context.system

  IO(Tcp) ! Connect(new InetSocketAddress("61.91.16.168", 80))
  implicit val timeout = Timeout(new FiniteDuration(5, SECONDS))

  def receive = {
    case CommandFailed(_: Connect) =>
      println("connect failed")
      context stop self

    case c @ Connected(remote, local) =>
      println("Connected" + c)
      val connection = sender()
      asker(connection) ? Write(ByteString("GET /\n", "UTF-8"), NoAck) onComplete {
          case x => println("Response" + x)
      } 
    case x => println("[ERROR] Received" + x)    
  }
}