Netty系列(5)Netty中解决TCP粘包和拆包

文章目录

1 TCP 粘包和拆包介绍

  • TCP 是面向连接的,面向流的,提供高可靠性服务。收发两端(客户端和服务器端)都要有一一成对的 socket,因此,发送端为了将多个发给接收端的包,更有效的发给对方,使用了优化方法(Nagle 算法),将多次间隔较小且数据量小的数据,合并成一个大的数据块,然后进行封包。这样做虽然提高了效率,但是接收端就难于分辨出完整的数据包了,因为面向流的通信是无消息保护边界的

  • 由于 TCP 无消息保护边界,需要在接收端处理消息边界问题,也就是我们所说的粘包、拆包

  • TCP 粘包、拆包图解

Netty系列(5)Netty中解决TCP粘包和拆包

​ 假设客户端分别发送了两个数据包 D1D2 给服务端,由于服务端一次读取到字节数是不确定的,故可能存在以下四种情况:

  1. 服务端分两次读取到了两个独立的数据包,分别是 D1D2,没有粘包和拆包
  2. 服务端一次接受到了两个数据包,D1D2 粘合在一起,称之为 TCP 粘包
  3. 服务端分两次读取到了数据包,第一次读取到了完整的 D1 包和 D2 包的部分内容,第二次读取到了 D2 包的剩余内容,这称之为 TCP 拆包
  4. 服务端分两次读取到了数据包,第一次读取到了 D1 包的部分内容 D1_1,第二次读取到了 D1 包的剩余部分内容 D1_2 和完整的 D2 包,这称之为 TCP 拆包。

2 Netty 粘包演示

  • 服务端代码

    package com.warybee.tcp;
    
    import com.warybee.simple.NettyServerHandler;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     */
    public class MyServer {
    
        public static void main(String[] args) {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                ServerBootstrap bootstrap=new ServerBootstrap();
                bootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,128)
                        .childOption(ChannelOption.SO_KEEPALIVE,true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new MyServerHandler());
                            }
                        });
                ChannelFuture channelFuture = bootstrap.bind(9999).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }
    
  • 服务端handler

    package com.warybee.tcp;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    import java.nio.charset.Charset;
    import java.util.UUID;
    
    /**
     */
    public class MyServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private int count;
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            byte[] buffer = new byte[msg.readableBytes()];
            msg.readBytes(buffer);
    
            //将buffer转成字符串
            String message = new String(buffer, Charset.forName("utf-8"));
    
            System.out.println("服务器接收到数据 " + message);
            System.out.println("服务器接收到消息量=" + (++this.count));
    
            //服务器回送数据给客户端, 回送一个随机id ,
            ByteBuf responseByteBuf = Unpooled.copiedBuffer(UUID.randomUUID().toString() + "\n",
                    Charset.forName("utf-8"));
            ctx.writeAndFlush(responseByteBuf);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
    

    服务端的 handler 主要逻辑是接收客户端发送过来的数据,看看是否是一条一条接收。然后每次接收到数据之后给客户端回复一个确认消息。

  • 客户端

    package com.warybee.tcp;
    
    import com.warybee.simple.NettyClientHandler;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    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.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    /**
     * @description
     */
    public class MyClient {
    
        public static void main(String[] args) {
            EventLoopGroup eventExecutors = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
    
                bootstrap.group(eventExecutors)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(new StringEncoder());
                                pipeline.addLast(new MyClientHandler());
                            }
                        });
                ChannelFuture cf = bootstrap.connect("127.0.0.1", 9999).sync();
                cf.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                eventExecutors.shutdownGracefully();
            }
        }
    }
    
  • 客户端Handler

package com.warybee.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;

/**
 * @author joy
 * @create 2021-10-27 14:50
 * @description
 */
public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    private int count;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);

        String message = new String(buffer, Charset.forName("utf-8"));
        System.out.println("客户端接收到消息=" + message);
        System.out.println("客户端接收消息数量=" + (++this.count));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 20; ++i) {
            ByteBuf buffer = Unpooled.copiedBuffer("你好服务端:" + i+"rn", Charset.forName("utf-8"));
            ctx.writeAndFlush(buffer);
        }
    }
}

客户端 handler 主要逻辑是:循环20次给服务端发送测试消息。接收服务端的确认消息

启动项目之后服务端收到的消息结果:

Netty系列(5)Netty中解决TCP粘包和拆包

这里能看到多条消息被粘到一起发送了。

3 Netty中解决TCP粘包和拆包

