Netty处理消息的粘包和拆包

netty socket 拆包 粘包
5天前

Netty通过固定的消息头处理消息的粘包和拆包

对于消息的拆包和粘包，通用的做法就是用一个消息头部来描述消息内容的长度。Netty提供了一个LengthFieldBasedFrameDecoder Decoder，可以非常方便的去处理各种带有长度头的私有协议。

LengthFieldBasedFrameDecoder

构造函数

  1. public LengthFieldBasedFrameDecoder(int maxFrameLength,int lengthFieldOffset, int lengthFieldLength)
  2. public LengthFieldBasedFrameDecoder(int maxFrameLength,int lengthFieldOffset, int lengthFieldLength,int lengthAdjustment, int initialBytesToStrip)
  3. public LengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,int lengthAdjustment, int initialBytesToStrip, boolean failFast)
  4. public LengthFieldBasedFrameDecoder(ByteOrder byteOrder, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,int lengthAdjustment, int initialBytesToStrip, boolean failFast)

  • byteOrder

    • 网络字节序

  • maxFrameLength

    • 单个包最大的长度,这个值根据实际场景而定

  • lengthFieldOffset

    • 表示数据长度字段开始的偏移量
    • 第几个字节开始,是表示数据长度的(0)

  • lengthFieldLength

    • 数据长度字段的所占的字节数

  • lengthAdjustment

    • 添加到长度字段的补偿值(0)
    • lengthAdjustment + 数据长度取值 = 数据长度字段之后剩下包的字节数
    • 对于某些协议,长度字段还包含了消息头的长度,在这种应用场景中,往往需要使用lengthAdjustment进行修正,由于整个消息(包含消息头)的长度往往大于消息体的长度,所以它要设置为负数(数据长度字段的长度取负)

  • initialBytesToStrip

    • 表示从整个包第一个字节开始,向后忽略的字节数(0)
    • 可以设置该参数,来忽略掉包头信息,仅仅留下数据包体,给下一个Handler处理

各种不同的协议，与LengthFieldBasedFrameDecoder的实现

在与LengthFieldBasedFrameDecoder的官方doc中就有各种协议及其实现的详细描述

  1. # 协议1
  2. lengthFieldOffset   = 0
  3. lengthFieldLength   = 2        // 通用的前面2字节表示数据长度
  4. lengthAdjustment    = 0
  5. initialBytesToStrip = 0 
  7. BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
  8. +--------+----------------+      +--------+----------------+
  9. | Length | Actual Content |----->| Length | Actual Content |
  10. | 0x000C | "HELLO, WORLD" |      | 0x000C | "HELLO, WORLD" |
  11. +--------+----------------+      +--------+----------------+
  13. # 协议2
  14. lengthFieldOffset   = 0
  15. lengthFieldLength   = 2
  16. lengthAdjustment    = 0
  17. initialBytesToStrip = 2    // 丢弃前面2字节的包头
  19. BEFORE DECODE (14 bytes)         AFTER DECODE (12 bytes)
  20. +--------+----------------+      +----------------+
  21. | Length | Actual Content |----->| Actual Content |
  22. | 0x000C | "HELLO, WORLD" |      | "HELLO, WORLD" |
  23. +--------+----------------+      +----------------+
  25. # 协议3
  26. lengthFieldOffset   =  0
  27. lengthFieldLength   =  2
  28. lengthAdjustment    = -2 // 长度头表示的长度,包含了自身头部的长度
  29. initialBytesToStrip =  0
  31. BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
  32. +--------+----------------+      +--------+----------------+
  33. | Length | Actual Content |----->| Length | Actual Content |
  34. | 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
  35. +--------+----------------+      +--------+----------------+
  37. # 协议4
  38. lengthFieldOffset   = 2    // 表示消息长度的头,不在首部
  39. lengthFieldLength   = 3
  40. lengthAdjustment    = 0
  41. initialBytesToStrip = 0
  43. BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
  44. +----------+----------+----------------+      +----------+----------+----------------+
  45. | Header 1 |  Length  | Actual Content |----->| Header 1 |  Length  | Actual Content |
  46. |  0xCAFE  | 0x00000C | "HELLO, WORLD" |      |  0xCAFE  | 0x00000C | "HELLO, WORLD" |
  47. +----------+----------+----------------+      +----------+----------+----------------+
  49. # 协议5
  50. lengthFieldOffset   = 0
  51. lengthFieldLength   = 3
  52. lengthAdjustment    = 2    // 整个消息体的长度,还要包含一个头部的长度,因为后面还有一个头部
  53. initialBytesToStrip = 0
  55. BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
  56. +----------+----------+----------------+      +----------+----------+----------------+
  57. |  Length  | Header 1 | Actual Content |----->|  Length  | Header 1 | Actual Content |
  58. | 0x00000C |  0xCAFE  | "HELLO, WORLD" |      | 0x00000C |  0xCAFE  | "HELLO, WORLD" |
  59. +----------+----------+----------------+      +----------+----------+----------------+
  61. # 协议6
  62. lengthFieldOffset   = 1 (= the length of HDR1)            // 第2个字节表示数据长度
  63. lengthFieldLength   = 2                                
  64. lengthAdjustment    = 1 (= the length of HDR2)            // 除此之外,还有1个字节的消息头
  65. initialBytesToStrip = 3 (= the length of HDR1 + LEN)    // 移除前面3个字节的数据
  67. BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
  68. +------+--------+------+----------------+      +------+----------------+
  69. | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
  70. | 0xCA | 0x000C | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
  71. +------+--------+------+----------------+      +------+----------------+
  73. # 协议7
  74. lengthFieldOffset   =  1
  75. lengthFieldLength   =  2                                
  76. lengthAdjustment    = -3 (= the length of HDR1 + LEN, negative)    //长度头表示的是整个消息体的长度
  77. initialBytesToStrip =  3
  79. BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
  80. +------+--------+------+----------------+      +------+----------------+
  81. | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
  82. | 0xCA | 0x0010 | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
  83. +------+--------+------+----------------+      +------+----------------+

