当前位置：首页 > news >正文

Netty实现通信框架

news 2025/7/22 10:50:06

一、LengthFieldBasedFrameDecoder的参数解释

1、LengthFieldBasedFrameDecoder的构造方法参数

看下最多参数的构造方法

/*** Creates a new instance.** @param byteOrder*        the {@link ByteOrder} of the length field* @param maxFrameLength*        the maximum length of the frame.  If the length of the frame is*        greater than this value, {@link TooLongFrameException} will be*        thrown.* @param lengthFieldOffset*        the offset of the length field* @param lengthFieldLength*        the length of the length field* @param lengthAdjustment*        the compensation value to add to the value of the length field* @param initialBytesToStrip*        the number of first bytes to strip out from the decoded frame* @param failFast*        If <tt>true</tt>, a {@link TooLongFrameException} is thrown as*        soon as the decoder notices the length of the frame will exceed*        <tt>maxFrameLength</tt> regardless of whether the entire frame*        has been read.  If <tt>false</tt>, a {@link TooLongFrameException}*        is thrown after the entire frame that exceeds <tt>maxFrameLength</tt>*        has been read.*/public LengthFieldBasedFrameDecoder(ByteOrder byteOrder, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,int lengthAdjustment, int initialBytesToStrip, boolean failFast) {this.byteOrder = checkNotNull(byteOrder, "byteOrder");checkPositive(maxFrameLength, "maxFrameLength");checkPositiveOrZero(lengthFieldOffset, "lengthFieldOffset");checkPositiveOrZero(initialBytesToStrip, "initialBytesToStrip");if (lengthFieldOffset > maxFrameLength - lengthFieldLength) {throw new IllegalArgumentException("maxFrameLength (" + maxFrameLength + ") " +"must be equal to or greater than " +"lengthFieldOffset (" + lengthFieldOffset + ") + " +"lengthFieldLength (" + lengthFieldLength + ").");}this.maxFrameLength = maxFrameLength;this.lengthFieldOffset = lengthFieldOffset;this.lengthFieldLength = lengthFieldLength;this.lengthAdjustment = lengthAdjustment;this.lengthFieldEndOffset = lengthFieldOffset + lengthFieldLength;this.initialBytesToStrip = initialBytesToStrip;this.failFast = failFast;}

byteOrder：表示字节顺序，有两个常量分别是BIG_ENDIAN(多字节值的字节从最高有效到最低有效排序)和LITTLE_ENDIAN(多字节值的字节从最低有效到最高有效排序)
maxFrameLength：包的最大长度，字面意思是最大帧的长度
lengthFieldOffset：指的是长度域的偏移量，表示跳过指定个数字节之后的才是长度域
lengthFieldLength：记录该帧数据长度的字段，也就是长度域本身的长度
lengthAdjustment：长度的一个修正值，可正可负，Netty在读取到数据包的长度值N后，认为接下来的N个字节都是需要读取的，但是根据实际情况，有可能需要增加N的值，也有可能需要减少N的值，具体增加多少，减少多少，写在这个参数里
initialBytesToStrip：从数据帧中跳过的字节数，表示得到一个完整的数据包之后，扔掉这个数据包中多少字节数，才是后续业务实际需要的业务数据
failFast：如果为 true，则表示读取到长度域，TA 的值的超过 maxFrameLength，就抛出一个TooLongFrameException，而为false表示只有当真正读取完长度域的值表示的字节之后，才会抛出TooLongFrameException，默认情况下设置为 true，建议不要修改，否则可能会造成内存溢出

2、LengthFieldBasedFrameDecoder的构造方法参数对应数据包

在LengthFieldBasedFrameDecoder上也有对构造方法主要参数的解释，下面表示从解码前到解码后的字节参数分别对应

