Netty 实战之解决粘包/半包问题

╰+攻爆jí腚メ 2022-10-22 04:10 214阅读 0赞

# 1 一个例子 #

## 1.1 EchoServer ##

package com.rosh.demo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/1 15:15 */
    public class EchoServer { 
    
    
    
        public static void main(String[] args) throws InterruptedException { 
    
            new EchoServer().start();
        }
    
    
        public void start() throws InterruptedException { 
            //服务端启动类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //线程组
            EventLoopGroup group = new NioEventLoopGroup();
            try { 
                /** * (1) 服务端绑定线程组 * (2) 设置channel通道类型为NIO * (3) 指定监听端口 * (4) 添加EchoServerHandler */
    
                final EchoServerHandler echoServerHandler = new EchoServerHandler();
                serverBootstrap.group(group)
                        .channel(NioServerSocketChannel.class)
                        .localAddress(8888)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception { 
                                ch.pipeline().addLast(echoServerHandler);
                            }
                        });
    
                //异步绑定服务器，sync阻塞到绑定完成
                ChannelFuture channelFuture = serverBootstrap.bind().sync();
                System.out.println("服务器已经启动......");
                //阻塞当前线程，直到服务器的channel关闭
                channelFuture.channel().closeFuture().sync();
            } finally { 
                /** * 优雅退出 */
                group.shutdownGracefully().sync();
            }
        }
    
    
    }

## 1.2 EchoServerHandler ##

package com.rosh.demo;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/1 15:29 */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter { 
    
        private int serverIndex = 0;
    
    
        /** * 读取数据 */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf in = (ByteBuf) msg;
            System.out.println("Server accept:" + in.toString(CharsetUtil.UTF_8) + ", serverIndex= " + ++serverIndex);
            ctx.writeAndFlush(in);
        }
    
    
        /** * 异常处理 */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    
    
    }

## 1.3 EchoClient ##

package com.rosh.demo;
    
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    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 java.net.InetSocketAddress;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/1 15:15 */
    public class EchoClient { 
    
    
    
        public static void main(String[] args) throws InterruptedException { 
            new EchoClient().start();
        }
    
    
    
        public void start() throws InterruptedException { 
            Bootstrap bootstrap = new Bootstrap();
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try { 
    
                /** * 设置线程组、设置通道类型、绑定远端地址、设置handler */
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .remoteAddress(new InetSocketAddress("127.0.0.1", 8888))
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception { 
                                ch.pipeline().addLast(new EchoClientHandler());
                         }
                });
    
                //异步链接服务器
                ChannelFuture channelFuture = bootstrap.connect().sync();
                //阻塞
                channelFuture.channel().closeFuture().sync();
    
            } finally { 
                eventLoopGroup.shutdownGracefully().sync();
            }
        }
    
    
    
    }

## 1.4 EchoClientHandler ##

package com.rosh.demo;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    /** * @Description: * @Author: Rosh * @Date: 2021/4/1 16:04 */
    public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    
        private int clientIndex = 0;
    
    
        /** * 读取到网络数据后进行处理 */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
    
            System.out.println("client Accept msg:" + msg.toString(CharsetUtil.UTF_8) + ", clientIndex=" + ++clientIndex);
        }
    
        /** * channel 连接成功后做业务处理 */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
    
            for (int i = 1; i <= 100; i++) { 
                ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Netty", CharsetUtil.UTF_8));
            }
    
        }
    }

## 1.5 运行结果 ##

**描述：** 当客户端循环100次给服务端发送消息，服务端并没有像我们想的那样收了100次消息，仅仅收到2次消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]

# 2 什么是 TCP 粘包半包？ #

假设客户端分别发送了两个数据包 D1 和 D2 给服务端，由于服务端一次读取到的字节数是不确定的，故可能存在以下 4 种情况。
    （1）服务端分两次读取到了两个独立的数据包，分别是 D1 和 D2，没有粘包和拆包；
    （2）服务端一次接收到了两个数据包，D1 和 D2 粘合在一起，被称为 TCP 粘包；
    （3）服务端分两次读取到了两个数据包，第一次读取到了完整的 D1 包和 D2 包的部分内容，第二次读取到了 D2 包的剩余内容，这被称为 TCP 拆包；
    （4）服务端分两次读取到了两个数据包，第一次读取到了 D1 包的部分内容 D1_1，第二次读取到了 D1 包的剩余内容 D1_2 和 D2 包的整包。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]

# 3 TCP 粘包/半包发生的原因 #

