在 Akka 中有什么方法可以像在 Erlang 中那样使用 {packet,4} 实现数据包框架？ 数据包看起来像这样:

4 bytes length in big endian | body...

例如:

00 00 00 05 H E L L O 0 0 0 5 W O R L D

将是两个数据包“HELLO”和“WORLD”，但它们被作为一个数据包接收。

或者

00 00 00 05 H E L L

现在 Akka 接收到这 8 个字节，但还缺少一个字节，它将在下一次调用“接收”时接收

问题是我的 Actor 的接收总是被部分或全部请求调用，但我只想在接收中获得“ body ”部分，并且只有在它被完全接收时。

因此，它所需要的只是首先读取这 4 个字节，然后等待读取其他 N 个字节(N = 4 字节 header 中的长度)，然后它向我的 Actor 发送消息。有可能吗？

我的服务器代码:

import java.net.InetSocketAddress

import akka.actor.{Props, Actor}
import akka.io.Tcp.Bind
import akka.io.{IO, Tcp}

class Server extends Actor{
    import context.system
    import Tcp._
    IO(Tcp) ! Bind(self, new InetSocketAddress("0.0.0.0", 1234))

    def receive ={
        case bound @ Bound(localAddr) =>
            println("Server is bound to "+localAddr.toString())
        case failed @ CommandFailed(_ : Bind) =>
            context stop self
        case connected @ Connected(remote, local) =>
            val handler = context.actorOf(Props[ClientHandler])
            val connection = sender()
            println(remote.toString + "connected to "+local.toString())

            connection ! Register(handler)
    }
}

最佳答案

据我所知，Akka 或 Scala 中没有这方面的库函数。 Akka 在 ByteString 中进行读写交易，因此我将一个 trait 放在一起，它可以完全满足您的要求。您将发送给您的 Actor 的 ByteString 传递给它。然后它根据 header 中的数据包长度将流分解。它是无状态的，因此它返回一个元组，其中包含提取的数据包列表和 TCP 流中任何未使用的数据作为 ByteString。您将新的 TCP 数据连接到此字节字符串中返回的流的未使用部分，如下例所示。

trait Buffering {

  val MAX_PACKET_LEN: Short = 10000

  /**
   * Extracts complete packets of the specified length, preserving remainder
   * data. If there is no complete packet, then we return an empty list. If
   * there are multiple packets available, all packets are extracted, Any remaining data
   * is returned to the caller for later submission
   * @param data A list of the packets extracted from the raw data in order of receipt
   * @return A list of ByteStrings containing extracted packets as well as any remaining buffer data not consumed
   */
  def getPacket(data: ByteString): (List[ByteString], ByteString) = {

    val headerSize = 2

    @tailrec
    def multiPacket(packets: List[ByteString], current: ByteString): (List[ByteString], ByteString) = {
      if (current.length < headerSize) {
        (packets.reverse, current)
      } else {
        val len = current.iterator.getShort
        if (len > MAX_PACKET_LEN || len < 0) throw new RuntimeException(s"Invalid packet length: $len")
        if (current.length < len + headerSize) {
          (packets.reverse, current)
        } else {
          val rem = current drop headerSize // Pop off header
          val (front, back) = rem.splitAt(len) // Front contains a completed packet, back contains the remaining data
          // Pull of the packet and recurse to see if there is another packet available
          multiPacket(front :: packets, back)
        }
      }
    }
    multiPacket(List[ByteString](), data)
  }

actor 的用法如下:

def receive = buffer(CompactByteString())

def buffer(buf: ByteString): Receive = {
  // Messages inbound from the network
  case Received(data) =>
    val (pkt, remainder) = getPacket(buf ++ data)
    // Do something with your packet
    context become buffer(remainder) 
  case Other Stuff => // Etc
}

关于Scala - 使用 Akka 的 TCP 数据包框架，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/30665811/