lengthFieldOffset   = 0
lengthFieldLength   = 2
lengthAdjustment    = 0
initialBytesToStrip = 0 (= do not strip header)BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
+--------+----------------+      +--------+----------------+
| Length | Actual Content |----->| Length | Actual Content |
| 0x000C | "HELLO, WORLD" |      | 0x000C | "HELLO, WORLD" |
+--------+----------------+      +--------+----------------+
// lengthFieldOffset长度存在位置的偏移量是0
// lengthFieldLength，一个十六进制4位，000C是16位，所以是占两个字节是2
// lengthAdjustment，000C就是12，"HELLO, WORLD"正好是12个字节，所以是0
// initialBytesToStrip没有丢数据，所以是0

lengthFieldOffset   = 0
lengthFieldLength   = 2
lengthAdjustment    = 0
initialBytesToStrip = 2 (= the length of the Length field)BEFORE DECODE (14 bytes)         AFTER DECODE (12 bytes)
+--------+----------------+      +----------------+
| Length | Actual Content |----->| Actual Content |
| 0x000C | "HELLO, WORLD" |      | "HELLO, WORLD" |
+--------+----------------+      +----------------+
// lengthFieldOffset长度存在位置的偏移量是0
// lengthFieldLength，一个十六进制4位，000C是16位，所以是占两个字节是2
// lengthAdjustment，000C就是12，"HELLO, WORLD"正好是12个字节，所以是0
// initialBytesToStrip丢弃了长度两个字节，所以是2

lengthFieldOffset   =  0
lengthFieldLength   =  2
lengthAdjustment    = -2 (= the length of the Length field)
initialBytesToStrip =  0BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
+--------+----------------+      +--------+----------------+
| Length | Actual Content |----->| Length | Actual Content |
| 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
+--------+----------------+      +--------+----------------+
// lengthFieldOffset长度存在位置的偏移量是0
// lengthFieldLength，一个十六进制4位，000C是16位，所以是占两个字节是2
// lengthAdjustment，000E就是14，而"HELLO, WORLD"是12个字节，少了两个字节所以是-2
// initialBytesToStrip丢弃了长度两个字节，所以是2

lengthFieldOffset   = 2 (= the length of Header 1)
lengthFieldLength   = 3
lengthAdjustment    = 0
initialBytesToStrip = 0BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
+----------+----------+----------------+      +----------+----------+----------------+
| Header 1 |  Length  | Actual Content |----->| Header 1 |  Length  | Actual Content |
|  0xCAFE  | 0x00000C | "HELLO, WORLD" |      |  0xCAFE  | 0x00000C | "HELLO, WORLD" |
+----------+----------+----------------+      +----------+----------+----------------+
// lengthFieldOffset，Header占用了两个字节，所以长度存在位置的偏移量是2
// lengthFieldLength，一个十六进制4位，00000C是24位，所以是占两个字节是3
// lengthAdjustment，00000C就是12，"HELLO, WORLD"正好是12个字节，所以是0
// initialBytesToStrip没有丢数据，所以是0

lengthFieldOffset   = 0
lengthFieldLength   = 3
lengthAdjustment    = 2 (= the length of Header 1)
initialBytesToStrip = 0BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
+----------+----------+----------------+      +----------+----------+----------------+
|  Length  | Header 1 | Actual Content |----->|  Length  | Header 1 | Actual Content |
| 0x00000C |  0xCAFE  | "HELLO, WORLD" |      | 0x00000C |  0xCAFE  | "HELLO, WORLD" |
+----------+----------+----------------+      +----------+----------+----------------+
// lengthFieldOffset长度存在位置的偏移量是0
// lengthFieldLength，一个十六进制4位，00000C是24位，所以是占两个字节是3
// lengthAdjustment，00000C就是12，"HELLO, WORLD"正好是12个字节，而Header多占用了两个字节，所以是2
// initialBytesToStrip没有丢数据，所以是0