由于 TCP 协议本身的机制（面向连接的可靠地协议-三次握手机制）客户端与服务器会维持一个连接（Channel），数据在连接不断开的情况下，可以持续不断地将多个数据包发往服务器，但是如果发送的网络数据包太小，那么他本身会启用 Nagle 算法（可配置是否启用）对较小的数据包进行合并（基于此，TCP 的网络延迟要 UDP 的高些）然后再发送（超时或者包大小足够）。那么这样的话，服务器在接收到消息（数据流）的时候就无法区分哪些数据包是客户端自己分开发送的，这样产生了粘包；服务器在接收到数据库后，放到缓冲区中，如果消息没有被及时从缓存区取走，下次在取数据的时候可能就会出现一次取出多个数据包的情况，造成粘包现象。

UDP：本身作为无连接的不可靠的传输协议（适合频繁发送较小的数据包），他不会对数据包进行合并发送（也就没有 Nagle 算法之说了），他直接是一端发送什么数据，直接就发出去了，既然他不会对数据合并，每一个数据包都是完整的（数据+UDP 头+IP 头等等发一次数据封装一次）也就没有粘包一说了。

分包产生的原因就简单的多：可能是 IP 分片传输导致的，也可能是传输过程中丢失部分包导致出现的半包，还有可能就是一个包可能被分成了两次传输，在取数据的时候，先取到了一部分（还可能与接收的缓冲区大小有关系），总之就是一个数据包被分成了多次接收。

# 4 分隔符解决粘包/半包问题 #

**描述：** 客户端发送消息，服务端响应消息时加上消息分隔符，并且加相应的channelHandler进行消息解码。

## 4.1 LineBasedFrameDecoder ##

**描述：** netty提供解码器，按照\\n、\\r\\n进行解码。

### 4.1.1 LineBaseClient ###

public class LineBaseClient { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            new LineBaseClient().start();
    
        }
    
    
        public void start() throws InterruptedException { 
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
            try { 
                Bootstrap bootstrap = new Bootstrap();
                final ChannelInitializerImpl channelInitializer = new ChannelInitializerImpl();
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .remoteAddress(new InetSocketAddress("127.0.0.1", 8888))
                        .handler(channelInitializer);
    
    
                ChannelFuture channelFuture = bootstrap.connect().sync();
                channelFuture.channel().closeFuture().sync();
    
            } finally { 
                eventLoopGroup.shutdownGracefully().sync();
            }
        }
    
    
        private static class ChannelInitializerImpl extends ChannelInitializer<Channel> { 
    
            @Override
            protected void initChannel(Channel ch) throws Exception { 
    
                /** * 创建一个解码器，\n、\r\n均会被处理 */
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new LineBaseClientHandler());
            }
        }
    
    
    }

### 4.1.2 LineBaseClientHandler ###

public class LineBaseClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    
        private int clientIndex = 0;
    
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            for (int i = 1; i <= 100; i++) { 
                String str = "Hello NettyServer" + System.getProperty("line.separator");
                ctx.writeAndFlush(Unpooled.copiedBuffer(str, CharsetUtil.UTF_8));
            }
        }
    
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    
        /** * 读取到网络数据后进行处理 */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
            System.out.println("client Accept msg:" + msg.toString(CharsetUtil.UTF_8) + " , clientIndex=" + ++clientIndex);
        }
    
    }

### 4.1.3 LineBaseServer ###

public class LineBaseServer { 
    
    
        public static void main(String[] args) throws InterruptedException { 
    
            new LineBaseServer().start();
        }
    
        public void start() throws InterruptedException { 
            EventLoopGroup group = new NioEventLoopGroup();
            try { 
    
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(group)
                        .channel(NioServerSocketChannel.class)
                        .localAddress(new InetSocketAddress(8888))
                        .childHandler(new ChannelInitializerImp());
    
                ChannelFuture channelFuture = serverBootstrap.bind().sync();
                System.out.println("服务器已经启动......");
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) { 
                group.shutdownGracefully().sync();
            }
        }
    
    
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> { 
    
            @Override
            protected void initChannel(Channel ch) throws Exception { 
                /** * 创建一个解码器，\n、\r\n均会被处理 */
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new LineBaseServerHandler());
            }
        }
    
    
    }

### 4.1.4 LineBaseServerHandler ###

public class LineBaseServerHandler extends ChannelInboundHandlerAdapter { 
    
    
        private int serverIndex = 0;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("Server accept:" + byteBuf.toString(CharsetUtil.UTF_8) + ", serverIndex=" + ++serverIndex);
    
            String str = "Hello NettyClient" + System.getProperty("line.separator");
            ctx.writeAndFlush(Unpooled.copiedBuffer(str, CharsetUtil.UTF_8));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    }

### 4.1.5 运行结果 ###

**服务端：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]  
**客户端：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]

## 4.2 DelimiterBasedFrameDecoder ##

**描述：** 按照自定义的分割符进行消息解码。

### 4.2.1 DelimiterConsts ###

package com.rosh.delimiter;
    
    /** * @Description: * @Author: rosh * @Date: 2021/4/3 15:22 */
    public  final  class DelimiterConsts { 
    
        private DelimiterConsts(){ 
    
        }
    
        public static final String DELIMITER_SYMBOL = "@";
    
    }

