netty报文解析之粘包半包问题

粘包问题

Netty 的粘包问题是指在网络传输过程中，由于 TCP 协议本身的特点，导致发送方发送的若干个小数据包被接收方合并成了一个大数据包。这种情况称为粘包。

TCP 协议是面向流的协议，没有数据边界，发送方发送的数据可能会被分成多个数据包进行发送，接收方则需要将这些数据包重新组装为原始数据。当接收方处理不当时，就可能会发生粘包等问题。

造成粘包问题的原因主要有以下几点：

1. 传输的数据量过大或者传输速度过快。
2. 数据包长度不固定，或者协议自定义导致变长。
3. 接收方的读取缓存区大小设置不当。

解决粘包问题的方法有很多种，其中比较常用的方式包括以下几点：

1. 定长解码器：针对长度固定的数据包，采用定长的编码和解码方式，可以有效避免粘包问题。
2. 分隔符解码器：使用特定字符或字符串作为数据包的分隔符，在接收方收到分隔符时进行消息的解码。
3. 消息头加长度字段：在数据包中添加一部分用来表示数据包长度的信息，以便于接收方进行消息的解码和切割。
4. 自定义协议：设计自己的消息传输协议，包括消息格式、头部、长度字段等，来解决粘包问题。

半包问题

半包问题是指在网络传输过程中，接收方无法完整地接收到一个数据包，而只接收到了部分数据包的情况。这种情况称为半包。

造成半包问题的主要原因是数据包的长度超过了接收方的缓存区大小，导致接收方无法一次性接收完整的数据包。协议设计不合理、网络延迟等也可能引起半包问题。

解决半包问题方法和解决粘包问题基本一致。

下面看下具体的例子

定长报文

定长报文就是收发双方约定一次通信的报文长度是固定长度的，服务端按照规定长度接收，客户端按照固定长度返送。这里主要用到FixedLengthFrameDecoder解码器，其构造函数有一个入参来指定报文的长度。

server:

pipeline.addLast(new FixedLengthFrameDecoder(1024))
1

发送数据

// 消息解析
ByteBuf buf = (ByteBuf) msg;
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
String receivedMessage = new String(bytes, "UTF-8");
System.out.println("接收到消息：" + receivedMessage);
// 发送响应
String responseMessage = "Response";
byte[] responseBytes = responseMessage.getBytes("UTF-8");
ByteBuf responseBuf = ctx.alloc().buffer(responseBytes.length);
responseBuf.writeBytes(responseBytes);
ctx.writeAndFlush(responseBuf);

// 释放资源
buf.release();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

client：

客户端只要每次发送按约定长度组装报文即可

固定长度头

固定长度头就是报文整体有两部分组成：报文头+报文体。齐总报文头是固定位置长度，里面会表明报文体长度，消息接收方先定长读取报文头，然后根据报文头指定的报文体长度来定量读取报文体。

这里用到了LengthFieldBasedFrameDecoder解码器。

该解析其有几个重要参数：

maxFrameLength：最大消息长度，报文最大长度

lengthFieldOffset：长度字段的偏移量，如有些报文可能报文头上还有一些其它的标识位，可以将这些标识位跳过

lengthFieldLength：长度字段的长度

lengthAdjustment：长度调整值，这个值也有一定的用处。有些情况长度标识的是包含header头的长度，这个时候可以将该值配置成负数，最后继续往后解析的长度是：lengthFieldLength+lengthAdjustment

initialBytesToStrip：从开始位置截取掉的字节长度，可以把header去掉再往后传给下一个handler，不过一般会保留报文头，业务代理再去解析。LengthFieldBasedFrameDecoder只负责报文接收完整。

整个处理流程：

当接收到来自网络的字节流时，LengthFieldBasedFrameDecoder 首先根据指定的 lengthFieldOffset 和 lengthFieldLength 定位长度字段的位置，并读取长度字段的值。

接下来，根据读取到的长度字段值计算出消息的长度。如果消息的长度超过了指定的 maxFrameLength，则会触发异常处理机制。

如果消息的长度合法，则 LengthFieldBasedFrameDecoder 会读取接下来的指定长度的字节，构成一个完整的消息。