lengthFieldOffset   = 1 (= the length of HDR1)
lengthFieldLength   = 2
lengthAdjustment    = 1 (= the length of HDR2)
initialBytesToStrip = 3 (= the length of HDR1 + LEN)BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
+------+--------+------+----------------+      +------+----------------+
| HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
| 0xCA | 0x000C | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
+------+--------+------+----------------+      +------+----------------+
// lengthFieldOffset，HDR1占用了一个字节，所以长度存在位置的偏移量是1
// lengthFieldLength，一个十六进制4位，000C是16位，所以是占两个字节是2
// lengthAdjustment，000C就是12，"HELLO, WORLD"正好是12个字节，而HDR2多占用了一个字节，所以是1
// initialBytesToStrip丢弃了HDR1和Length共三个字节，所以是3

lengthFieldOffset   =  1
lengthFieldLength   =  2
lengthAdjustment    = -3 (= the length of HDR1 + LEN, negative)
initialBytesToStrip =  3BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
+------+--------+------+----------------+      +------+----------------+
| HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
| 0xCA | 0x0010 | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
+------+--------+------+----------------+      +------+----------------+
// lengthFieldOffset，HDR1占用了一个字节，所以长度存在位置的偏移量是1
// lengthFieldLength，一个十六进制4位，0010是16位，所以是占两个字节是2
// lengthAdjustment，0010就是16，"HELLO, WORLD"加上HDR2是13个字节，所以是-3
// initialBytesToStrip丢弃了HDR1和Length共三个字节，所以是3

EmbeddedChannel的单元测试暂时略过，后面有空再看

三、手写Netty大体结构

1、功能描述

基于 Netty 的 NIO 通信框架

提供消息的编解码框架，可以实现 POJO 的序列化和反序列化（【编解码】与【序列化】一块）

消息内容防篡改机制（就跟我们web开发的鉴权一样，在处理之前先校验一下内容合法性）

提供基于 IP 地址的白名单接入认证机制

断线重连机制

2、通信模型

（1）客户端发送应用握手请求消息，携带节点ID等有效身份认证信息；

（2）服务端对应用握手请求消息进行合法性校验，包括节点ID有效性校验、节点重复登录校验和IP地址合法性校验，校验通过后，返回登录成功的应用握手应答消息；

（3）链路建立成功之后，客户端发送业务消息；

（4）链路成功之后，服务端发送心跳消息；

（5）链路建立成功之后，客户端发送心跳消息；

（6）链路建立成功之后，服务端发送业务消息；

（7）服务端退出时，服务端关闭连接，客户端感知对方关闭连接后，被动关闭客户端连接。

协议通信双方链路建立成功之后，双方可以进行全双工通信，无论客户端还是服务端，都可以主动发送请求消息给对方，通信方式可以是 TWO WAY 或者 ONE WAY。双方之间的心跳采用 Ping-Pong 机制，当链路处于空闲状态时，客户端主动发送 Ping 消息给服务端，服务端接收到 Ping 消息后发送应答消息 Pong 给客户端，如果客户端连续发送 N 条 Ping 消息都没有接收到服务端返回的 Pong 消息，说明链路已经挂死或者对方处于异常状态，客户端主动关闭连接，间隔周期 T 后发起重连操作，直到重连成功。

3、消息体定义

消息定义包含两部分：

消息头；消息体。

在消息的定义上，因为是同步处理模式，不考虑应答消息需要填入请求消息 ID，所以消息头中只有一个消息的 ID。如果要支持异步模式，则请求消息头和应答消息头最好分开设计，应答消息头中除了包括本消息的 ID 外，还应该包括请求消息 ID，以方便请求消息的发送方根据请求消息 ID 做对应的业务处理。

消息体则支持 Java 对象类型的消息内容。

Netty 消息定义表

名称	类型	长度	描述
header	Header	变长	消息头定义
body	Object	变长	消息的内容

消息头定义（Header）

名称	类型	长度	描述
md5	String	变长	消息体摘要，缺省 MD5 摘要
msgId	Long	64	消息的ID
Type	Byte	8	0：业务请求消息 1：业务响应消息 2：业务one way消息 3：握手请求消息 4：握手应答消息 5：心跳请求消息 6：心跳应答消息
Priority	Byte	8	消息优先级：0~255
Attachment	Map<String, Object>	变长	可选字段，用于扩展消息头

