netty 数据包黏包拆包处理器使用及遇到的问题

古城微笑少年丶 2022-05-24 10:51 165阅读 0赞

# netty 数据包黏包拆包处理器使用及遇到的问题 #

<table> 
 <thead> 
  <tr> 
   <th align="center">结构</th> 
   <th align="center">类型</th> 
   <th align="center">解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">tag</td> 
   <td align="center">byte</td> 
   <td align="center">标签，默认值为0x01</td> 
  </tr> 
  <tr> 
   <td align="center">encode</td> 
   <td align="center">byte</td> 
   <td align="center">编码格式，默认值为0x01</td> 
  </tr> 
  <tr> 
   <td align="center">encrypt</td> 
   <td align="center">byte</td> 
   <td align="center">加密类型，默认值为0x01</td> 
  </tr> 
  <tr> 
   <td align="center">command</td> 
   <td align="center">int</td> 
   <td align="center">指令，根据指令去解析body</td> 
  </tr> 
  <tr> 
   <td align="center">length</td> 
   <td align="center">int</td> 
   <td align="center">长度，body内容的长度</td> 
  </tr> 
  <tr> 
   <td align="center">body</td> 
   <td align="center">string</td> 
   <td align="center">内容，json序列化之后的对象</td> 
  </tr> 
 </tbody> 
</table>

刚开始使用继承`ByteToMessageDecoder`和`MessageToByteEncoder`做拆包黏包处理。  
`ByteToMessageDecoder` 抽象方法实现

public static final byte PACKAGE_TAG = 0x01;
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
            buf.markReaderIndex();
            byte tag = buf.readByte();
            if (tag != PACKAGE_TAG) {
                throw new CorruptedFrameException("标志错误");
            }
            byte encode = buf.readByte();
            byte encrypt = buf.readByte();
            int command = buf.readInt();
            int length = buf.readInt();
            byte[] data = new byte[length];
            buf.readBytes(data);
            Message message = new Message(tag, encode, encrypt, command, length,
             new String(data, "UTF-8"));
            out.add(message);
        }

`MessageToByteEncoder` 抽象方法实现

@Override
        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
            out.writeByte(MessageDecoder.PACKAGE_TAG);
            out.writeByte(msg.getEncode());
            out.writeByte(msg.getEncrypt());
            out.writeInt(msg.getCommand());
            byte[] bytes = msg.getBody().getBytes("UTF-8");
            out.writeInt(bytes.length);
            out.writeBytes(bytes);
        }

`Message.class`

public class Message { 
    
        private byte tag;
        /* 编码*/
        private byte encode;
        /*加密*/
        private byte encrypt;
        /* 类型**/
        private int command;
        /*包的长度*/
        private int length;
        /*内容*/
        private String body;
    }

这样在刚开始的工作中数据包传输没有问题，不过数据包的大小超过**512b**的时候就会抛出异常了。

> io.netty.handler.codec.DecoderException: java.lang.IndexOutOfBoundsException:  
> readerIndex(11) + length(565) exceeds writerIndex(512): PooledUnsafeDirectByteBuf(ridx: 11, widx: 512, cap: 512)

数据包的长度为565，而`ByteToMessageDecoder`只处理到了512。我并没有找到控制`ByteToMessageDecoder`最大读写的方法。  
**但是，因为解码器继承`ChannelInboundHandlerAdapter`类，而我们可以使用多个处理器一起处理数据**。

### 解决办法 ###

配合解码器`DelimiterBasedFrameDecoder`一起使用，在数据包的末尾使用换行符`\n`表示本次数据包已经结束，当`DelimiterBasedFrameDecoder`把数据切割之后，再使用`ByteToMessageDecoder`实现`decode`方法把数据流转换为`Message`对象。

我们在`ChannelPipeline`加入`DelimiterBasedFrameDecoder`解码器

在`MessageToByteEncoder`的实现方法`encode()`增加`out.writeBytes(new byte[]{'\n'});`

这时候就可以愉快的继续处理数据了。  
等我还没有高兴半天的时候，问题又来了。

> io.netty.handler.codec.DecoderException: java.lang.IndexOutOfBoundsException: readerIndex(11) + length(379) exceeds writerIndex(276): PooledUnsafeDirectByteBuf(ridx: 11, widx: 276, cap: 276)

等等等，，，怎么又报错了，不是已经加了黏包处理了吗？？，解决问题把，首先看解析的数据包结构