最后，根据配置的 initialBytesToStrip 参数，可以选择是否去除消息长度头。

解码器完成后，将解析出的完整消息传递给下一个处理器进行进一步的处理。

用例：

如我们定义以下一种报文：

长度头(4字节,只是报文体长度)+标识位(1字节)+报文体长度。

则创建LengthFieldBasedFrameDecoder要指定。

lengthFieldOffset=0，lengthFieldLength=4，lengthAdjustment=1，initialBytesToStrip=0(保留报文头)

具体代码：

server端pipeline添加LengthFieldBasedFrameDecoder解码器和FixedLengthServerHandler

pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,4,1,0));
pipeline.addLast(new FixedLengthServerHandler());
1
2

FixedLengthServerHandler处理方法如下：

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf buf = (ByteBuf) msg;

    //消息解析
    int length = buf.readInt();
    byte[] bytes = new byte[length];
    char flag = (char) buf.readByte();
    buf.readBytes(bytes);
    String receivedMessage = new String(bytes, "UTF-8");
    System.out.println("接收到消息：" + receivedMessage+",消息标识："+flag);

    // 发送响应
    String responseMessage = "SUCC";
    byte[] responseBytes = responseMessage.getBytes("UTF-8");
    int responseLength = responseBytes.length;
    ByteBuf responseBuf = ctx.alloc().buffer(4 +1+ responseLength);
    responseBuf.writeInt(responseLength);
    responseBuf.writeBytes("Y".getBytes());
    responseBuf.writeBytes(responseBytes);
    ctx.writeAndFlush(responseBuf);

    buf.release();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

client端：

同样的pipeline添加两个handler

pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,4,1,0));
pipeline.addLast(new FixedLengthClientHandler());
1
2

构造消息发送：

ByteBuf buffer = Unpooled.buffer();
byte[] bytes = "hello".getBytes();
buffer.writeInt(bytes.length);
buffer.writeBytes("X".getBytes());
buffer.writeBytes(bytes);
channel.writeAndFlush(buffer);
1
2
3
4
5
6

FixedLengthClientHandler处理响应报文:

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf buf = (ByteBuf) msg;

    // 消息处理
    int length = buf.readInt();
    char flag = (char) buf.readByte();
    byte[] bytes = new byte[length];
    buf.readBytes(bytes);
    String receivedMessage = new String(bytes, "UTF-8");
    System.out.println("接收到消息：" + receivedMessage+",flag="+flag);
    buf.release();
}
1
2
3
4
5
6
7
8
9
10
11
12

另外这里处理的都是字节流数据，使用原先阻塞BIO socket也是可以的，不局限于ByteBuf。

如socket发送接收上面定长报文头数据：

Socket socket = new Socket("localhost", 8080);
OutputStream outputStream = socket.getOutputStream();
InputStream inputStream = socket.getInputStream();

// 发送消息
String message = "Hello";
byte[] data = message.getBytes();

ByteBuffer buffer = ByteBuffer.allocate(4);
buffer.putInt(data.length);

outputStream.write(buffer.array());
outputStream.write("X".getBytes());
outputStream.write(data);
outputStream.flush();
//接收响应
byte[] lenB = new byte[4];
inputStream.read(lenB);
char flag = (char) inputStream.read();
ByteBuffer buff = ByteBuffer.wrap(lenB);
int len = buff.getInt();

byte[] resp = new byte[len];
inputStream.read(resp);
System.out.println("响应:"+new String(resp) +",flag="+flag);

outputStream.close();
inputStream.close();
socket.close();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

分隔符报文

分隔符报文就是将报文按固定字符进行分割，这里使用DelimiterBasedFrameDecoder 解析器。

入参可指定分隔符及最大报文长度。

与之相似的还有LineBasedFrameDecoder按行读取，就是以 '\n’换行符当作分隔符。

自定义报文

基本上LengthFieldBasedFrameDecoder解码器已经满足解决报文粘包问题，如果还有其它比较复杂的报文，可以自定义协议报文格式进行处理，一个基本原则还是要有一个报文长度标识，然后按具体长度进行读取。