Netty之TCP粘包/拆包问题（四）

落日映苍穹つ 2023-03-01 10:59 3阅读 0赞

### 文章目录 ###

*   *  粘包与拆包
     *  代码
     *  分析
     *  解决
     *  分析
     *  思考

## 粘包与拆包 ##

但凡是使用的TCP协议的都会遇到这个问题，TCP是个“流”协议，所谓流，就是没有界限的一串数据。大家可以想想河里的流水，是连成一片的，其间并没有分界线。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。  
举个栗子：  
本来客户端要想服务端发送两条数据分别是：第一条：11，第二条22，会发生以下几种情况：

1.  服务端接受到了两个数据包11,22，即没有发生粘包和拆包；
2.  服务端接受到了一个数据包：1122，发生了粘包；
3.  服务端接受到了两个数据包：1,122，发生了拆包；
4.  服务端接收到了四个数据包：1，1，2，2，发生了多次拆包；

原因：

*  应用程序 write写入的字节大小大于套接口发送缓冲区大小
 *  进行MSS大小的TCP分段
 *  以太网帧的 pay load大于MTU进行IP分片。

解决：

*  消息定长，例如每个报文的大小为固定长度200字节，如果不够，空位补空格
 *  在包尾增加回车换行符进行分割，例如FTP协议
 *  将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）  
    的字段，通常设计思路为消息头的第一个字段使用int32来表示消息的总长度；
 *  更复杂的应用层协议。

## 代码 ##

package main.com.phei.netty.frame.fault;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeServer { 
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new TimeServer().bind(port);
        }
    
        public void bind(int port) throws Exception { 
            // 配置服务端的NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try { 
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChildChannelHandler());
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
            @Override
            protected void initChannel(SocketChannel arg0) throws Exception { 
                arg0.pipeline().addLast(new TimeServerHandler());
            }
    
        }
    }

package main.com.phei.netty.frame.fault;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeClient { 
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new TimeClient().connect(port, "127.0.0.1");
        }
    
        public void connect(int port, String host) throws Exception { 
            // 配置客户端NIO线程组
            EventLoopGroup group = new NioEventLoopGroup();
            try { 
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            public void initChannel(SocketChannel ch)
                                    throws Exception { 
    
                                ch.pipeline().addLast(new TimeClientHandler());
                            }
                        });
    
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
    
                // 当代客户端链路关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    }

package main.com.phei.netty.frame.fault;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeServerHandler extends ChannelHandlerAdapter { 
    
        private int counter;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "UTF-8").substring(0, req.length
                    - System.getProperty("line.separator").length());
            System.out.println("The time server receive order : " + body
                    + " ; the counter is : " + ++counter);
            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
                    System.currentTimeMillis()).toString() : "BAD ORDER";
            currentTime = currentTime + System.getProperty("line.separator");
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.writeAndFlush(resp);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            ctx.close();
        }
    }

package main.com.phei.netty.frame.fault;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.util.logging.Logger;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeClientHandler extends ChannelHandlerAdapter { 
    
        private static final Logger logger = Logger
                .getLogger(TimeClientHandler.class.getName());
    
        private int counter;
    
        private byte[] req;
    
        /** * Creates a client-side handler. */
        public TimeClientHandler() { 
            req = ("QUERY TIME ORDER" + System.getProperty("line.separator"))
                    .getBytes();
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            ByteBuf message = null;
            for (int i = 0; i < 100; i++) { 
                message = Unpooled.buffer(req.length);
                message.writeBytes(req);
                ctx.writeAndFlush(message);
            }
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "UTF-8");
            System.out.println("Now is : " + body + " ; the counter is : "
                    + ++counter);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            // 释放资源
            logger.warning("Unexpected exception from downstream : "
                    + cause.getMessage());
            ctx.close();
        }
    }

运行结果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]

## 分析 ##

代码里面使用Netty从客户端想服务端发送了100条QUERY TIME ORDER，然后服务端接受到一条消息就打印一句The time server receive order : ; the counter is : ，然后我们发现服务端只打印了两条数据，也就是说发生改了粘包，而客户端也打印了一条，也发生了粘包。

## 解决 ##

package main.com.phei.netty.frame.correct;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeServer { 
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new TimeServer().bind(port);
        }
    
        public void bind(int port) throws Exception { 
            // 配置服务端的NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try { 
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChildChannelHandler());
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
            @Override
            protected void initChannel(SocketChannel arg0) throws Exception { 
                arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
                arg0.pipeline().addLast(new StringDecoder());
                arg0.pipeline().addLast(new TimeServerHandler());
            }
        }
    }

