Akka Classic TCP I/O 使用指南：基于 Actor 的底层 TCP 连接、回压与关闭机制详解

后端并发编程异步编程【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址https://gitcode.com/gh_mirrors/ak/akka-core点击查看免费下载导读本文以 Akka 官方文档 io-tcp.md 为主体系统讲解如何通过Classic Actor经典 Actor API使用 Akka 提供的底层 TCP 传输能力。你将掌握如何获取 TCP Manager、建立出站连接Connect、监听入站连接Bind、关闭连接Close / ConfirmedClose / Abort、通过WriteCommand族写入数据以及最重要的 ACK/NACK 写入回压与 Push/Pull 读取回压模型并辅以仓库内完整可运行的 Echo 服务器源码作为实战范本。1. 依赖引入TCP I/O API 属于akka-actor模块所有 Akka I/O API 均内置于该模块无需额外引入独立模块。在项目中添加如下依赖以 Akka BOM 统一管理版本// sbt libraryDependencies com.typesafe.akka %% akka-actor % AkkaVersion!-- Maven -- dependency groupIdcom.typesafe.akka/groupId artifactIdakka-actor_2.13/artifactId versionAkkaVersion/version /dependency// Gradle implementation com.typesafe.akka:akka-actor_2.13:AkkaVersion注意Akka 依赖托管于 Akka 官方安全库仓库secure library repository需要按 https://account.akka.io/token 中的说明配置带 token 的 URL 才能访问。2. 总体模型一切经由 Manager ActorAkka I/O 的所有 API 都通过manager 对象访问。使用任何 I/O API 的第一步就是获取对应 manager 的引用。TCP 的 manager 是akka.io.Tcp扩展ExtensionId[TcpExt]获取方式如下Scalaimport akka.io.{ IO, Tcp } import context.system // 隐式提供给 IO(Tcp) val manager IO(Tcp)Javafinal ActorRef tcpManager Tcp.get(getContext().getSystem()).manager();完整示例见 IODocSpec.scala 与 EchoManager.java。Manager 的角色它本身是一个 Actor负责管理底层 I/O 资源selector、channel并为具体任务实例化 worker——例如为监听入站连接创建专门的监听器。因此与 TCP 的交互本质上都是“向 manager 发送命令消息、由 manager 派生的子 actor 回复事件消息”的纯 Actor 消息驱动过程。提示所有 TCP 相关的命令/事件消息均定义于 Tcp.scalaScala 直接使用Tcp.XXXJava 通过TcpMessage.xxx()工厂方法构造。3. 出站连接Connect → Connected → Register3.1 发送 Connect 命令连接远程地址的第一步是向 TCP manager 发送Connect消息ScalaIO(Tcp) ! Connect(remote)Javatcp.tell(TcpMessage.connect(remote), getSelf());Connect消息除最简单的形式外还支持两个可选参数源码见 Tcp.scala参数类型说明localAddressOption[InetSocketAddress]/InetSocketAddress指定本地要绑定的地址不指定则由内核自动选择optionsIterable[SocketOption]套接字选项列表如SO.TcpNoDelaytimeoutOption[FiniteDuration]/Duration连接超时pullModeBoolean默认false是否启用 Pull 读取模式详见第 8 节SO_NODELAY 默认开启在 Akka 中SO_NODELAYWindows 上为TCP_NODELAY套接字选项默认即为true与操作系统默认值无关。这会禁用 Nagle 算法显著降低大多数应用中的延迟。如需覆盖可在 options 列表中传入SO.TcpNoDelay(false)。3.2 连接建立流程TCP manager 收到Connect后要么回复CommandFailed要么派生一个内部 actor 代表这条新连接。这个连接 actor 会向原 Connect 命令的发送者发送Connected消息。连接在被激活之前无法使用必须向连接 actor 发送Register消息告诉它“谁将接收来自套接字的数据”。若迟迟不注册连接 actor 内部存在一个超时超时后会自动关闭自身并清理资源。完整客户端示例Scalaclass Client(remote: InetSocketAddress, listener: ActorRef) extends Actor { import Tcp._ import context.system IO(Tcp) ! Connect(remote) def receive { case CommandFailed(_: Connect) listener ! connect failed context.stop(self) case c Connected(remote, local) listener ! c val connection sender() connection ! Register(self) context.become { case data: ByteString connection ! Write(data) case CommandFailed(_: Write) // O/S buffer was full listener ! write failed case Received(data) listener ! data case close connection ! Close case _: ConnectionClosed listener ! connection closed context.stop(self) } } }Java 对应版本使用TcpMessage.register(getSelf())激活连接并用getContext().become(connected(...))切换到已连接状态完整代码见 IODocTest.java。3.3 Register 的语义Register消息Tcp.scala携带三个字段final case class Register( handler: ActorRef, keepOpenOnPeerClosed: Boolean false, useResumeWriting: Boolean true)handler接收所有入站数据、并获知连接关闭事件的 actorkeepOpenOnPeerClosed为true时对端关闭其写半端后连接不会自动关闭需显式发出关闭命令用于实现半关闭详见第 5 节useResumeWriting为false时启用NACK 模式写入回压不暂停写入失败即回复CommandFailed为true时启用带挂起的 NACK 模式详见第 7 节。另外连接 actor 会watch监视注册的 handler当 handler 终止时连接会被关闭并释放全部内部资源。4. 入站连接Bind → Bound → Connected4.1 发起绑定创建 TCP 服务器并监听入站连接需要向 TCP manager 发送Bind命令ScalaIO(Tcp) ! Bind(self, new InetSocketAddress(localhost, 0))Javatcp.tell(TcpMessage.bind(getSelf(), new InetSocketAddress(localhost, 0), 100), getSelf());Bind消息的参数Tcp.scala参数默认值说明handler—接收所有Connected消息的 actorlocalAddress—监听的地址端口指定为0表示绑定随机端口实际端口见Bound消息backlog100内核为该端口保留的未 accept 连接数上限超出则拒绝连接optionsNil套接字选项pullModefalse是否启用 Pull 模式见第 8.2 节4.2 绑定成功与接受连接发送Bind的 actor 会收到Bound消息表示服务器已就绪Bound携带实际绑定的InetSocketAddress即解析后的 IP 与正确端口号。此后处理连接的方式与出站连接一致收到Connected后为每个连接派生一个 handler actor并将 handler 通过Register注册给连接 actor。写入数据则可由系统中任意 actor向连接 actor即发送过Connected的 actor发出。服务器示例Scalaclass Server extends Actor { import Tcp._ import context.system IO(Tcp) ! Bind(self, new InetSocketAddress(localhost, 0)) def receive { case b Bound(_) context.parent ! b case CommandFailed(_: Bind) context.stop(self) case c Connected(_, _) context.parent ! c val handler context.actorOf(Props[SimplisticHandler]()) val connection sender() connection ! Register(handler) } }最简单的 handlerclass SimplisticHandler extends Actor { import Tcp._ def receive { case Received(data) sender() ! Write(data) case PeerClosed context.stop(self) } }该 handler 收到数据即原样写回Echo对端关闭则终止自身。4.3 监听端口的生命周期出站连接与入站监听的一个关键差异管理监听端口的内部 actor即Bound消息的发送者会watch 监听 actor。当监听 actor 终止时监听端口随之关闭、相关资源全部释放但已建立的连接不会因此被终止。5. 关闭连接连接可通过向连接 actor 发送三种关闭命令之一来关闭5.1 Close正常关闭命令Close/TcpMessage.close()行为发送FIN关闭连接不等待对端确认待写数据pending writes会先被 flush。成功通知Closed。5.2 ConfirmedClose确认式关闭命令ConfirmedClose/TcpMessage.confirmedClose()行为发送FIN关闭本端发送方向但继续接收数据直到对端也关闭连接为止待写数据会先被 flush。成功通知ConfirmedClosed。5.3 Abort立即终止命令Abort/TcpMessage.abort()行为向对端发送RST立即终止连接待写数据不会被 flush。成功通知Aborted。5.4 对端关闭与错误关闭事件PeerClosed对端关闭连接时发送给监听者listener。默认情况下本端随后也会自动关闭连接若希望在Register中将keepOpenOnPeerClosed设为true则连接保持打开直到收到上述关闭命令之一支持半关闭连接场景。ErrorClosed发生错误导致连接被迫关闭时发送给监听者携带错误原因。上述关闭通知全部是ConnectionClosed的子类型源码见 Tcp.scala不需要细粒度区分关闭事件的监听者可以统一按ConnectionClosed处理。6. 写入数据WriteCommand 的三种实现连接建立后系统中任意 actor 都可以向连接 actor 发送WriteCommand写入数据。WriteCommand是抽象类有 3 个具体实现6.1 Tcp.Write最简单的写入命令包装一个ByteString实例和一个 ack 事件final case class Write(data: ByteString, ack: Event) extends SimpleWriteCommandByteString是 Akka 的不可变内存数据模型详见 io.md 的 ByteString 一节一个或多个数据块的最大总大小为2 GB2^31 字节。6.2 Tcp.WriteFile直接发送文件中的原始数据final case class WriteFile(filePath: String, position: Long, count: Long, ack: Event)它指定磁盘上的一段连续字节范围直接经由连接发送无需先加载进 JVM 内存因此可持有超过 2 GB 的数据同样支持 ack 事件。对发送大文件场景非常高效。6.3 Tcp.CompoundWrite把多个Write和/或WriteFile组合成一个原子写命令一次性写往连接带来三大好处最小开销连接 actor 同一时刻只能处理一个写命令合并成一个CompoundWrite可避免用 ACK 协议逐个喂给连接 actor原子性保证WriteCommand是原子的合并后其他 actor 无法把写入插入到你的写序列中间——多 actor 并发写同一连接时这是很难靠其他手段实现的重要特性子写可独立 ACKCompoundWrite的子写本身就是普通Write/WriteFile各自可以请求 ack这些 ACK 在对应子写完成时发出。这允许你通过组合一个空的请求 ACK 的写来为一次写入附加多个 ACK或在任意位置插入中间 ACK 来跟踪CompoundWrite的传输进度。组合语法Scala 使用:/:Tcp.scalaJava 使用prepend系列方法。7. 写入回压TCP 连接 actor 的三种背压模式TCP 连接 actor 的基本模型是没有内部缓冲同一时刻只能处理一个写命令即只能缓冲一个尚未完全交给操作系统内核的写。因此拥塞必须由用户层处理——写入和读取两侧都是如此。7.1 ACK 模式每个Write命令携带一个任意对象作为 ack只要该对象不是NoAck数据全部成功写入套接字后ack 对象就会返回给Write的发送者。若在收到此确认前不发起新的写入就不会因缓冲溢出而失败。7.2 NACK 模式前一个写尚未完成时到达的每个写都会被回复CommandFailed携带失败的写命令。仅依赖此机制要求实现的协议能够容忍跳过写入例如每个写本身都是独立有效消息、不要求全部送达。启用方式在Register时将useResumeWriting设为false。7.3 带挂起的 NACK 模式NACK-based with write suspending与 NACK 模式类似但一旦某个写失败后续写入将全部失败直到收到ResumeWriting消息。ResumeWriting会在最后一个已接受写完成后以WritingResumed消息应答。若驱动连接的 actor 实现了缓冲并在收到WritingResumed后重发被 NACK 的消息则每条消息都能恰好一次送达网络套接字。7.4 读取回压Push 与 PullPush 读取连接 actor 一有数据就作为Received事件发给注册的 reader actor。reader 想向对端施加回压时发送SuspendReading暂停新数据接收直到发送ResumeReading后才恢复Received事件。Pull 读取每次发出Received事件后连接 actor自动挂起从套接字接收数据直到 reader 发送ResumeReading。因此新数据通过发送ResumeReading拉取而来。详见第 8 节。重要限制以上所有流控方案只在一个写者/读者对一个连接 actor时成立多个 actor 同时向同一连接发送写命令时无法获得一致的结果。8. 完整实战ACK 回压 Echo 服务器以下 Echo 服务器演示 ACK 写入回压 Push 读取 跨 TCP 连接的端到端回压传播。完整源码见 EchoServer.scalaScala与 EchoHandler.java、SimpleEchoHandler.javaJava。8.1 关键准备保持半开连接为保证在关闭连接前能把所有未写数据回写客户端需要在激活连接时设置keepOpenOnPeerClosed trueScalacase Connected(remote, local) log.info(received connection from {}, remote) val handler context.actorOf(Props(handlerClass, sender(), remote)) sender() ! Register(handler, keepOpenOnPeerClosed true)Javaconnection.tell( TcpMessage.register( handler, true, // -- keepOpenOnPeerClosed flag true), getSelf());服务器管理者EchoManager使用SupervisorStrategy.stoppingStrategy连接损坏不可恢复、preStart中发起Bind、postRestart中直接stop(self)不重启详见 EchoManager.java 与 EchoServer.scala。8.2 SimpleEchoHandlerACK 模式核心逻辑ScalaSimpleEchoHandlerclass SimpleEchoHandler(connection: ActorRef, remote: InetSocketAddress) extends Actor with ActorLogging { import Tcp._ context.watch(connection) case object Ack extends Event def receive { case Received(data) buffer(data) connection ! Write(data, Ack) context.become({ case Received(data) buffer(data) case Ack acknowledge() case PeerClosed closing true }, discardOld false) case PeerClosed context.stop(self) } ... }原则很简单写完一块数据后必须等Ack回来才能发送下一块。等待期间切换行为把新到的数据先缓冲起来。辅助函数实现缓冲与水位控制private def buffer(data: ByteString): Unit { storage : data stored data.size if (stored maxStored) { log.warning(sdrop connection to [$remote] (buffer overrun)) context.stop(self) } else if (stored highWatermark) { log.debug(ssuspending reading) connection ! SuspendReading suspended true } } private def acknowledge(): Unit { require(storage.nonEmpty, storage was empty) val size storage(0).size stored - size transferred size storage storage.drop(1) if (suspended stored lowWatermark) { log.debug(resuming reading) connection ! ResumeReading suspended false } if (storage.isEmpty) { if (closing) context.stop(self) else context.unbecome() } else connection ! Write(storage(0), Ack) }其中最有趣的是最后一段逻辑Ack移除缓冲中最旧的数据块若这是最后一块则视对端是否已关闭关闭连接或回到空闲行为否则发送下一块缓冲数据并继续等待Ack。同时缓冲量超过高水位highWatermark maxStored * 5 / 10时发送SuspendReading低于低水位lowWatermark时发送ResumeReading。8.3 端到端回压传播原理读取侧回压可以跨连接传播回对端的写者向连接 actor 发送SuspendReading后本端不再从套接字读取数据存在一定延迟因为连接 actor 处理该命令需要时间因此缓冲需留出足够余量。这导致本端 OS 内核缓冲区逐渐填满进而触发 TCP 窗口机制阻止对端发送、填满对端的写缓冲区最终对端写者无法再向套接字推送数据——这就是跨 TCP 连接的端到端背压实现机制。8.4 EchoHandlerNACK 模式 写挂起完整的EchoHandler演示 NACK 模式源码 EchoServer.scala / EchoHandler.java写透传状态writingdef writing: Receive { case Received(data) connection ! Write(data, Ack(currentOffset)) buffer(data) case Ack(ack) acknowledge(ack) case CommandFailed(Write(_, Ack(ack))) connection ! ResumeWriting context.become(buffering(ack)) case PeerClosed if (storage.isEmpty) context.stop(self) else context.become(closing) }原则是持续写入直到收到CommandFailed用 ACK 只做重发缓冲的剪枝。收到失败后进入buffering状态处理所有排队数据的重发def buffering(nack: Int): Receive { var toAck 10 var peerClosed false { case Received(data) buffer(data) case WritingResumed writeFirst() case PeerClosed peerClosed true case Ack(ack) if ack nack acknowledge(ack) case Ack(ack) acknowledge(ack) if (storage.nonEmpty) { if (toAck 0) { writeFirst() // 失败后先用 ACK 模式 toAck - 1 } else { writeAll() // 然后恢复乐观写透传 context.become(if (peerClosed) closing else writing) } } else if (peerClosed) context.stop(self) else context.become(writing) } }要点进入该状态时所有已缓冲的写都已发送给连接 actor因此ResumeWriting排在这些写之后——在收到WritingResumed之前会先收到全部未完成的CommandFailed此状态下被忽略。WritingResumed表示内部排队的写已全部完成后续写入不会失败。EchoHandler据此在失败后先以 ACK 模式写 10 次toAck 10再恢复乐观写透传。关闭状态closing始终发送全部未完成消息、确认所有成功写入若失败则切换行为等待WritingResumed后重新开始def closing: Receive { case CommandFailed(_: Write) connection ! ResumeWriting context.become({ case WritingResumed writeAll() context.unbecome() case ack: Int acknowledge(ack) }, discardOld false) case Ack(ack) acknowledge(ack) if (storage.isEmpty) context.stop(self) }9. Pull 模式读取与入站 Pull 监听Push 读取模式下数据一到就发给 actor因此 Echo 服务器必须为写入慢于到达的情况维护入站缓冲。Pull 模式则可以完全消除该缓冲。9.1 PullEcho零缓冲回显Scala完整代码见 ReadBackPressure.scalacase object Ack extends Event class PullEcho(connection: ActorRef) extends Actor { override def preStart(): Unit connection ! ResumeReading def receive { case Received(data) connection ! Write(data, Ack) case Ack connection ! ResumeReading } }Java见 JavaReadBackPressure.javaOverride public void preStart() throws Exception { connection.tell(TcpMessage.resumeReading(), getSelf()); } Override public Receive createReceive() { return receiveBuilder() .match(Tcp.Received.class, message - { ByteString data message.data(); connection.tell(TcpMessage.write(data, new Ack()), getSelf()); }) .match(Ack.class, message - { connection.tell(TcpMessage.resumeReading(), getSelf()); }) .build(); }原理直到上一次写入被连接 actor 完全 ACK 才恢复读取。每个 Pull 模式连接 actor 初始都处于挂起状态在preStart中发送ResumeReading通知连接 actor 准备接收第一块数据。因为只在上一块数据完全写完后才恢复读取无需维护缓冲。9.2 出站连接的 Pull 模式将Connect的pullMode参数设为trueScalaIO(Tcp) ! Connect(listenAddress, pullMode true)Javatcp.tell( TcpMessage.connect( new InetSocketAddress(localhost, 3000), null, options, timeout, true), getSelf());9.3 入站连接的 Pull 模式Pull 式监听器将Bind的pullMode参数设为true则该监听器接受的所有连接都使用 Pull 模式读取Scalaoverride def preStart(): Unit IO(Tcp) ! Bind(self, new InetSocketAddress(localhost, 0), pullMode true)Javatcp.tell( TcpMessage.bind(getSelf(), new InetSocketAddress(localhost, 0), 100, options, true), getSelf());Pull 式监听器的两个额外效应所有入站连接自动启用 Pull 读取接受连接本身也变为 Pull 式处理完一个或多个Connected事件后监听器必须发送ResumeAccepting才能继续接受新连接。Pull 模式监听器初始为挂起状态因此绑定成功后必须先发送ResumeAccepting才开始接受连接Scaladef receive { case Bound(localAddress) // Accept connections one by one sender() ! ResumeAccepting(batchSize 1) context.become(listening(sender())) monitor ! localAddress } def listening(listener: ActorRef): Receive { case Connected(remote, local) val handler context.actorOf(Props(classOf[PullEcho], sender())) sender() ! Register(handler, keepOpenOnPeerClosed true) listener ! ResumeAccepting(batchSize 1) }Java.match(Tcp.Bound.class, x - { listener getSender(); // Accept connections one by one listener.tell(TcpMessage.resumeAccepting(1), getSelf()); }) .match(Tcp.Connected.class, x - { ActorRef handler getContext().actorOf(Props.create(PullEcho.class, getSender())); getSender().tell(TcpMessage.register(handler), getSelf()); // Resume accepting connections listener.tell(TcpMessage.resumeAccepting(1), getSelf()); })如示例所示处理完一个入站连接后需要再次ResumeAccepting。ResumeAccepting的batchSize参数指定在需要下一次ResumeAccepting之前可以接受多少个新连接。10. 小结Akka 的 TCP I/O API 将底层 socket 操作完全封装进 Actor 模型Manager 单点入口IO(Tcp)/Tcp.get(system).manager()获取管理 actor其余全部通过消息驱动连接生命周期出站Connect、入站Bind成功后以Register激活并注册数据接收者三种关闭语义Close立即 FIN、ConfirmedClose半关闭、AbortRST关闭事件统一归入ConnectionClosed写入回压三模式ACK、NACK、NACK挂起useResumeWritingCompoundWrite提供原子批量写读取回压两模式PushSuspendReading/ResumeReading与 PullpullMode trueResumeReading拉取、监听器ResumeAccepting按批接受连接。文中所有示例均取自仓库内可运行测试与文档代码IODocSpec.scala、IODocTest.java、EchoServer.scala、EchoManager.java、ReadBackPressure.scala 与 JavaReadBackPressure.java读者可直接对照源码深入研习。赞分享后端并发编程异步编程【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址https://gitcode.com/gh_mirrors/ak/akka-core点击查看免费下载相关推荐Akka Classic Networking 完全指南基于 Actor 模型的 TCP/UDP/DNS 底层 I/O 编程Akka Classic Networking 完全指南基于 Actor 模型的 TCP/UDP/DNS 底层 I/O 编程 导读 本文围绕 Akka 经典后端并发编程异步编程Envoy TCP Proxy 负载卸载点 tcp_proxy_on_data 详解资源压力下的连接关闭机制Envoy TCP Proxy 负载卸载点 tcp_proxy_on_data 详解资源压力下的连接关闭机制 导读 本文围绕 Envoy 在 TCP 代理T云原生服务网格网络微服务为什么TCP TIME_WAIT需要等待2MSL彻底搞懂连接关闭的安全机制为什么TCP TIME_WAIT需要等待2MSL彻底搞懂连接关闭的安全机制 TCP连接关闭过程中TIME_WAIT状态就像一位忠诚的卫士默默守护着数据传输文档教程知识库创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考