• Netty 进阶学习(九)-- 粘包与半包


    1、粘包与半包

    服务器端代码:

    @Slf4j
    public class HelloWorldServer {
        public static void main(String[] args) {
            start();
        }
        public static void start() {
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup worker = new NioEventLoopGroup();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.channel(NioServerSocketChannel.class);
                serverBootstrap.group(boss, worker);
                serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    }
                });
                ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("server error", e);
            }finally {
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }
    }
    
    • 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

    客户端代码:

    public class HelloWorldClient {
        private static Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
    
        public static void main(String[] args) {
            NioEventLoopGroup worker = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(worker);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override   // Channel 建立连接成功后触发
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            }
                        });
                    }
                });
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("client error", e);
            } finally {
                worker.shutdownGracefully();
            }
        }
    }
    
    • 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
    1.1、粘包现象

    客户端发送 1016 字节的数据:

    @Override   // Channel 建立连接成功后触发
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 10; i++) {
            ByteBuf buffer = ctx.alloc().buffer(16);
            buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
            ctx.writeAndFlush(buffer);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    服务端一次性接收到了 160 字节的数据:

    在这里插入图片描述

    1.2、半包现象

    服务端增加代码:

    // 设置TCP接收缓冲区的大小 10字节
    serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
    
    • 1
    • 2

    客户端发送 516 字节的数据,查看结果,存在接收到 4 字节的情况:

    在这里插入图片描述

    1.3、滑动窗口
    • TCP 以一个段 (segment)为单位,每发送一个段就需要进行一次确认应答(ack)处理,但如果这么做,缺点是包往返时间越长,性能越差。
    • 为了解决此问题,引入了窗口概念,窗口大小决定了无需等待应答而可以继续发送数据的最大值
    • 窗口实际起到一个缓冲区的作用,同时也能起到流量控制的作用

    滑动窗口

    • TCP 以一个段(segment)为单位,每发送一个段就需要进行一次确认应答(ack)处理,但如果这么做,缺点是包的往返时间越长性能就越差
      在这里插入图片描述

    • 为了解决此问题,引入了窗口概念,窗口大小即决定了无需等待应答而可以继续发送的数据最大值

      在这里插入图片描述

    • 窗口实际就起到一个缓冲区的作用,同时也能起到流量控制的作用

      • 图中深色的部分即要发送的数据,高亮的部分即窗口
      • 窗口内的数据才允许被发送,当应答未到达前,窗口必须停止滑动
      • 如果 1001~2000 这个段的数据 ack 回来了,窗口就可以向前滑动
      • 接收方也会维护一个窗口,只有落在窗口内的数据才能允许接收

    MSS 限制

    • 链路层对一次能够发送的最大数据有限制,这个限制称之为 MTU(maximum transmission unit),不同的链路设备的 MTU 值也有所不同,例如

    • 以太网的 MTU 是 1500

    • FDDI(光纤分布式数据接口)的 MTU 是 4352

    • 本地回环地址的 MTU 是 65535 - 本地测试不走网卡

    • MSS 是最大段长度(maximum segment size),它是 MTU 刨去 tcp 头和 ip 头后剩余能够作为数据传输的字节数

    • ipv4 tcp 头占用 20 bytes,ip 头占用 20 bytes,因此以太网 MSS 的值为 1500 - 40 = 1460

    • TCP 在传递大量数据时,会按照 MSS 大小将数据进行分割发送

    • MSS 的值在三次握手时通知对方自己 MSS 的值,然后在两者之间选择一个小值作为 MSS

    在这里插入图片描述

    Nagle 算法

    • 即使发送一个字节,也需要加入 tcp 头和 ip 头,也就是总字节数会使用 41 bytes,非常不经济。因此为了提高网络利用率,tcp 希望尽可能发送足够大的数据,这就是 Nagle 算法产生的缘由
    • 该算法是指发送端即使还有应该发送的数据,但如果这部分数据很少的话,则进行延迟发送
      • 如果 SO_SNDBUF 的数据达到 MSS,则需要发送
      • 如果 SO_SNDBUF 中含有 FIN(表示需要连接关闭)这时将剩余数据发送,再关闭
      • 如果 TCP_NODELAY = true,则需要发送
      • 已发送的数据都收到 ack 时,则需要发送
      • 上述条件不满足,但发生超时(一般为 200ms)则需要发送
      • 除上述情况,延迟发送
    1.4、现象分析
    1)粘包
    • 现象:发送abcedf 接收到 abcdef
    • 原因:
      • 应用层:接收方 ByteBuf 设置太大(Netty 默认 1024)
      • 滑动窗口:假设发送方 256 byte 表示一个完整的报文,但由于接收方处理不及时且窗口大小足够大,这 256 byte 字节就会缓冲在接收方的滑动窗口中,当滑动窗口中缓冲了多个报文就会粘包
      • Nagle 算法:会造成粘包(尽可能多的发送数据)
    2)半包
    • 现象:发送 abcdef 接收到 abcedf
    • 原因:
      • 应用层:接收方的 ByteBuf 小于实际发送数据量
      • 滑动窗口:假设接收方的窗口只剩 128 byte,发送方的报文大小是256 byte,这时放不下了,只能发送前 128 byte,等待 ack 后才能发送剩余部分,就会造成了半包
      • MSS 限制:但发送的数据超过 MSS 限制后,会将数据切分发送,就会造成半包(传输层的报文荷载长度(不包含报头))
    3)本质

    本质是因为 TCP 是流式协议,消息无边界。

    1.5、粘包解决
    1)短连接

    解决不了半包问题,效率低。消息边界是建立连接和断开连接。TCP 的连接和断开都涉及三次握手和四次挥手,十分浪费性能,不推荐使用。

    客户端代码:

    public class HelloWorldClient {
        private static Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
    
        public static void main(String[] args) {
            for (int i = 0; i < 10; i++) {
                send();
            }
            System.out.println("finnish");
        }
    
        private static void send() {
            NioEventLoopGroup worker = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(worker);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override   // Channel 建立连接成功后触发
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                ByteBuf buffer = ctx.alloc().buffer(16);
                                buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
                                ctx.writeAndFlush(buffer);
                                ctx.channel().close();
                            }
                        });
                    }
                });
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("client error", e);
            } finally {
                worker.shutdownGracefully();
            }
        }
    }
    
    • 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
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    2)定长解码器

    封装成帧(Framing), 自定义边界。

    消息边界也就是固定长度,这种方式实现简单,但是空间有极大的浪费,不推荐使用。

    /**
     * A decoder that splits the received ByteBufs by the fixed number
     * of bytes. For example, if you received the following four fragmented packets:
     * 
     * +---+----+------+----+
     * | A | BC | DEFG | HI |
     * +---+----+------+----+
     * 
     * A FixedLengthFrameDecoder(3) will decode them into the
     * following three packets with the fixed length:
     * 
     * +-----+-----+-----+
     * | ABC | DEF | GHI |
     * +-----+-----+-----+
     */
    public class FixedLengthFrameDecoder extends ByteToMessageDecoder {}
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    编辑代码测试

    编辑客户端代码:

    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));	// 日志输出
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override   // Channel 建立连接成功后触发
                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                    ByteBuf byteBuf = ctx.alloc().buffer();
                    char c = '0';
                    Random random = new Random();
                    for (int i = 0; i < 10; i++) {
                        // 发送 10 次消息,每次长度随机 1~10 个字节
                        byte[] bytes = fill10Bytes(c++, random.nextInt(10) + 1);
                        byteBuf.writeBytes(bytes);
                    }
                    ctx.writeAndFlush(byteBuf);
                }
            });
        }
    });
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    public static byte[] fill10Bytes(char ch, int len){
        byte[] bytes = new byte[10];
        Arrays.fill(bytes, (byte) '_');
        for (int i = 0; i < len; i++) {
            bytes[i] = (byte) ch;
        }
        System.out.println(new String(bytes));
        return bytes;
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    编辑服务端代码:

    serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            // 注意:定长解码器要放在日志输出处理器的前面
            ch.pipeline().addLast(new FixedLengthFrameDecoder(10));			
            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
        }
    });
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    测试结果

    在这里插入图片描述

    在这里插入图片描述

    3)行解码器

    封装成帧(Framing), 自定义边界。

    消息边界就是分隔符,不再浪费空间,但是内容使用分隔符就需要进行转义,效率也不是很高。

    /**
     * A decoder that splits the received ByteBufs on line endings.
     * 
     * Both "\n" and "\r\n" are handled.
     * For a more general delimiter-based decoder, see DelimiterBasedFrameDecoder.
     */
    public class LineBasedFrameDecoder extends ByteToMessageDecoder {
        public LineBasedFrameDecoder(final int maxLength) {	// 需要指定最大长度,不能收不到分隔符就一直接收下去
            this(maxLength, true, false);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    /**
     * A decoder that splits the received ByteBufs by one or more
     * delimiters.  It is particularly useful for decoding the frames which ends
     * with a delimiter such as {Delimiters#nulDelimiter() NUL} or
     * {Delimiters#lineDelimiter() newline characters}.
     *
     * 

    Predefined delimiters

    *

    * Delimiters defines frequently used delimiters for convenience' sake. * *

    Specifying more than one delimiter

    *

    * DelimiterBasedFrameDecoder allows you to specify more than one * delimiter. If more than one delimiter is found in the buffer, it chooses * the delimiter which produces the shortest frame. For example, if you have * the following data in the buffer: * * +--------------+ * | ABC\nDEF\r\n | * +--------------+ * * a DelimiterBasedFrameDecoder (Delimiters.lineDelimiter()) * will choose '\n' as the first delimiter and produce two frames: * * +-----+-----+ * | ABC | DEF | * +-----+-----+ * * rather than incorrectly choosing '\r\n' as the first delimiter: * * +----------+ * | ABC\nDEF | * +----------+ * */ public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf delimiter) { // 可以指定 ByteBuf this(maxFrameLength, true, delimiter); } }

    • 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
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40

    编辑代码测试

    编辑客户端代码:

    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override   // Channel 建立连接成功后触发
                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                    ByteBuf byteBuf = ctx.alloc().buffer();
                    char c = '0';
                    Random random = new Random();
                    for (int i = 0; i < 10; i++) {
                        StringBuilder sb = makeString(c++, random.nextInt(256) + 1);
                        byteBuf.writeBytes(sb.toString().getBytes());
                    }
                    ctx.writeAndFlush(byteBuf);
                }
            });
        }
    });
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    public static StringBuilder makeString(char ch, int len){
        StringBuilder sb = new StringBuilder(len);
        for (int i = 0; i < len; i++) {
            sb.append(ch);
        }
        sb.append('\n');
        return sb;
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    编辑服务端代码:

    serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
        }
    });
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    测试结果

    在这里插入图片描述

    4)LTC 解码器

    封装成帧(Framing), 自定义边界。

    基于长度字段的解码器。有一个专门 length 字段,就能知道内容的长度。

    public class LengthFieldBasedFrameDecoder extends ByteToMessageDecoder {
        private final ByteOrder byteOrder;
        private final int maxFrameLength;			// 帧的最大长度
        private final int lengthFieldOffset;		// 长度字段的偏移量
        private final int lengthFieldLength;		// 长度字段的长度
        private final int lengthFieldEndOffset;		
        private final int lengthAdjustment;			// 长度字段为基准,还有几个字节是内容
        private final int initialBytesToStrip;		// 从头剥离几个字节
        private final boolean failFast;
        private boolean discardingTooLongFrame;
        private long tooLongFrameLength;
        private long bytesToDiscard;
        
        public LengthFieldBasedFrameDecoder(
            int maxFrameLength,
            int lengthFieldOffset, int lengthFieldLength,
            int lengthAdjustment, int initialBytesToStrip) {
            this(
                maxFrameLength,
                lengthFieldOffset, lengthFieldLength, 
                lengthAdjustment,initialBytesToStrip, true);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    源码注释案例

    在这里插入图片描述

    测试

    /**
     * @desc
     * @auth llp
     * @date 2022/8/9 11:45
     */
    public class TestLengthFieldDecoder {
        public static void main(String[] args) {
            EmbeddedChannel channel = new EmbeddedChannel(
                    new LengthFieldBasedFrameDecoder(1024, 0, 4, 1, 0),
                    new LoggingHandler(LogLevel.DEBUG)
            );
            // 4 个字节的内容长度, 实际内容
            ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
            send(byteBuf, "Hello, World");
            send(byteBuf, "Hi!");
            channel.writeInbound(byteBuf);
        }
    
        private static void send(ByteBuf byteBuf, String content) {
            // 实际内容
            byte[] bytes = content.getBytes();
            // 实际内容长度
            int length = bytes.length;
            byteBuf.writeInt(length);
            byteBuf.writeByte(1);   // 例如版本号
            byteBuf.writeBytes(bytes);
        }
    }
    
    • 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

    在这里插入图片描述

  • 相关阅读:
    腾讯安全在2022:出租车、地铁和爆发的火山
    代码随想录训练营 DP
    eax和ax、ah、al的区别简介
    Nginx学习(12)—— 配置指令整理
    C++【个人笔记1】
    C++11中篇
    Flink容错机制
    FFmpeg v4l2m2m的capture和output
    离线语音与IoT结合:智能家居发展新增长点
    MemFire Cloud: 一种全新定义后端即服务的解决方案
  • 原文地址:https://blog.csdn.net/weixin_43989102/article/details/126756167