netty处理拆包

时间  2020-05-03
标签 netty 处理 栏目 Netty 繁體版
原文   原文链接

https://blog.csdn.net/u010853261/article/details/55803933java

[netty]--最通用TCP黏包解决方案：LengthFieldBasedFrameDecoder和LengthFieldPrepender服务器

 

前面已经说过： 
TCP以流的方式进行数据传输，上层应用协议为了对消息进行区分，每每采用以下4种方式。 
（1）消息长度固定：累计读取到固定长度为LENGTH以后就认为读取到了一个完整的消息。而后将计数器复位，从新开始读下一个数据报文。app

（2）回车换行符做为消息结束符：在文本协议中应用比较普遍。socket

（3）将特殊的分隔符做为消息的结束标志，回车换行符就是一种特殊的结束分隔符。ide

（4）经过在消息头中定义长度字段来标示消息的总长度。大数据

netty中针对这四种场景均有对应的解码器做为解决方案，好比：ui

（1）经过FixedLengthFrameDecoder 定长解码器来解决定长消息的黏包问题；编码

（2）经过LineBasedFrameDecoder和StringDecoder来解决以回车换行符做为消息结束符的TCP黏包的问题；spa

（3）经过DelimiterBasedFrameDecoder 特殊分隔符解码器来解决以特殊符号做为消息结束符的TCP黏包问题；.net

（4）最后一种，也是本文的重点，经过LengthFieldBasedFrameDecoder 自定义长度解码器解决TCP黏包问题。

大多数的协议在协议头中都会携带长度字段，用于标识消息体或则整包消息的长度。LengthFieldBasedFrameDecoder经过指定长度来标识整包消息，这样就能够自动的处理黏包和半包消息，只要传入正确的参数，就能够轻松解决“读半包”的问题。

 

 

https://blog.csdn.net/bestone0213/article/details/47108419

netty处理粘包问题——1

 

咱们知道经过TCP协议发送接收数据时，若是数据过大，接收到的数据会是分包的，好比：
                                    +-----+-----+-----+
         发送数据是： | ABC | DEF | GHI |
                            +-----+-----+-----+
         而咱们想接受到的数据是： | ABCDEFGHI |
                    
该如何处理这种状况呢？Netty提供了一个专门处理TCP协议数据的Handler：LengthFieldBasedFrameDecoder ，它的原理是服务器端和客户端约定一个协议格式：数据包=协议长度+协议体
 
      --------------------------------数据包------------------------------
 
     | 协议长度部分（接收数据长度） | 协议体部分（要接收的数据）|
 
举个例子，假如咱们的TCP客户端发送了10MB字节的数据，如何让Netty服务器一次就接收到这10MB数据呢？那就须要客户端告诉服务端我发送的数据大小是多少，即在发送的数据中加入一个“数据包长度”便可，上面提到的Handler就是用来和客户端约定这个协议格式的，它有几个参数，下面我介绍一下它的参数意义：
     int maxFrameLength：定义接收数据包的最大长度，若是发送的数据包超过此值，则抛出异常；
     int lengthFieldOffset：长度属性部分的偏移值，0表示长度属性位于数据包头部；
     int lengthFieldLength：长度属性的字节长度，若是设置为4，就是咱们用4个字节存放数据包的长度；
     int lengthAdjustment：协议体长度调节值，修正信息长度，若是设置为4，那么解码时再向后推4个字节；
     int initialBytesToStrip：跳过字节数，如咱们想跳过长度属性部分。
 
2、实例-客户端发送10MB字节的数据，Netty服务端一次接收到所有10MB数据
 
客户端：定义一个消息体，用头部四个字节存放数据包长度
 
[java]  view plain copy 在CODE上查看代码片 派生到个人代码片
 
 
 
  1. public byte[] send(byte[] sendData) throws UnknownHostException, IOException {  
  2.         Socket socket = new Socket(serverIp, serverPort);  
  3.         OutputStream os = socket.getOutputStream();  
  4.         InputStream is = socket.getInputStream();  
  5.         byte resultArray[] = null;  
  6.         try {  
  7.             // 定义一个发送消息协议格式：|--header:4 byte--|--content:10MB--|  
  8.             // 获取一个4字节长度的协议体头  
  9.             byte[] dataLength = intToByteArray(4, sendData.length);  
  10.             // 和请求的数据组成一个请求数据包  
  11.             byte[] requestMessage = combineByteArray(dataLength, sendData);  
  12.             //发送数据-------------------------------  
  13.             os.write(requestMessage);  
  14.             os.flush();  
  15.             //接收数据-------------------------------  
  16.             resultArray = IOUtils.toByteArray(is);    
  17.         } catch (Exception e) {  
  18.             e.printStackTrace();  
  19.         } finally {  
  20.             os.close();  
  21.             is.close();  
  22.             socket.close();  
  23.         }  
  24.         return resultArray;  
  25.     }  