package main.com.phei.netty.frame.correct;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeClient { 
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
            int port = 8080;
            if (args != null && args.length > 0) { 
                try { 
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) { 
                    // 采用默认值
                }
            }
            new TimeClient().connect(port, "127.0.0.1");
        }
    
        public void connect(int port, String host) throws Exception { 
            // 配置客户端NIO线程组
            EventLoopGroup group = new NioEventLoopGroup();
            try { 
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            public void initChannel(SocketChannel ch)
                                    throws Exception { 
                                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new TimeClientHandler());
                            }
                        });
    
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
    
                // 当代客户端链路关闭
                f.channel().closeFuture().sync();
            } finally { 
                // 优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    }

package main.com.phei.netty.frame.correct;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeServerHandler extends ChannelHandlerAdapter { 
    
        private int counter;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            String body = (String) msg;
            System.out.println("The time server receive order : " + body
                    + " ; the counter is : " + ++counter);
            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
                    System.currentTimeMillis()).toString() : "BAD ORDER";
            currentTime = currentTime + System.getProperty("line.separator");
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.writeAndFlush(resp);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            ctx.close();
        }
    }

package main.com.phei.netty.frame.correct;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.util.logging.Logger;
    
    /** * @author lilinfeng * @version 1.0 * @date 2014年2月14日 */
    public class TimeClientHandler extends ChannelHandlerAdapter { 
    
        private static final Logger logger = Logger
                .getLogger(TimeClientHandler.class.getName());
    
        private int counter;
    
        private byte[] req;
    
        /** * Creates a client-side handler. */
        public TimeClientHandler() { 
            req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            ByteBuf message = null;
            for (int i = 0; i < 100; i++) { 
                message = Unpooled.buffer(req.length);
                message.writeBytes(req);
                ctx.writeAndFlush(message);
            }
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception { 
            String body = (String) msg;
            System.out.println("Now is : " + body + " ; the counter is : "
                    + ++counter);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            // 释放资源
            logger.warning("Unexpected exception from downstream : "
                    + cause.getMessage());
            ctx.close();
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]

## 分析 ##

在ChildChannelHandler中增加了两个Handler，其中，`LineBasedFrameDecoder`就是用来处理粘包和拆包问题的，没错，就这样就已经解决了，不需要额外的代码。  
`Linebasedframedecoder`的工作原理是它依次遍历 Bytebuf中的可读字节，判断看是否有“\\n”或者“\\r\\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流  
String Decoder 的功能非常简单，就是将接收到的对象转换成字符串，然后继续调用后面的 handler. LineBasedFramedecoder+ Stringdecoder组合就是按行切换的文本解码器，它被设计用来支持TCP的粘包和拆包。  
那么问题来了，如果我要发送的字符中有 \\n 这个字符串怎么办，是不是就会被拆成两个包了，于是我们再QUERY TIME \\n ORDER这个中间加一个反斜杠n，再运行一下，运行结果如下，发现收到了双倍的消息，确实会出问题，那么Netty是可以让用户自己去定义这个分隔符的，`DelimiterBasedFrameDecoder`这个类是用来自定义分隔符，`FixedLengthFrameDecoder`定长解码器也可以解决，以及`LengthFieldBasedFrameDecoder`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]

## 思考 ##

FixedLengthFrameDecoder是定长解码，也就是规定每个数据包的长度不够长的就补长，每个单独的`消息块`就是一个单独的消息，就不会出现粘包和拆包问题，  
LengthFieldBasedFrameDecoder是和UDP协议类似，将这次发送的数据的长度保存在数据的定长头部，这样先读头部是个字节，为这次数据的长度，然后把下面的长度读出来，接着再读数据的长度，同样可以解决粘包拆包问题。

> 那么亲爱的读者，你们有没有更好的逻辑算法来解决这个问题呢(自定义解码器)？

> [Netty之Java I/O演进之路（一）][Netty_Java I_O]  
> [Netty之NIO入门（二）][Netty_NIO]  
> [Netty之Netty入门应用（三）][Netty_Netty]  
> [Netty之TCP粘包/拆包问题（四）][Netty_TCP]  
> [Netty之编解码开发（五）][Netty]  
> [Netty之ChannelPipeline(六)][Netty_ChannelPipeline]  
> [Netty之EventLoop和EventLoopGroup(七)][Netty_EventLoop_EventLoopGroup]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70]: /images/20230208/180b6c94bf8842e595a814ab36ffb66f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 1]: /images/20230208/8fe4382ab1534feca8753395fef0c5e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200727160158885.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200727160210527.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200727162115928.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYW5qdW5fZmVp,size_16,color_FFFFFF,t_70
[Netty_Java I_O]: https://blog.csdn.net/jianjun_fei/article/details/107505729
[Netty_NIO]: https://blog.csdn.net/jianjun_fei/article/details/107513219
[Netty_Netty]: https://blog.csdn.net/jianjun_fei/article/details/107555647
[Netty_TCP]: https://blog.csdn.net/jianjun_fei/article/details/107563110
[Netty]: https://blog.csdn.net/jianjun_fei/article/details/107772642
[Netty_ChannelPipeline]: https://blog.csdn.net/jianjun_fei/article/details/107787864
[Netty_EventLoop_EventLoopGroup]: https://blog.csdn.net/jianjun_fei/article/details/107956949