netty解决TCP的拆包和粘包的解决办法

ゞ浴缸里的玫瑰 2022-01-06 14:05 263阅读 0赞

TCP粘包、拆包问题

熟悉tcp编程的可能知道，无论是服务端还是客户端，当我们读取或者发送数据的时候，都需要考虑TCP底层的粘包个拆包机制。  
tcp是一个“流”协议，所谓流就是没有界限的传输数据，在业务上，一个完整的包可能会被TCP分成多个包进行发送，也可能把多个小包封装成一个大的数据包发送出去，这就是所谓的tcp粘包、拆包问题。  
分析TCP粘包拆包问题的产生原因：  
1. 应用程序write写入的字节大于套接口缓冲区的大小  
2. 进行mss大小的TCP分段  
3. 以太网的payload大于MTU进行IP分片  
粘包拆包问题，根据业界主流协议，有三种主要方案：  
1. 消息定长，例如每个报文的大小固定200个字节，如果不够，空位补空格  
2. 在包尾部增加特殊字符进行分割，例如加回车等  
3. 将消息分为消息头和消息体，在消息头中包含表示消息总长度的字段，然后进行业务逻辑的处理。

## TCP粘包拆包之分隔符类 DellmiterBasedFrameDecoder(自定义分隔符) ##

代码在bhz.netty.ende1下  
Server端

package com.weiyuan.test;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class Server {
    
        public static void main(String[] args) throws Exception{
            //1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的
            EventLoopGroup pGroup = new NioEventLoopGroup();
            EventLoopGroup cGroup = new NioEventLoopGroup();
            
            //2 创建服务器辅助类
            ServerBootstrap b = new ServerBootstrap();
            b.group(pGroup, cGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .option(ChannelOption.SO_SNDBUF, 32*1024)
             .option(ChannelOption.SO_RCVBUF, 32*1024)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    //设置特殊分隔符
                    ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
                    sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                    //设置字符串形式的解码
                    sc.pipeline().addLast(new StringDecoder());
                    sc.pipeline().addLast(new ServerHandler());
                }
            });
            
            //4 绑定连接
            ChannelFuture cf = b.bind(8765).sync();
            
            //等待服务器监听端口关闭
            cf.channel().closeFuture().sync();
            pGroup.shutdownGracefully();
            cGroup.shutdownGracefully();
            
        }
        
    }

ServerHandle端

package com.weiyuan.test;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ServerHandler extends ChannelHandlerAdapter {
    
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(" server channel active... ");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String request = (String)msg;
            System.out.println("Server :" + msg);
            String response = "服务器响应：" + msg + "$_";
            ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
            ctx.close();
        }
    
    
    
    
    }

Client端

package com.weiyuan.test;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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 io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class Client {
    
        public static void main(String[] args) throws Exception {
            
            EventLoopGroup group = new NioEventLoopGroup();
            
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    //
                    ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
                    sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
                    sc.pipeline().addLast(new StringDecoder());
                    sc.pipeline().addLast(new ClientHandler());
                }
            });
            
            ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
            
            cf.channel().writeAndFlush(Unpooled.wrappedBuffer("bbbb$_".getBytes()));
            cf.channel().writeAndFlush(Unpooled.wrappedBuffer("cccc$_".getBytes()));
            
            
            //等待客户端端口关闭
            cf.channel().closeFuture().sync();
            group.shutdownGracefully();
            
        }
    }

ClientHandle端

package com.weiyuan.test;
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.ReferenceCountUtil;
    
    public class ClientHandler extends ChannelHandlerAdapter{
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("client channel active... ");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try {
                String response = (String)msg;
                System.out.println("Client: " + response);
            } finally {
                ReferenceCountUtil.release(msg);
            }
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    
    }

先启动Server端，再启动Client端  
Client端输出

client channel active...  
Client: 服务器响应：bbbb  
Client: 服务器响应：cccc

server端输出

server channel active...  
Server :bbbb  
Server :cccc

## TCP粘包拆包之FixedLengthFrameDecoder(定长) ##

和上面的代码类似  
Server端

package com.weiyuan.test;
    
    import java.nio.ByteBuffer;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class Server {
    
        public static void main(String[] args) throws Exception{
            //1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的
            EventLoopGroup pGroup = new NioEventLoopGroup();
            EventLoopGroup cGroup = new NioEventLoopGroup();
            
            //2 创建服务器辅助类
            ServerBootstrap b = new ServerBootstrap();
            b.group(pGroup, cGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .option(ChannelOption.SO_SNDBUF, 32*1024)
             .option(ChannelOption.SO_RCVBUF, 32*1024)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    //设置定长字符串接收
                    sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
                    //设置字符串形式的解码
                    sc.pipeline().addLast(new StringDecoder());
                    sc.pipeline().addLast(new ServerHandler());
                }
            });
            
            //4 绑定连接
            ChannelFuture cf = b.bind(8765).sync();
            
            //等待服务器监听端口关闭
            cf.channel().closeFuture().sync();
            pGroup.shutdownGracefully();
            cGroup.shutdownGracefully();
            
        }
        
    }

ServerHandler

package com.weiyuan.test;
    
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ServerHandler extends ChannelHandlerAdapter {
    
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(" server channel active... ");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String request = (String)msg;
            System.out.println("Server rcv :" + msg);
            String response =  request ;
            ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
    
        }
    
    
    
    
    }

client端

package com.weiyuan.test;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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 io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.FixedLengthFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class Client {
    
        public static void main(String[] args) throws Exception {
            
            EventLoopGroup group = new NioEventLoopGroup();
            
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception {
                    sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
                    sc.pipeline().addLast(new StringDecoder());
                    sc.pipeline().addLast(new ClientHandler());
                }
            });
            
            ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
            
            cf.channel().writeAndFlush(Unpooled.wrappedBuffer("aaaaabbbbb".getBytes()));
            cf.channel().writeAndFlush(Unpooled.copiedBuffer("ccccccc".getBytes()));
            
            //等待客户端端口关闭
            cf.channel().closeFuture().sync();
            group.shutdownGracefully();
            
        }
    }

ClientHandler端

package com.weiyuan.test;
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ClientHandler extends ChannelHandlerAdapter{
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("client channel active... ");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String response = (String)msg;
            System.out.println("Client: " + response);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        }
    
    }

先运行Server端，再运行Client端  
Client端输出

client channel active...  
Client: aaaaa  
Client: bbbbb  
Client: ccccc

server端输出

server channel active...  
Server rcv :aaaaa  
Server rcv :bbbbb  
Server rcv :ccccc

可以看到，每次定长都是5个字符， 有两个cc，由于不够五个，导致丢掉了

## 自定义协议粘包拆包 ##

使用消息头和消息体， 暂未记录笔记。 TODO zhaikaishun

特别感谢互联网架构师白鹤翔老师，本文大多出自他的视频讲解。  
笔者主要是记录笔记，以便之后翻阅，正所谓好记性不如烂笔头，烂笔头不如云笔记

转载于:https://www.cnblogs.com/kebibuluan/p/8477446.html