### 4.2.2 DelimiterServer ###

package com.rosh.delimiter;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    
    import java.net.InetSocketAddress;
    
    /** * @Description: * @Author: rosh * @Date: 2021/4/3 14:59 */
    public class DelimiterServer { 
    
        public static void main(String[] args) throws InterruptedException { 
            new DelimiterServer().start();
        }
    
    
    
        public void start() throws InterruptedException { 
    
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try { 
    
                final ChannelInitializerImpl channelInitializer = new ChannelInitializerImpl();
    
    
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(eventLoopGroup)
                        .channel(NioServerSocketChannel.class)
                        .localAddress(new InetSocketAddress(8888))
                        .childHandler(channelInitializer);
    
                ChannelFuture channelFuture = serverBootstrap.bind().sync();
    
                System.out.println("DelimiterServer 已经启动");
                channelFuture.channel().closeFuture().sync();
    
            } finally { 
                eventLoopGroup.shutdownGracefully().sync();
            }
    
        }
    
    
        private static class ChannelInitializerImpl extends ChannelInitializer<Channel> { 
    
            @Override
            protected void initChannel(Channel ch) throws Exception     { 
                ByteBuf delimiter = Unpooled.copiedBuffer(DelimiterConsts.DELIMITER_SYMBOL.getBytes());
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                ch.pipeline().addLast(new DelimiterServerHandler());
            }
        }
    
    }

### 4.2.3 DelimiterServerHandler ###

package com.rosh.delimiter;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /** * @Description: * @Author: rosh * @Date: 2021/4/3 14:59 */
    public class DelimiterServerHandler extends ChannelInboundHandlerAdapter { 
    
    
    
        private int index = 0;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("Server accept:" + byteBuf.toString(CharsetUtil.UTF_8) + ", serverIndex = " + ++index);
    
    
            String str = "Hello NettyClient" + DelimiterConsts.DELIMITER_SYMBOL;
            ctx.writeAndFlush(Unpooled.copiedBuffer(str, CharsetUtil.UTF_8));
    
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    
    }

### 4.2.4 DelimiterClient ###

package com.rosh.delimiter;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    
    import java.net.InetSocketAddress;
    
    /** * @Description: * @Author: rosh * @Date: 2021/4/3 15:15 */
    public class DelimiterClient { 
    
    
    
        public static void main(String[] args) throws InterruptedException { 
    
            new DelimiterClient().start();
    
        }
    
    
        public void start() throws InterruptedException { 
    
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
            try { 
    
                final ChannelInitializerImpl channelInitializer = new ChannelInitializerImpl();
    
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        .remoteAddress(new InetSocketAddress("127.0.0.1", 8888))
                        .handler(channelInitializer);
    
                ChannelFuture channelFuture = bootstrap.connect().sync();
                channelFuture.channel().closeFuture().sync();
            } finally { 
                eventLoopGroup.shutdownGracefully().sync();
            }
    
        }
    
        private static class ChannelInitializerImpl extends ChannelInitializer<Channel> { 
    
    
            @Override
            protected void initChannel(Channel ch) throws Exception { 
                ByteBuf byteBuf = Unpooled.copiedBuffer(DelimiterConsts.DELIMITER_SYMBOL.getBytes());
                //自定义分隔符
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, byteBuf));
                ch.pipeline().addLast(new DelimiterClientHandler());
            }
        }
    
    }

### 4.2.5 DelimiterClientHandler ###

package com.rosh.delimiter;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    /** * @Description: * @Author: rosh * @Date: 2021/4/3 15:20 */
    public class DelimiterClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    
        private int clientIndex = 0;
    
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            for (int i = 1; i <= 100; i++) { 
                String str = "Hello NettyServer" + DelimiterConsts.DELIMITER_SYMBOL;
                ctx.writeAndFlush(Unpooled.copiedBuffer(str, CharsetUtil.UTF_8));
            }
        }
    
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
            ctx.close();
        }
    
        /** * 读取到网络数据后进行处理 */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
            System.out.println("client Accept msg:" + msg.toString(CharsetUtil.UTF_8) + " , clientIndex=" + ++clientIndex);
        }
    
    }

### 4.2.6 运行结果 ###

**服务端：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]

**客户端：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20221022/840aea4a2446471f88200ed094950da0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20221022/0ca4b45bccb643ed8cd3d76350279cc4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]: /images/20221022/9a10625193dc4785b06753e1fa650b19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]: /images/20221022/4266781c5a8d445fb4609661f6127170.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]: /images/20221022/86f9c58dca9d4423a1bf7758d1d54160.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]: /images/20221022/35403ffb70ba4677ae41eb5b551c234f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]: /images/20221022/f9724855887f454280f94c93943b7443.png