Akka Streams: Using TCP

引子

准确来说，IO API 不算是 Akka Streams的内容，但它却是 Streaming IO (TCP) 的基础。

IO API 编程的第一步，是获取一个管理对象的引用。比如我们想针对TCP编程，可以这么获取 TCP Manager：

import akka.io.{ IO, Tcp }
 
val manager = IO(Tcp)

import akka.io.{ IO, Tcp }

val manager = IO(Tcp)

manager 是一个 actor 对象，用于处理底层 I/O资源和操作（诸如channels/selectors，或是workers的初始化），同时对于上层提供基于actor 模型的抽象

IO TCP

IO TCP 编程也是 CS 模型的，好处是不需要写繁琐的 SOCKET 逻辑——当然，如果你乐意，也可以掺合这些 SOCKET 事务，比如：在客户端进行连接时，指定SOCKET的SO_NODELAY( windows里叫TCP_NODELAY)选项，以启用 Nagle’s algorithm（Akka里默认是禁用）：

IO(Tcp) ! Connect(remote, 
                  options = Vector(SO.TcpNoDelay(false)))

1 2	IO(Tcp) ! Connect(remote, options = Vector(SO.TcpNoDelay(false)))

在 IO TCP编程中，无论是客户端还是服务端，都是一个actor对象，通过它与manger对象及对端的连接进行消息上的交互：

客户端

io-tcp-client

客户端流程：

连接到服务器，即向 manger 发送Connect消息
接收消息：CommandFailed 意味着连接失败；Connected 代表了一个成功的连接
Connected 消息的sender()就代表了与服务器建立TCP连接的代理对象，它亦是一个 ActorRef，我们可以向它Register(handler)消息，以指定处理它消息的handler（还是一个ActorRef)
利用hander，可以向对端的服务器发送消息，即写入ByteString到SOCKET以抵达服务器，写入消息，是特定的Tcp.Write、Tcp.WriteFile、Tcp.CompoundWrite消息，分别对应了：写入ByteString、写入文件以及混合写入多种消息
同时，如果在Tcp.Write消息中指定了ACK事件参数，那么在发送成功后，handler将会接受到ACK消息

服务端

服务端流程：

绑定到服务器，即向 manger 发送Bind消息
接收消息：CommandFailed 代表绑定失败；Bound 代表绑定成功；Connected 代表了一个从客户端发来的连接
Connected 消息的sender()就代表了与客户端建立TCP连接的代理对象，它亦是一个 ActorRef，我们可以向它Register(handler)消息，以指定处理它消息的handler……
利用hander，可以向对端的服务器发送消息，即写入ByteString到SOCKET以抵达服务器，写入消息，是特定的Tcp.Write、Tcp.WriteFile、Tcp.CompoundWrite消息，分别对应了：写入ByteString、写入文件以及混合写入多种消息
同时，如果在Tcp.Write消息中指定了ACK事件参数，那么在发送成功后，handler将会接受到ACK消息

除了绑定等前置步骤，代理对象的处理和客户端是一样的……

TCP CS 模型实例

客户端

import java.net.InetSocketAddress
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.io.Tcp._
import akka.io.{IO, Tcp}
import akka.stream.ActorMaterializer
import akka.util.ByteString

object TcpClient_ extends App {

  class Client(remote: InetSocketAddress, handler: ActorRef) extends Actor {
    override def preStart(): Unit = {
      IO(Tcp) ! Connect(remote, options = Vector(SO.TcpNoDelay(false)))
    }

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

      case c@Connected(remote, local) =>
        handler ! c
        val connection = sender()
        connection ! Register(self)
        context become {
          case data: ByteString =>
            connection ! Write(data)
          case CommandFailed(w: Write) =>
            handler ! "write failed"
          case Received(data) =>
            handler ! data
          case "close" =>
            connection ! Close
          case _: ConnectionClosed =>
            handler ! "connection closed"
            context stop self
        }
    }
  }

  class ClientHandler extends Actor {
    def receive: Receive = {
      case c@Connected(remote, local) =>
        println(s"Connnected -> remote: $remote, local: $local")
        sender() ! ByteString("hello")

      case b: ByteString => println(b.utf8String)

      case f@("connect failed" | "connection closed" | "write failed") =>
        println(f)
    }
  }

  object Client {
    def generate = system.actorOf(Props(classOf[Client],
      serverAddress,
      system.actorOf(Props[ClientHandler])))
  }

  // 调用部分
  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val serverAddress = new InetSocketAddress("localhost", 8000)

  val client = Client.generate
  Thread.sleep(1000)
  val client2 = Client.generate
}

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

import akka.io.Tcp._

import akka.io.{IO, Tcp}

import akka.stream.ActorMaterializer

import akka.util.ByteString