+-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 01 01 01 00 00 00 06 00 00 01 0a 7b 22 69 64 22 |...........{
       "id"|
    |00000010| 3a 33 2c 22 75 73 65 72 6e 61 6d 65 22 3a 22 31 |:3,"username":"1| |00000020| 38 35 30 30 33 34 30 31 36 39 22 2c 22 6e 69 63 |8500340169","nic| |00000030| 6b 6e 61 6d 65 22 3a 22 e4 bb 96 e5 9b 9b e5 a4 |kname":"........| |00000040| a7 e7 88 b7 22 2c 22 72 6f 6f 6d 49 64 22 3a 31 |....","roomId":1|
    |00000050| 35 32 37 32 33 38 35 36 39 34 37 34 2c 22 74 65 |527238569474,"te| |00000060| 61 6d 4e 61 6d 65 22 3a 22 e4 bf 84 e7 bd 97 e6 |amName":".......| |00000070| 96 af 22 2c 22 75 6e 69 74 73 22 3a 7b 22 75 6e |..","units":{
       "un| |00000080| 69 74 31 22 3a 7b 22 78 22 3a 31 30 2e 30 2c 22 |it1":{
       "x":10.0,"| |00000090| 79 22 3a 31 30 2e 30 7d 2c 22 75 6e 69 74 32 22 |y":10.0},"unit2"|
    |000000a0| 3a 7b 22 78 22 3a 31 30 2e 30 2c 22 79 22 3a 31 |:{
       "x":10.0,"y":1|
    |000000b0| 30 2e 30 7d 2c 22 75 6e 69 74 33 22 3a 7b 22 78 |0.0},"unit3":{
       "x| |000000c0| 22 3a 31 30 2e 30 2c 22 79 22 3a 31 30 2e 30 7d |":10.0,"y":10.0}|
    |000000d0| 2c 22 75 6e 69 74 34 22 3a 7b 22 78 22 3a 31 30 |,"unit4":{
       "x":10|
    |000000e0| 2e 30 2c 22 79 22 3a 31 30 2e 30 7d 2c 22 75 6e |.0,"y":10.0},"un| |000000f0| 69 74 35 22 3a 7b 22 78 22 3a 31 30 2e 30 2c 22 |it5":{
       "x":10.0,"| |00000100| 79 22 3a 31 30 2e 30 7d 7d 2c 22 73 74 61 74 75 |y":10.0}},"statu| |00000110| 73 22 3a 31 7d 0a |s":1}.          |
    +--------+-------------------------------------------------+----------------+

接收到的数据是完整的没错，但是还是报错了，而且数据结尾的字节的确是`0a`，转化成字符就是`\n`没有问题啊。

在`ByteToMessageDecoder`的`decode`方法里打印`ByteBuf buf`的长度之后，问题找到了

> 长度 : 10

这就是说在进入到`ByteToMessageDecoder`这个解码器的时候，数据包已经只剩下10个长度了，那么长的数据被上个解码器`DelimiterBasedFrameDecoder`隔空劈开了- -。问题出现在哪呢，看上面那块字节流的字节，找到第11个字节，是`0a`。。。。因为不是标准的json格式，最前面使用了3个字节 加上2个int长度的属性，所以 数据包头应该是11个字节长。  
而`DelimiterBasedFrameDecoder`在读到第11个字节的时候读成了`\n`，自然而然的就认为这个数据包已经结束了，而数据进入到`ByteToMessageDecoder`的时候就会因为规定的`body`长度不等于`length`长度而出现问题。

### 再次解决问题 ###

思来想去 不实用`\n` 这样的单字节作为换行符，很容易在数据流中遇到，转而使用`\r\n`俩字节来处理，而这俩字节出现在前面两个int长度中的几率应该很小。

看最后的代码

public class ServerInitializer extends ChannelInitializer<SocketChannel> { 
        @Override
        protected void initChannel(SocketChannel ch) {
            ChannelPipeline pipeline = ch.pipeline();
            //这里使用自定义分隔符
            ByteBuf delimiter = Unpooled.copiedBuffer("\r\n".getBytes());
            pipeline.addFirst(new DelimiterBasedFrameDecoder(8192, delimiter));
            pipeline.addLast(new MessageEncoder());
            pipeline.addLast(new MessageDecoder());
            pipeline.addLast(new MessageHandler());
        }
    }
    
    
    public class MessageEncoder extends MessageToByteEncoder<Message> { 
        @Override
        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
            out.writeByte(MessageDecoder.PACKAGE_TAG);
            out.writeByte(msg.getEncode());
            out.writeByte(msg.getEncrypt());
            out.writeInt(msg.getCommand());
            byte[] bytes = msg.getBody().getBytes("UTF-8");
            out.writeInt(bytes.length);
            out.writeBytes(bytes);
            //这里最后修改使用\r\n
            out.writeBytes(new byte[]{
       '\r','\n'});
        }
    }

再次运行程序 数据包可以正常接收了。

### 总结 ###

*  以前使用netty的时候也仅限于和硬件交互，而当时的硬件受限于成本问题是一条一条处理数据包的，所以基本上不会考虑黏包问题
 *  然后就是`ByteToMessageDecoder`和`MessageToByteEncoder`两个类是比较底层实现数据流处理的，并没有带有拆包黏包的处理机制，需要自己在数据包头规定包的长度，而且无法处理过大的数据包，因为我一开始首先使用了这种方式处理数据，所以后来就没有再换成`DelimiterBasedFrameDecoder`加 `StringDecoder`来解析数据包，最后使用json直接转化为对象。