[java]  view plain copy 在CODE上查看代码片 派生到个人代码片
 
 
 
  1. private static byte[] intToByteArray(int byteLength, int intValue) {  
  2.         return ByteBuffer.allocate(byteLength).putInt(intValue).array();  
  3.     }  
  4. private static byte[] combineByteArray(byte[] array1, byte[] array2) {  
  5.         byte[] combined = new byte[array1.length + array2.length];  
  6.         System.arraycopy(array1, 0, combined, 0, array1.length);  
  7.         System.arraycopy(array2, 0, combined, array1.length, array2.length);  
  8.         return combined;  
  9.     }  
 
Netty服务端：定义一个LengthFieldBasedFrameDecoder(1024*1024*1024, 0, 4,0,4))，最大数据量是1GB,长度属性位于数据包头部，占4个字节，协议体调节值为0，跳过头部协议长度四个字节
[java]  view plain copy 在CODE上查看代码片 派生到个人代码片
 
 
 
  1. @Override  
  2.             public void initChannel(SocketChannel ch) throws Exception {  
  3.                 ChannelPipeline pipeline = ch.pipeline();  
  4.   
  5.                 pipeline.addLast("framedecoder",new LengthFieldBasedFrameDecoder(1024*1024*1024, 0, 4,0,4));  
  6.                 pipeline.addLast(new TCPServiceHandler());// 处理业务Handler  
  7.                   
  8.   
  9.             }  


3、总结：客户端和服务端定义消息格式必须一致

 

 

 

1.            frame包总体功能描述

 

此包主要做用于对TCP/IP数据包的分包和包重组，经常使用于数据的流传输，是扩展的解码器。

包目录结构以下：

netty 数据分包、组包、粘包处理机制（一） - 断鸿零雁 - 断鸿零雁的博客
 

 

2.            包中各种功能详解

(1)  FrameDecoder

抽象类，将ChannelBuffers中的二进制数据转换成有意义的数据帧（frame）对象，通常不直接调用，提供给此包中的FixedLengthFrameDecoder类、DelimiterBasedFrameDecoder类和LengthFieldBasedFrameDecoder类使用，也能够提供给其余类使用（暂不探讨）；

 

在数据传输中，咱们发送的数据包以下所示

 

+-----+-----+-----+

 | ABC | DEF | GHI |

 +-----+-----+-----+

 

而实际接收的包的格式为：

 

+----+-------+---+---+
| AB | CDEFG | H | I |
+----+-------+---+---+

 

 

产生的缘由为：数据在传输过程当中，产生数据包碎片（TCP/IP数据传输时大数据包没法一次传输，被拆分红小数据包，小数据包即为数据包碎片），这就形成了实际接收的数据包和发送的数据包不一致的状况。

 

而经过FrameDecoder便可实现对上述接收到的数据包的整理，从新还原成以下格式：

 

+-----+-----+-----+
| ABC | DEF | GHI |
+-----+-----+-----+

 

以下是一个自定义的Decoder类

 

public class MyFrameDecoder extends FrameDecoder {

 

         @Override

   protected Object decode(ChannelHandlerContext ctx,

                           channel,

                           ChannelBuffer buf) throws Exception {

 

     // Make sure if the length field was received.

     if (buf.readableBytes() < 4) {

        // The length field was not received yet - return null.

        // This method will be invoked again when more packets are

        // received and appended to the buffer.

        return null;

     }

 

     // The length field is in the buffer.

 

     // Mark the current buffer position before reading the length field

     // because the whole frame might not be in the buffer yet.

     // We will reset the buffer position to the marked position if

     // there's not enough bytes in the buffer.

     buf.markReaderIndex();
 

     // Read the length field.

     int length = buf.readInt();
 

     // Make sure if there's enough bytes in the buffer.

     if (buf.readableBytes() < length) {

        // The whole bytes were not received yet - return null.

        // This method will be invoked again when more packets are

        // received and appended to the buffer.

        // Reset to the marked position to read the length field again

        // next time.

        buf.resetReaderIndex();
 
        return null;

     }

 

     // There's enough bytes in the buffer. Read it.

     ChannelBuffer frame = buf.readBytes(length);
 

     // Successfully decoded a frame.  Return the decoded frame.

     return frame;

   }

 }

 