object TcpClient_ extends App {

class Client(remote: InetSocketAddress, handler: ActorRef) extends Actor {

override def preStart(): Unit = {

IO(Tcp) ! Connect(remote, options = Vector(SO.TcpNoDelay(false)))

}

def receive: Receive = {

case CommandFailed(_: Connect) =>

handler ! "connect failed"

context stop self

case c@Connected(remote, local) =>

handler ! c

val connection = sender()

connection ! Register(self)

context become {

case data: ByteString =>

connection ! Write(data)

case CommandFailed(w: Write) =>

handler ! "write failed"

case Received(data) =>

handler ! data

case "close" =>

connection ! Close

case _: ConnectionClosed =>

handler ! "connection closed"

context stop self

}

class ClientHandler extends Actor {

def receive: Receive = {

case c@Connected(remote, local) =>

println(s"Connnected -> remote: $remote, local: $local")

sender() ! ByteString("hello")

case b: ByteString => println(b.utf8String)

case f@("connect failed" | "connection closed" | "write failed") =>

println(f)

}

object Client {

def generate = system.actorOf(Props(classOf[Client],

serverAddress,

system.actorOf(Props[ClientHandler])))

}

// 调用部分

implicit val system = ActorSystem()

implicit val ec = system.dispatcher

implicit val materializer = ActorMaterializer()

val serverAddress = new InetSocketAddress("localhost", 8000)

val client = Client.generate

Thread.sleep(1000)

val client2 = Client.generate

}

服务端

import java.net.InetSocketAddress
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.io.{IO, Tcp}
import Tcp._
import akka.stream.ActorMaterializer
import akka.util.ByteString

object TcpServer_ extends App {

  case object Ack extends Event

  var connections = Map.empty[String, ActorRef]

  def send(msg: String): Unit = {
    for (c <- connections) {
      c._2 ! msg
    }
  }

  class Server(serverAddress: InetSocketAddress) extends Actor {
    override def preStart(): Unit = {
      IO(Tcp) ! Bind(self, serverAddress)
    }

    override def postRestart(thr: Throwable): Unit =
      context stop self

    def receive: Receive = {
      case Bound(localAddress) =>
        println(s"Server is bound to ${localAddress}")

      case CommandFailed(_: Bind) => context stop self

      case Connected(remote, local) =>
        val connection = sender()
        val handler = ConnectionHandler.instanceOf(connection)
        connection ! Register(handler)
        connections += connection.path.toSerializationFormat -> handler
    }
  }

  object ConnectionHandler {
    def instanceOf(connection: ActorRef) =
      system.actorOf(Props(classOf[ConnectionHandler], connection))
  }

  class ConnectionHandler(connection: ActorRef) extends Actor {

    def receive: Receive = {
      case Received(data) =>
        println(data.utf8String)

      case PeerClosed =>
        context stop self

      case msg: String =>
        connection ! Write(ByteString(
          s"${msg} to: ${connection.path.toSerializationFormat}"), Ack)

      case Ack =>
        println("ACK")
    }
  }

  // 调用部分
  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val serverAddress = new InetSocketAddress("localhost", 8000)
  val server = system.actorOf(Props(classOf[Server], serverAddress))

  Thread.sleep(5000)
  send("hi")
  Thread.sleep(5000)
  send("holy")
}

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

import akka.io.{IO, Tcp}

import Tcp._

import akka.stream.ActorMaterializer

import akka.util.ByteString

object TcpServer_ extends App {

case object Ack extends Event

var connections = Map.empty[String, ActorRef]

def send(msg: String): Unit = {

for (c <- connections) {

c._2 ! msg

}

class Server(serverAddress: InetSocketAddress) extends Actor {

override def preStart(): Unit = {

IO(Tcp) ! Bind(self, serverAddress)

}

override def postRestart(thr: Throwable): Unit =

context stop self

def receive: Receive = {

case Bound(localAddress) =>

println(s"Server is bound to ${localAddress}")

case CommandFailed(_: Bind) => context stop self

case Connected(remote, local) =>

val connection = sender()

val handler = ConnectionHandler.instanceOf(connection)

connection ! Register(handler)

connections += connection.path.toSerializationFormat -> handler

}

object ConnectionHandler {

def instanceOf(connection: ActorRef) =

system.actorOf(Props(classOf[ConnectionHandler], connection))

}

class ConnectionHandler(connection: ActorRef) extends Actor {

def receive: Receive = {

case Received(data) =>

println(data.utf8String)

case PeerClosed =>

context stop self

case msg: String =>

connection ! Write(ByteString(

s"${msg} to: ${connection.path.toSerializationFormat}"), Ack)

case Ack =>

println("ACK")

}

// 调用部分

implicit val system = ActorSystem()

implicit val ec = system.dispatcher

implicit val materializer = ActorMaterializer()

val serverAddress = new InetSocketAddress("localhost", 8000)

val server = system.actorOf(Props(classOf[Server], serverAddress))

Thread.sleep(5000)

send("hi")

Thread.sleep(5000)

send("holy")

}

运行结果

客户端

Connnected -> remote: localhost/127.0.0.1:8000, local: /127.0.0.1:64775
Connnected -> remote: localhost/127.0.0.1:8000, local: /127.0.0.1:64776
hi to: akka://default/system/IO-TCP/selectors/$a/2#1599711018
hi to: akka://default/system/IO-TCP/selectors/$a/1#629943617
holy to: akka://default/system/IO-TCP/selectors/$a/2#1599711018
holy to: akka://default/system/IO-TCP/selectors/$a/1#629943617

Connnected -> remote: localhost/127.0.0.1:8000, local: /127.0.0.1:64775

Connnected -> remote: localhost/127.0.0.1:8000, local: /127.0.0.1:64776

hi to: akka://default/system/IO-TCP/selectors/$a/2#1599711018

hi to: akka://default/system/IO-TCP/selectors/$a/1#629943617

holy to: akka://default/system/IO-TCP/selectors/$a/2#1599711018

holy to: akka://default/system/IO-TCP/selectors/$a/1#629943617

服务端

Server is bound to /127.0.0.1:8000
hello
hello
ACK
ACK
ACK
ACK

Server is bound to /127.0.0.1:8000

hello

ACK

打赏作者

您的支持将激励我继续创作！

去打赏

Mo Ye

A Conservative

Published

September 15, 2016

Mo Ye in FP | September 15, 2016

Akka Streams: Using TCP

引子

IO TCP

客户端

服务端

TCP CS 模型实例

客户端

服务端

运行结果

客户端

服务端

Mo Ye

A Conservative

Published

September 15, 2016

Cancel Reply

Write a Comment