使用自定义协议+编解码器来解决。关键就是要解决服务器端每次读取数据长度的问题,这个问题解决,就不会出现服务器多读或少读数据的问题,从而避免的 TCP 粘包、拆包。

  • 自定义消息协议

    package com.warybee.tcp;
    
    /**
     * @description 自定义协议包
     */
    public class MessageProtocol {
        //消息长度
        private int len;
        /**
         * 消息
         */
        private byte[] content;
    
        public int getLen() {
            return len;
        }
    
        public void setLen(int len) {
            this.len = len;
        }
    
        public byte[] getContent() {
            return content;
        }
    
        public void setContent(byte[] content) {
            this.content = content;
        }
    }
    
  • 自定义编码器

    package com.warybee.tcp;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    /**
     * @description 自定义协议包编码
     */
    public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {
        @Override
        protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
            out.writeInt(msg.getLen());
            out.writeBytes(msg.getContent());
        }
    }
    
  • 自定义解码器

    package com.warybee.tcp;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    /**
     * @description
     */
    public class MyMessageDecoder extends ReplayingDecoder<Void> {
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            //需要将得到二进制字节码转化为MessageProtocol 数据包(对象)
            int length = in.readInt();
            byte[] content = new byte[length];
            in.readBytes(content);
    
            //封装成 MessageProtocol 对象,放入 out, 传递下一个handler业务处理
            MessageProtocol messageProtocol = new MessageProtocol();
            messageProtocol.setLen(length);
            messageProtocol.setContent(content);
            out.add(messageProtocol);
        }
    }
    
    • 服务端代码

      package com.warybee.tcp;
      
      import com.warybee.simple.NettyServerHandler;
      import io.netty.bootstrap.ServerBootstrap;
      import io.netty.channel.*;
      import io.netty.channel.nio.NioEventLoopGroup;
      import io.netty.channel.socket.SocketChannel;
      import io.netty.channel.socket.nio.NioServerSocketChannel;
      
      /**
       * @author joy
       */
      public class MyServer {
      
          public static void main(String[] args) {
              EventLoopGroup bossGroup = new NioEventLoopGroup();
              EventLoopGroup workerGroup = new NioEventLoopGroup();
      
              try {
                  ServerBootstrap bootstrap=new ServerBootstrap();
                  bootstrap.group(bossGroup,workerGroup)
                          .channel(NioServerSocketChannel.class)
                          .option(ChannelOption.SO_BACKLOG,128)
                          .childOption(ChannelOption.SO_KEEPALIVE,true)
                          .childHandler(new ChannelInitializer<SocketChannel>() {
      
                              @Override
                              protected void initChannel(SocketChannel ch) throws Exception {
                                  ChannelPipeline pipeline = ch.pipeline();
                                  //添加自定义编码器
                                  pipeline.addLast("encoder",new MyMessageEncoder());
                                  //添加自定义解码器
                                  pipeline.addLast("decoder",new MyMessageDecoder());
                                  pipeline.addLast(new MyServerHandler());
                              }
                          });
                  ChannelFuture channelFuture = bootstrap.bind(9999).sync();
                  channelFuture.channel().closeFuture().sync();
              } catch (InterruptedException e) {
                  e.printStackTrace();
              } finally {
                  bossGroup.shutdownGracefully();
                  workerGroup.shutdownGracefully();
              }
          }
      }
      
    • 服务端handler

      package com.warybee.tcp;
      
      import io.netty.buffer.ByteBuf;
      import io.netty.buffer.Unpooled;
      import io.netty.channel.ChannelHandlerContext;
      import io.netty.channel.SimpleChannelInboundHandler;
      
      import java.nio.charset.Charset;
      import java.util.UUID;
      
      /**
       * @description
       */
      public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
      
          private int count;
      
          @Override
          protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
              //接收到数据,并处理
              int len = msg.getLen();
              byte[] content = msg.getContent();
      
              System.out.println();
              System.out.println();
              System.out.println();
              System.out.println("服务器接收到信息如下");
              System.out.println("长度=" + len);
              System.out.println("内容=" + new String(content, Charset.forName("utf-8")));
      
              System.out.println("服务器接收到消息包数量=" + (++this.count));
      
              //回复消息
              String responseContent = UUID.randomUUID().toString();
              int responseLen = responseContent.getBytes("utf-8").length;
              byte[] responseContent2 = responseContent.getBytes("utf-8");
              //构建一个协议包
              MessageProtocol messageProtocol = new MessageProtocol();
              messageProtocol.setLen(responseLen);
              messageProtocol.setContent(responseContent2);
      
              ctx.writeAndFlush(messageProtocol);
          }
      
          @Override
          public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
              cause.printStackTrace();
              ctx.close();
          }
      }
      

服务端的 handler 主要逻辑是接收客户端发送过来的数据,看看是否是一条一条接收。然后每次接收到数据之后给客户端回复一个确认消息。**

  • 客户端handler

    package com.warybee.tcp;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    import java.nio.charset.Charset;
    
    /**
     * @author joy
     * @create 2021-10-27 14:50
     * @description
     */
    public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    
        private int count;
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
            int len = msg.getLen();
            byte[] content = msg.getContent();
    
            System.out.println("客户端接收到消息如下");
            System.out.println("长度=" + len);
            System.out.println("内容=" + new String(content, Charset.forName("utf-8")));
    
            System.out.println("客户端接收消息数量=" + (++this.count));
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //使用客户端发送5条数据 "今天天气冷,吃火锅" 编号
    
            for (int i = 0; i < 5; i++) {
                String mes = "今天天气冷,吃火锅";
                byte[] content = mes.getBytes(Charset.forName("utf-8"));
                int length = mes.getBytes(Charset.forName("utf-8")).length;
                //创建协议包对象
                MessageProtocol messageProtocol = new MessageProtocol();
                messageProtocol.setLen(length);
                messageProtocol.setContent(content);
                ctx.writeAndFlush(messageProtocol);
            }
        }
    }
    
上一篇:QQ音乐API分析记录


下一篇:Excel阅读模式/聚光灯开发技术序列作品之三 高级自定义任务窗格开发原理简述—— 隐鹤