此时，咱们无需关注数据包是如何重组的，只须要作简单的验证（按照一个包验证）就能够了，FrameDecoder内部实现了组包的机制，不过，此时，需在数据的最前面封装整个数据的长度，示例中数据长度占了四个字节，即前四个字节是数据长度，后面的才是真实的数据。

(2)  FixedLengthFrameDecoder

FixedLengthFrameDecoder主要是将诸如

+----+-------+---+---+
| AB | CDEFG | H | I |
+----+-------+---+---+

 

此类的数据包按照指定的frame长度从新组包，好比肯定长度为3，则组包为

 

+-----+-----+-----+
| ABC | DEF | GHI |
+-----+-----+-----+

 

构造方法为：new FixedLengthFrameDecoder(int frameLength);

 

frameLength即修正后的帧长度

 

另外一个构造方法为new FixedLengthFrameDecoder(int frameLength, boolean allocateFullBuffer);

 

allocateFullBuffer若是为真，则表示初始化的ChannelBuffer大小为frameLength。

 

 

(3)  Delimiters

分隔符类，DelimiterBasedFrameDecoder类的辅助类。

 

对Flash XML的socket通讯采用nulDelimiter()方法，对于通常的文本采用lineDelimiter()方法

 

(4)  DelimiterBasedFrameDecoder

对接收到的ChannelBuffers按照指定的分隔符Delimiter分隔，分隔符能够是一个或者多个

 

如将如下数据包按照“\n”分隔：

 
+--------------+
| ABC\nDEF\r\n |
+--------------+

 

即为：

 

+-----+-----+
| ABC | DEF |
+-----+-----+
 

而若是按照“\r\n”分隔，则为：

 

+----------+
| ABC\nDEF |
+----------+

 

对于DelimiterBasedFrameDecoder中的构造方法，其中一些参数说明：

 

maxFrameLength：解码的帧的最大长度

stripDelimiter：解码时是否去掉分隔符

failFast：为true，当frame长度超过maxFrameLength时当即报TooLongFrameException异常，为false，读取完整个帧再报异常

delimiter：分隔符

(5)  LengthFieldBasedFrameDecoder

经常使用的处理大数据分包传输问题的解决类，先对构造方法LengthFieldBasedFrameDecoder中的参数作如下解释说明“

maxFrameLength：解码的帧的最大长度

lengthFieldOffset ：长度属性的起始位（偏移位），包中存放有整个大数据包长度的字节，这段字节的其实位置

lengthFieldLength：长度属性的长度，即存放整个大数据包长度的字节所占的长度

lengthAdjustmen：长度调节值，在总长被定义为包含包头长度时，修正信息长度。initialBytesToStrip：跳过的字节数，根据须要咱们跳过lengthFieldLength个字节，以便接收端直接接受到不含“长度属性”的内容

failFast ：为true，当frame长度超过maxFrameLength时当即报TooLongFrameException异常，为false，读取完整个帧再报异常

 

下面对各类状况分别描述：

 

1. 2 bytes length field at offset 0, do not strip header

 

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" |

 +--------+----------------+      +--------+----------------+

 

此时数据格式不作任何改变（没有跳过任何字节）

 

2. 2 bytes length field at offset 0, strip header

 

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" |
 +--------+----------------+      +----------------+

 

此时帧长度为14个字节，但因为前（lengthFieldOffset = 0）两个（lengthFieldLength = 2）字节是表示帧长度的字节，不计入数据，故真实的数据长度为12个字节。

 

3. 2 bytes length field at offset 0, do not strip header, the length field represents the length of the whole message

 

lengthFieldOffset   =  0

 lengthFieldLength   =  2

 lengthAdjustment    = -2 (= the length of the Length field)

 initialBytesToStrip =  0

 

 BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
 +--------+----------------+     +--------+----------------+
 | Length | Actual Content |---->| Length | Actual Content |
 | 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
 +--------+----------------+     +--------+----------------+
 