4、链路的建立

客户端的说明如下：如果 A 节点需要调用 B 节点的服务，但是 A 和 B 之间还没有建立物理链路，则有调用方主动发起连接，此时，调用方为客户端，被调用方为服务端。考虑到安全，链路建立需要通过基于 Ip 地址或者号段的黑白名单安全认证机制，作为样例，本协议使用基于 IP 地址的安全认证，如果有多个 Ip，通过逗号进行分割。在实际的商用项目中，安全认证机制会更加严格，例如通过密钥对用户名和密码进行安全认证。

客户端与服务端链路建立成功之后，由客户端发送业务握手请求的认证消息，服务端接收到客户端的握手请求消息之后，如果 IP 校验通过，返回握手成功应答消息给客户端，应用层链路建立成功。握手应答消息中消息体为 byte 类型的结果，0：认证成功；-1 认证失败；服务端关闭连接。

链路建立成功之后，客户端和服务端就可以互相发送业务消息了，在客户端和服务端的消息通信过程中，业务消息体的内容需要通过 MD5 进行摘要防篡改。

5、可靠性设计

1）、心跳机制

在凌晨等业务低谷时段，如果发生网络闪断、连接被 Hang 住等问题时，由于没有业务消息，应用程序很难发现。到了白天业务高峰期时，会发生大量的网络通信失败，严重的会导致一段时间进程内无法处理业务消息。为了解决这个问题，在网络空闲时采用心跳机制来检测链路的互通性，一旦发现网络故障，立即关闭链路，主动重连。

当读或者写心跳消息发生 I/O 异常的时候，说明已经中断，此时需要立即关闭连接，如果是客户端，需要重新发起连接。如果是服务端，需要清空缓存的半包信息，等到客户端重连。

空闲的连接和超时

检测空闲连接以及超时对于及时释放资源来说是至关重要的。由于这是一项常见的任务， Netty 特地为它提供了几个 ChannelHandler 实现。

IdleStateHandler 当连接空闲时间太长时，将会触发一个 IdleStateEvent 事件。然后，可以通过在 ChannelInboundHandler 中重写 userEventTriggered()方法来处理该 IdleStateEvent 事件。

ReadTimeoutHandler 如果在指定的时间间隔内没有收到任何的入站数据，则抛出一个 ReadTimeoutException 并关闭对应的 Channel。可以通过重写你的 ChannelHandler 中的 exceptionCaught()方法来检测该 Read-TimeoutException。

2）、重连机制

如果链路中断，等到 INTEVAL 时间后，由客户端发起重连操作，如果重连失败，间隔周期 INTERVAL 后再次发起重连，直到重连成功。

为了保持服务端能够有充足的时间释放句柄资源，在首次断连时客户端需要等待 INTERVAL 时间之后再发起重连，而不是失败后立即重连。

为了保证句柄资源能够及时释放，无论什么场景下重连失败，客户端必须保证自身的资源被及时释放，包括但不现居 SocketChannel、Socket 等。

重连失败后，可以打印异常堆栈信息，方便后续的问题定位。

3）、重复登录保护

当客户端握手成功之后，在链路处于正常状态下，不允许客户端重复登录，以防止客户端在异常状态下反复重连导致句柄资源被耗尽。

服务端接收到客户端的握手请求消息之后，对 IP 地址进行合法性校验，如果校验成功，在缓存的地址表中查看客户端是否已经登录，如果登录，则拒绝重复登录，同时关闭 TCP 链路，并在服务端的日志中打印握手失败的原因。

客户端接收到握手失败的应答消息之后，关闭客户端的 TCP 连接，等待 INTERVAL 时间之后，再次发起 TCP 连接，直到认证成功。

6、实现

Handler示意图如下：

其中认证申请和认证检查可以在完成后移除。

7、前期准备