Demo

服务端

  1. import java.net.InetSocketAddress;
  2. import java.nio.charset.StandardCharsets;
  3. import java.security.cert.CertificateException;
  5. import javax.net.ssl.SSLException;
  7. import io.netty.bootstrap.ServerBootstrap;
  8. import io.netty.buffer.ByteBuf;
  9. import io.netty.channel.ChannelFuture;
  10. import io.netty.channel.ChannelHandlerContext;
  11. import io.netty.channel.ChannelInitializer;
  12. import io.netty.channel.ChannelOption;
  13. import io.netty.channel.ChannelPipeline;
  14. import io.netty.channel.EventLoopGroup;
  15. import io.netty.channel.SimpleChannelInboundHandler;
  16. import io.netty.channel.nio.NioEventLoopGroup;
  17. import io.netty.channel.socket.SocketChannel;
  18. import io.netty.channel.socket.nio.NioServerSocketChannel;
  19. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  21. public class Server {
  23.     public static void main(String[] args) throws CertificateException, SSLException, InterruptedException {
  25.         EventLoopGroup bossEventLoopGroup = new NioEventLoopGroup();
  26.         EventLoopGroup workerEventLoopGroup = new NioEventLoopGroup();
  27.         try {
  28.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  29.             serverBootstrap.group(bossEventLoopGroup, workerEventLoopGroup);
  30.             serverBootstrap.channel(NioServerSocketChannel.class);
  31.             serverBootstrap.option(ChannelOption.SO_REUSEADDR, true);
  33.             serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
  34.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  35.                 @Override
  36.                 protected void initChannel(SocketChannel ch) throws Exception {
  37.                     ChannelPipeline channelPipeline = ch.pipeline();
  38.                     // 头偏移:0,头长度:4,长度补充:0,丢弃:4
  39.                     channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
  40.                     channelPipeline.addLast(new SimpleChannelInboundHandler<ByteBuf>() {
  41.                         @Override
  42.                         protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  43.                             // msg里面只剩下消息体
  44.                             System.out.println(msg.toString(StandardCharsets.UTF_8));
  45.                         }
  46.                     });
  47.                 }
  48.             });
  50.             ChannelFuture channelFuture = serverBootstrap.bind(new InetSocketAddress(1024)).sync();
  51.             channelFuture.channel().closeFuture().sync();
  52.         } finally {
  53.             bossEventLoopGroup.shutdownGracefully();
  54.             workerEventLoopGroup.shutdownGracefully();
  55.         }
  56.     }
  57. }

客户端

  1. import socket
  2. import struct
  4. connector = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
  5. connector.connect(('127.0.0.1', 1024))
  7. # 对消息编码,添加4个字节长消息头
  8. def encode(data):
  9.     data = bytes(data,'UTF_8')
  10.     length = len(data)
  11.     return struct.pack('>I%ds'%(length),length,data)
  13. # 可以对消息进行解码,分离消息头和消息
  14. def decode(data):
  15.     length = len(data) - 4
  16.     return struct.unpack('>I%ds'%(length),data)
  18. while True:
  19.     connector.send(encode(input('输入消息内容\n')))

成功解决了消息的粘包和拆包


Netty提供的其他一些解决方案

LineBasedFrameDecoder

根据换行符来分割消息

DelimiterBasedFrameDecoder

根据指定分隔符来分割消息

FixedLengthFrameDecoder

固定消息的长度

0
0
0

添加评论

正在回复：
取消
2
0
0
0