此处定义的Length为0x000E共占了两个字节，表示的帧长度为14个字节，前（lengthFieldOffset = 0）两个（lengthFieldLength = 2）字节为Length，因为设置的lengthAdjustment    = -2 (= the length of the Length field)，故修正的信息实际长度补2，即解码时往前推2个字节，解码后仍是14个字节长度（此种状况是把整个长度封装，通常来说，咱们只封装数据长度）

 

4. 3 bytes length field at the end of 5 bytes header, do not strip header

lengthFieldOffset   = 2 (= the length of Header 1)

 lengthFieldLength   = 3

 lengthAdjustment    = 0

 initialBytesToStrip = 0

 

BEFORE 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   = 2，从第3个字节开始表示数据长度，长度占3个字节，真实数据长度为0x00000C 即12个字节，而lengthAdjustment=0，initialBytesToStrip = 0，故解码后的数据与解码前的数据相同。

4. 3 bytes length field at the beginning of 5 bytes header, do not strip header

lengthFieldOffset   = 0

 lengthFieldLength   = 3

 lengthAdjustment    = 2 (= the length of Header 1)

 initialBytesToStrip = 0

 

BEFORE 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" |
 +----------+----------+----------------+      +----------+----------+----------------+

此处因为修正的字节数是2，且initialBytesToStrip = 0，故整个数据的解码数据保持不变

总字节数是17，开始的三个字节表示字节长度：12，修正的字节是2,（即从第三个字节开始，再加两个开始是真正的数据，其中跳过的字节数是0）

 

5. 2 bytes length field at offset 1 in the middle of 4 bytes header, strip the first header field and the length field

 

 

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" |

 +------+--------+------+----------------+      +------+----------------+

 

从第2个字节开始解码，取两个字节做为帧长度，为12个字节，而后，修正一个字节，从第5个字节到最后表示帧数据，解码时，因为initialBytesToStrip=3，表示跳过前三个字节（去掉），故从第四个字节开始解析，解析出来后，如右图所示。

 

6. 2 bytes length field at offset 1 in the middle of 4 bytes header, strip the first header field and the length field, the length field represents the length of the whole message

 

lengthFieldOffset   =  1

 lengthFieldLength   =  2

 lengthAdjustment    = -3 (= the length of HDR1 + LEN, negative)

 initialBytesToStrip =  3

 

BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)

 +------+--------+------+----------------+      +------+----------------+

 | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |

 | 0xCA | 0x0010 | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |

 +------+--------+------+----------------+      +------+----------------+

 

从第二个字节开始，取两个字节做为帧长度，为16个字节，而后补3个字节，故往前找三个字节，从HDP1开始解码，而又由于initialBytesToStrip=3，解码时忽略掉前三个字节，故从第四个字节开始解析，解析结果如右图所示。

 

总结：通常来说，当lengthAdjustment 为负数时，Length表示的是整个帧的长度，当lengthAdjustment为正数或0时，表示真实数据长度。

(6)  LengthFieldPrepender

编码类，自动将

+----------------+

| "HELLO, WORLD" |

+----------------+

格式的数据转换成

+--------+----------------+

 + 0x000C | "HELLO, WORLD" |

 +--------+----------------+

格式的数据，

若是lengthIncludesLengthFieldLength设置为true,则编码为

+--------+----------------+

+ 0x000E | "HELLO, WORLD" |

+--------+----------------+

格式的数据

 

应用场景：自定义pipelineFactory类: MyPipelineFactory implements ChannelPipelineFactory

中

pipeline.addLast("frameEncode", new LengthFieldPrepender(4, false));

(7)  TooLongFrameException

定义的数据包超过预约义大小异常类

(8)  CorruptedFrameException

定义的数据包损坏异常类

3.            frame包应用demo

解决分包问题，一般配置MyPipelineFactory中设置，示例以下：

 

public class MyPipelineFactory implements ChannelPipelineFactory {

 

    @Override

    public ChannelPipeline getPipeline() throws Exception {

       ChannelPipeline pipeline = Channels.pipeline();

       pipeline.addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); 

        pipeline.addLast("encoder", new LengthFieldPrepender(4, false));

       pipeline.addLast("handler", new MyHandler());

       return pipeline;

    }

 

}

 

 

在客户端设置pipeline.addLast("encoder", new LengthFieldPrepender(4, false));

       pipeline.addLast("handler", new MyHandler());

 

前四个字节表示真实的发送的数据长度Length，编码时会自动加上；

 

在服务器端设置pipeline.addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));

真实数据最大字节数为Integer.MAX_VALUE，解码时自动去掉前面四个字节

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程