定义了消息有关的实体类，为了防篡改，消息体需要进行摘要， vo 包下提供了 EncryptUtils 类，可以对消息体进行摘要，目前支持 MD5、SHA-1 和 SHA-256 这三种，缺省为 MD5，其中 MD5 额外提供了加盐摘要。同时定义了有关序列化和反序列化的工具类和Handler，本项目中序列化使用了 Kryo 序列化框架。

8、服务端

服务端中 NettyServe 类是服务端的主入口，内部使用了 ServerInit 类进行 Handler 的安装。

最先安装的当然是解决粘包和半包问题的 Handler，很自然，这里应该用 LengthFieldBasedFrameDecoder 进行解码，为了实现方便，我们也没有在消息报文中附带消息的长度，由 Netty 帮我们在消息报文的最开始增加长度，所以编码器选择了 LengthFieldPrepender。

接下来，自然就是序列化和反序列化，直接使用我们在 kryocodec 下已经准备好的 KryoDecoder 和 KryoEncoder 即可。

服务端需要进行登录检查、心跳应答、业务处理，对应着三个 handler，于是我们分别安装了 LoginAuthRespHandler、HeartBeatRespHandler、ServerBusiHandler。

为了节约网络和服务器资源，如果客户端长久没有发送业务和心跳报文，我们认为客户端出现了问题，需要关闭这个连接，我们引入 Netty 的 ReadTimeoutHandler，当一定周期内（默认值 50s，我们设定为 15s）没有读取到对方任何消息时，会触发一个 ReadTimeouttException，这时我们检测到这个异常，需要主动关闭链路，并清除客户端登录缓存信息，等待客户端重连。

9、客户端

客户端的主类是 NettyClient，并对外提供一个方法 send，供业务使用内部使用了 ClientInit 类进行 Handler 的安装。

最先安装的当然是解决粘包和半包问题的 Handler，同样这里应该用 LengthFieldBasedFrameDecoder 进行解码，编码器选择了 LengthFieldPrepender。

接下来，自然就是序列化和反序列化，依然使用 KryoDecoder 和 KryoEncoder 即可。

客户端需要主动发出认证请求和心跳请求。

在 TCP 三次握手，链路建立后，客户端需要进行应用层的握手认证，才能使用服务，这个功能由 LoginAuthReqHandler 负责，而这个 Handler 在认证通过后，其实就没用了，所以在认证通过后，可以将这个 LoginAuthReqHandler 移除（其实服务端的认证应答 LoginAuthRespHandler 同样也可以移除）。

对于发出心跳请求有两种实现方式，一是定时发出，本框架的第一个版本就是这种实现方式，但是这种方式其实有浪费的情况，因为如果客户端和服务器正在正常业务通信，其实是没有必要发送心跳的；所以第二种方式就是，当链路写空闲时，为了维持通道，避免服务器关闭链接，发出心跳请求。为了实现这一点，我们首先在整个 pipeline 的最前面安装一个 CheckWriteIdleHandler进行写空闲检测，空闲时间定位8S，取服务器读空闲时间15S的一半，然后再安装一个 HearBeatReqHandler，因为写空闲会触发一个 FIRST_WRITER_IDLE_STATE_EVENT 入站事件，我们在 HearBeatReqHandler 的 userEventTriggered 方法中捕捉这个事件，并发出心跳请求报文。

考虑到在我们的实现中并没有双向心跳（即是客户端向服务器发送心跳请求，是服务器也向客户端发送心跳请求），客户端这边同样需要检测服务器是否存活，所以我们客户端这边安装了一个 ReadTimeoutHandler，捕捉 ReadTimeoutException 后提示调用者，并关闭通信链路，触发重连机制。

为了测试，单独建立一个 BusiClient，模拟业务方的调用。因为客户端的网络通信代码是在一个线程中单独启动的，为了协调主线程和通信线程的工作，我们引入了线程中的等待通知机制。

Netty对于我来说过于复杂，后面再深究细节吧