京东的Netty实践,京麦TCP网关长链接容器架构

背景java

早期京麦搭建 HTTP 和 TCP 长链接功能主要用于消息通知的推送,并未应用于 API 网关。随着逐步对 NIO 的深刻学习和对 Netty 框架的了解,以及对系统通讯稳定能力愈来愈高的要求,开始有了采用 NIO 技术应用网关实现 API 请求调用的想法,最终在 2016 年实现,并彻底支撑业务化运行。后端

因为诸多的改进,包括 TCP 长链接容器、Protobuf 的序列化、服务泛化调用框架等等,性能比 HTTP 网关提高 10 倍以上,稳定性也远远高于 HTTP 网关。服务器

架构网络

基于 Netty 构建京麦 TCP 网关的长链接容器,做为网关接入层提供服务 API 请求调用。session

1、网络结构架构

客户端经过域名 + 端口访问 TCP 网关,域名不一样的运营商对应不一样的 VIP,VIP 发布在 LVS 上,LVS 将请求转发给后端的 HAProxy,再由 HAProxy 把请求转发给后端的 Netty 的 IP+Port。框架

LVS 转发给后端的 HAProxy,请求通过 LVS,可是响应是 HAProxy 直接反馈给客户端的,这也就是 LVS 的 DR 模式。异步

2、TCP 网关长链接容器架构

TCP 网关的核心组件是 Netty,而 Netty 的 NIO 模型是 Reactor 反应堆模型(Reactor 至关于有分发功能的多路复用器 Selector)。每个链接对应一个 Channel(多路指多个 Channel,复用指多个链接复用了一个线程或少许线程,在 Netty 指 EventLoop),一个 Channel 对应惟一的 ChannelPipeline,多个 Handler 串行的加入到 Pipeline 中,每一个 Handler 关联惟一的 ChannelHandlerContext。oop

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,达到连接的复用。

1、TCP 网关 Netty Server 的 IO 模型

  1. 建立 ServerBootstrap,设定 BossGroup 与 WorkerGroup 线程池。
  2. bind 指定的 port,开始侦听和接受客户端连接。(若是系统只有一个服务端 port 须要监听,则 BossGroup 线程组线程数设置为 1。)
  3. 在 ChannelPipeline 注册 childHandler,用来处理客户端连接中的请求帧。

2、TCP 网关的线程模型

TCP 网关使用 Netty 的线程池,共三组线程池,分别为 BossGroup、WorkerGroup 和 ExecutorGroup。其中,BossGroup 用于接收客户端的 TCP 链接,WorkerGroup 用于处理 I/O、执行系统 Task 和定时任务,ExecutorGroup 用于处理网关业务加解密、限流、路由,及将请求转发给后端的抓取服务等业务操做。

 

 

 

NioEventLoop 是 Netty 的 Reactor 线程,其角色:

  1. Boss Group:做为服务端 Acceptor 线程,用于 accept 客户端连接,并转发给 WorkerGroup 中的线程。
  2. Worker Group:做为 IO 线程,负责 IO 的读写,从 SocketChannel 中读取报文或向 SocketChannel 写入报文。
  3. Task Queue/Delay Task Queue:做为定时任务线程,执行定时任务,例如链路空闲检测和发送心跳消息等。

3、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 等事件。

4、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 的消息推送等等。

 参考:

京东的Netty实践,京麦TCP网关长链接容器架构