| 编辑推荐: |
| 本文来自于infoq，本文重点介绍京麦
TCP 网关的背景、架构及 Netty 的应用实践。 |
|
背景
早期京麦搭建 HTTP 和 TCP 长连接功能主要用于消息通知的推送，并未应用于 API 网关。随着逐步对
NIO 的深入学习和对 Netty 框架的了解，以及对系统通信稳定能力越来越高的要求，开始有了采用
NIO 技术应用网关实现 API 请求调用的想法，最终在 2016 年实现，并完全支撑业务化运行。
由于诸多的改进，包括 TCP 长连接容器、Protobuf 的序列化、服务泛化调用框架等等，性能比
HTTP 网关提升 10 倍以上，稳定性也远远高于 HTTP 网关。
架构
基于 Netty 构建京麦 TCP 网关的长连接容器，作为网关接入层提供服务 API 请求调用。
一、网络结构
客户端通过域名 + 端口访问 TCP 网关，域名不同的运营商对应不同的 VIP，VIP 发布在 LVS
上，LVS 将请求转发给后端的 HAProxy，再由 HAProxy 把请求转发给后端的 Netty
的 IP+Port。
LVS 转发给后端的 HAProxy，请求经过 LVS，但是响应是 HAProxy 直接反馈给客户端的，这也就是
LVS 的 DR 模式。
二、TCP 网关长连接容器架构
TCP 网关的核心组件是 Netty，而 Netty 的 NIO 模型是 Reactor 反应堆模型（Reactor
相当于有分发功能的多路复用器 Selector）。每一个连接对应一个 Channel（多路指多个 Channel，复用指多个连接复用了一个线程或少量线程，在
Netty 指 EventLoop），一个 Channel 对应唯一的 ChannelPipeline，多个
Handler 串行的加入到 Pipeline 中，每个 Handler 关联唯一的 ChannelHandlerContext。
TCP 网关长连接容器的 Handler 就是放在 Pipeline 的中。我们知道 TCP 属于
OSI 的传输层，所以建立 Session 管理机制构建会话层来提供应用层服务，可以极大的降低系统复杂度。
所以，每一个 Channel 对应一个 Connection，一个 Connection 又对应一个
Session，Session 由 Session Manager 管理，Session 与 Connection
是一一对应的，Connection 保存着 ChannelHandlerContext（ChannelHanderContext
可以找到 Channel），Session 通过心跳机制来保持 Channel 的 Active 状态。
每一次 Session 的会话请求（ChannelRead）都是通过 Proxy 代理机制调用 Service
层，数据请求完毕后通过写入 ChannelHandlerConext 再传送到 Channel 中。数据下行主动推送也是如此，通过
Session Manager 找到 Active 的 Session，轮询写入 Session 中的
ChannelHandlerContext，就可以实现广播或点对点的数据推送逻辑。
Netty 的应用实践
京麦 TCP 网关使用 Netty Channel 进行数据通信，使用 Protobuf 进行序列化和反序列化，每个请求都将被封装成
Byte 二进制字节流，在整个生命周期中，Channel 保持长连接，而不是每次调用都重新创建 Channel，达到链接的复用。
一、TCP 网关 Netty Server 的 IO 模型
创建 ServerBootstrap，设定 BossGroup 与 WorkerGroup 线程池。
bind 指定的 port，开始侦听和接受客户端链接。（如果系统只有一个服务端 port 需要监听，则
BossGroup 线程组线程数设置为 1。）
在 ChannelPipeline 注册 childHandler，用来处理客户端链接中的请求帧。
二、TCP 网关的线程模型
TCP 网关使用 Netty 的线程池，共三组线程池，分别为 BossGroup、WorkerGroup
和 ExecutorGroup。其中，BossGroup 用于接收客户端的 TCP 连接，WorkerGroup
用于处理 I/O、执行系统 Task 和定时任务，ExecutorGroup 用于处理网关业务加解密、限流、路由，及将请求转发给后端的抓取服务等业务操作。
NioEventLoop 是 Netty 的 Reactor 线程，其角色：
Boss Group：作为服务端 Acceptor 线程，用于 accept 客户端链接，并转发给
WorkerGroup 中的线程。
Worker Group：作为 IO 线程，负责 IO 的读写，从 SocketChannel 中读取报文或向
SocketChannel 写入报文。
Task Queue／Delay Task Queue：作为定时任务线程，执行定时任务，例如链路空闲检测和发送心跳消息等。
三、TCP 网关执行时序图
其中步骤一至步骤九是 Netty 服务端的创建时序，步骤十至步骤十三是 TCP 网关容器创建的时序。
步骤一：创建 ServerBootstrap 实例，ServerBootstrap 是 Netty
服务端的启动辅助类。
步骤二：设置并绑定 Reactor 线程池，EventLoopGroup 是 Netty 的 Reactor
线程池，EventLoop 负责所有注册到本线程的 Channel。
步骤三：设置并绑定服务器 Channel，Netty Server 需要创建 NioServerSocketChannel
对象。
步骤四：TCP 链接建立时创建 ChannelPipeline，ChannelPipeline 本质上是一个负责和执行
ChannelHandler 的职责链。
步骤五：添加并设置 ChannelHandler，ChannelHandler 串行的加入 ChannelPipeline
中。
步骤六：绑定监听端口并启动服务端，将 NioServerSocketChannel 注册到 Selector
上。
步骤七：Selector 轮训，由 EventLoop 负责调度和执行 Selector 轮询操作。
步骤八：执行网络请求事件通知，轮询准备就绪的 Channel，由 EventLoop 执行 ChannelPipeline。
步骤九：执行 Netty 系统和业务 ChannelHandler，依次调度并执行 ChannelPipeline
的 ChannelHandler。
步骤十：通过 Proxy 代理调用后端服务，ChannelRead 事件后，通过发射调度后端 Service。
步骤十一：创建 Session，Session 与 Connection 是相互依赖关系。
步骤十二：创建 Connection，Connection 保存 ChannelHandlerContext。
步骤十三：添加 SessionListener，SessionListener 监听 SessionCreate
和 SessionDestory 等事件。
四、TCP 网关源码分析
1. Session 管理
Session 是客户端与服务端建立的一次会话链接，会话信息中保存着 SessionId、连接创建时间、上次访问事件，以及
Connection 和 SessionListener，在 Connection 中保存了 Netty
的 ChannelHandlerContext 上下文信息。Session 会话信息会保存在 SessionManager
内存管理器中。
创建 Session 的源码
通过源码分析，如果 Session 已经存在销毁 Session，但是这个需要特别注意，创建 Session
一定不要创建那些断线重连的 Channel，否则会出现 Channel 被误销毁的问题。因为如果在已经建立
Connection(1) 的 Channel 上，再建立 Connection(2)，进入 session.close
方法会将 cxt 关闭，Connection(1) 和 Connection(2) 的 Channel
都将会被关闭。在断线之后再建立连接 Connection(3)，由于 Session 是有一定延迟，Connection(3)
和 Connection(1/2) 不是同一个，但 Channel 可能是同一个。
所以，如何处理是否是断线重练的 Channel，具体的方法是在 Channel 中存入 SessionId，每次事件请求判断
Channel 中是否存在 SessionId，如果 Channel 中存在 SessionId 则判断为断线重连的
Channel。
2. 心跳
心跳是用来检测保持连接的客户端是否还存活着，客户端每间隔一段时间就会发送一次心跳包上传到服务端，服务端收到心跳之后更新
Session 的最后访问时间。在服务端长连接会话检测通过轮询 Session 集合判断最后访问时间是否过期，如果过期则关闭
Session 和 Connection，包括将其从内存中删除，同时注销 Channel 等。
通过源码分析，在每个 Session 创建成功之后，都会在 Session 中添加 TcpHeartbeatListener
这个心跳检测的监听，TcpHeartbeatListener 是一个实现了 SessionListener
接口的守护线程，通过定时休眠轮询 Sessions 检查是否存在过期的 Session，如果轮训出过期的
Session，则关闭 Session。
同时，注意到 session.connect 方法，在 connect 方法中会对 Session
添加的 Listeners 进行添加时间，它会循环调用所有 Listner 的 sessionCreated
事件，其中 TcpHeartbeatListener 也是在这个过程中被唤起。
3. 数据上行
数据上行特指从客户端发送数据到服务端，数据从 ChannelHander 的 channelRead
方法获取数据。数据包括创建会话、发送心跳、数据请求等。这里注意的是，channelRead 的数据包括客户端主动请求服务端的数据，以及服务端下行通知客户端的返回数据，所以在处理
object 数据时，通过数据标识区分是请求 - 应答，还是通知 - 回复。
4. 数据下行
数据下行通过 MQ 广播机制到所有服务器，所有服务器收到消息后，获取当前服务器所持有的所有 Session
会话，进行数据广播下行通知。如果是点对点的数据推送下行，数据也是先广播到所有服务器，每天服务器判断推送的端是否是当前服务器持有的会话，如果判断消息数据中的信息是在当前服务，则进行推送，否则抛弃。
通过源码分析，数据下行则通过 NotifyProxy 的方式发送数据，需要注意的是 Netty 是
NIO，如果下行通知需要获取返回值，则要将异步转同步，所以 NotifyFuture 是实现 java.util.concurrent.Future
的方法，通过设置超时时间，在 channelRead 获取到上行数据之后，通过 seq 来关联 NotifyFuture
的方法。
下行的数据通过 TcpConnector 的 send 方法发送，send 方式则是通过 ChannelHandlerContext
的 writeAndFlush 方法写入 Channel，并实现数据下行，这里需要注意的是，之前有另一种写法就是
cf.await，通过阻塞的方式来判断写入是否成功，这种写法偶发出现 BlockingOperationException
的异常。
使用阻塞获取返回值的写法
关于 BlockingOperationException 的问题我在 StackOverflow
进行提问，非常幸运的得到了 Norman Maurer（Netty 的核心贡献者之一）的解答。
最终结论大致分析出，在执行 write 方法时，Netty 会判断 current thread
是否就是分给该 Channe 的 EventLoop，如果是则行线程执行 IO 操作，否则提交 executor
等待分配。当执行 await 方法时，会从 executor 里 fetch 出执行线程，这里就需要
checkDeadLock，判断执行线程和 current threads 是否时同一个线程，如果是就检测为死锁抛出异常
BlockingOperationException。
总结
本篇文章粗浅地向大家介绍了京麦 TCP 网关中使用 Netty 实现长连接容器的架构，对涉及 TCP
长连接容器搭建的关键点一一进行了阐述，以及对源码进行简单地分析。在京麦发展过程里 Netty 还有很多的实践应用，例如
Netty4.11+HTTP2 实现 APNs 的消息推送等等。 |
