netty粘包问题分析-阿里云开发者社区

Netty的粘包问题

在网络通信中，粘包问题是一种常见的情况，特别是在使用TCP协议进行数据传输时。本文将探讨Netty中的粘包问题及其解决方案，帮助读者更好地理解和应对这一问题。

1. 粘包问题介绍

1.1 什么是粘包问题

粘包问题是指发送端在将多个数据包连续发送到接收端时，接收端在接收数据时可能会将多个数据包粘合在一起，导致数据解析错误或丢失的情况。

1.2 粘包问题的影响

粘包问题可能导致接收端无法准确解析数据，从而影响系统的正常运行，甚至引发数据错误或丢失。

2. 粘包问题的原因

2.1 TCP协议的特点

TCP协议是面向连接的、可靠的数据传输协议，其特性会导致粘包问题的产生。

2.2 发送端造成的粘包

发送端在短时间内连续发送多个数据包时，可能会导致多个数据包在传输过程中被合并成一个数据包。

2.3 接收端造成的粘包

接收端在接收数据时，可能会将多个数据包一次性读取到缓冲区，造成粘包现象。

3. 粘包问题的解决方案

3.1 固定长度消息

通过设置固定长度的消息格式，使得每个数据包的长度固定，从而解决粘包问题。

服务端代码示例

ch.pipeline().addLast(new FixedLengthFrameDecoder(16));

服务端完整代码

// 示例代码
/**
 * @description: 固定长度方式解决粘包 问题服务端示例
 * @author: xz
 */
@Slf4j
public class NettyFixLengthServer {
    public static void main(String[] args) {
        new NettyFixLengthServer().start();
    }
    void start() {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .channel(NioServerSocketChannel.class)
                     //调整 netty 的接受缓冲区（byteBuf）
                    .childOption(ChannelOption.RCVBUF_ALLOCATOR,new AdaptiveRecvByteBufAllocator(16,16,16))
                    .group(boss, worker)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //设置定长解码器，位置必须再LoggingHandler之前，作用让所有数据包长度固定（假设长度为 16 字节）
                            ch.pipeline().addLast(new FixedLengthFrameDecoder(16));
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                //会在连接channel建立成功后，触发active事件
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    log.debug("connected>>>>>>>>>>>>>>>> {}", ctx.channel());
                                    super.channelActive(ctx);
                                }
                                @Override
                                public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                    log.debug("disconnect>>>>>>>>>>>>>>>> {}", ctx.channel());
                                    super.channelInactive(ctx);
                                }
                            });
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(8080);
            log.debug("{}>>>>>>>>>>>>>>>> binding...", channelFuture.channel());
            channelFuture.sync();
            log.debug("{}>>>>>>>>>>>>>>>> bound...", channelFuture.channel());
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            log.debug(">>>>>>>>>>>>>>>>stoped");
        }
    }
}

客户端代码

/**
 * @description: 固定长度方式解决粘包 问题客户端示例
 * @author: xz
 */
@Slf4j
public class NettyFixLengthClient {
    public static void main(String[] args) {
        send();
    }
    //剩余位置用下划线填充
    public static byte[] fill10Bytes(char c,int len){
        byte[] bytes = new byte[16];
        Arrays.fill(bytes, (byte) '_');
        for (int i = 0; i < len; i++) {
            bytes[i] = (byte) c;
        }
        System.out.println(new String(bytes));
        return bytes;
    }
    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 {
                    log.debug("connetted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            //设置ByteBuf
                            ByteBuf buffer = ctx.alloc().buffer();
                            // 发送内容随机的数据包
                            Random r = new Random();
                            char c = '0';
                            for (int i = 0; i < 10; i++) {
                                //剩余位置用下划线填充方法
                                byte[] bytes =fill10Bytes(c,r.nextInt(16)+1);
                                c++;
                                //写入到ByteBuf
                                buffer.writeBytes(bytes);
                            }
                            ctx.writeAndFlush(buffer);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

3.2 分隔符消息

在数据包之间添加特定的分隔符，通过分隔符来区分不同的数据包，从而解决粘包问题。

String recieveStr = ""$$$$这是一条粘包的消息&&&&$$$$这是一条粘包的消息&&&&"";
        if (recieveStr.contains("&&&&$$$$")) {
            log.debug("发生粘包");
            //去掉首尾的分隔符
            recieveStr = recieveStr.substring(4, recieveStr.length() - 4);
            //将分离好的消息写入字符串数组
            String[] totalBag = recieveStr.split("&&&&$$$$");
            for (String thisTotalBag : totalBag) {                
                //每条消息的业务逻辑。。。        
           
            }
        }

3.3 消息头+消息体

在消息中添加头部信息，包括消息长度等信息，接收端根据头部信息来解析数据包，从而解决粘包问题。

3.4 使用消息结束标志

在每个数据包的末尾添加特定的结束标志，接收端根据结束标志来识别数据包的边界，从而解决粘包问题。netty使用tcp/ip协议传输数据。而tcp/ip协议是类似水流一样的数据传输方式。多次访问的时候有可能出现数据粘包的问题。客户端和服务器，协商定义一个特殊的分隔符号，分隔符号长度自定义。如：‘#’、‘KaTeX parse error: Expected group after '_' at position 1: _̲’、‘AA@’。在通讯的时候，只要没有发送分隔符号，则代表一条数据没有结束。bootstrap.childHandler()方法，在该方法中，定义了一个ChannelHandler[] acceptorHandlers = new ChannelHandler[3];数组的三个元素分别对应：自定义结束符、自定义编码、自定义处理器。最后，将这个数组作为参数进行传递。

服务端代码

public class Server4Delimiter {
  // 监听线程组，监听客户端请求
  private EventLoopGroup acceptorGroup = null;
  // 处理客户端相关操作线程组，负责处理与客户端的数据通讯
  private EventLoopGroup clientGroup = null;
  // 服务启动相关配置信息
  private ServerBootstrap bootstrap = null;
  public Server4Delimiter(){
    init();
  }
  private void init(){
    acceptorGroup = new NioEventLoopGroup();
    clientGroup = new NioEventLoopGroup();
    bootstrap = new ServerBootstrap();
    // 绑定线程组
    bootstrap.group(acceptorGroup, clientGroup);
    // 设定通讯模式为NIO
    bootstrap.channel(NioServerSocketChannel.class);
    // 设定缓冲区大小
    bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
    // SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
    bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
      .option(ChannelOption.SO_RCVBUF, 16*1024)
      .option(ChannelOption.SO_KEEPALIVE, true);
  }
  public ChannelFuture doAccept(int port) throws InterruptedException{
    
    bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        // 数据分隔符, 定义的数据分隔符一定是一个ByteBuf类型的数据对象。
        ByteBuf delimiter = Unpooled.copiedBuffer("$E$".getBytes());
        ChannelHandler[] acceptorHandlers = new ChannelHandler[3];
        // 处理固定结束标记符号的Handler。这个Handler没有@Sharable注解修饰，
        // 必须每次初始化通道时创建一个新对象
        // 使用特殊符号分隔处理数据粘包问题，也要定义每个数据包最大长度。netty建议数据有最大长度。
        acceptorHandlers[0] = new DelimiterBasedFrameDecoder(1024, delimiter);
        // 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
        acceptorHandlers[1] = new StringDecoder(Charset.forName("UTF-8"));
        acceptorHandlers[2] = new Server4DelimiterHandler();
        ch.pipeline().addLast(acceptorHandlers);
      }
    });
    ChannelFuture future = bootstrap.bind(port).sync();
    return future;
  }
  public void release(){
    this.acceptorGroup.shutdownGracefully();
    this.clientGroup.shutdownGracefully();
  }
  
  public static void main(String[] args){
    ChannelFuture future = null;
    Server4Delimiter server = null;
    try{
      server = new Server4Delimiter();
      
      future = server.doAccept(9999);
      System.out.println("server started.");
      future.channel().closeFuture().sync();
    }catch(InterruptedException e){
      e.printStackTrace();
    }finally{
      if(null != future){
        try {
          future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
      
      if(null != server){
        server.release();
      }
    }
  }
  
}
 
 
public class Server4DelimiterHandler extends ChannelHandlerAdapter {
  
  // 业务处理逻辑
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    String message = msg.toString();
    System.out.println("from client : " + message);
    String line = "server message $E$ test delimiter handler!! $E$ second message $E$";
    ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
  }
  
 
  // 异常处理逻辑
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    System.out.println("server exceptionCaught method run...");
    // cause.printStackTrace();
    ctx.close();
  }
 
}

客户端代码

public class Client4Delimiter {
  
  // 处理请求和处理服务端响应的线程组
  private EventLoopGroup group = null;
  // 服务启动相关配置信息
  private Bootstrap bootstrap = null;
  
  public Client4Delimiter(){
    init();
  }
  
  private void init(){
    group = new NioEventLoopGroup();
    bootstrap = new Bootstrap();
    // 绑定线程组
    bootstrap.group(group);
    // 设定通讯模式为NIO
    bootstrap.channel(NioSocketChannel.class);
  }
  
  public ChannelFuture doRequest(String host, int port) throws InterruptedException{
    this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        // 数据分隔符
        ByteBuf delimiter = Unpooled.copiedBuffer("$E$".getBytes());
        ChannelHandler[] handlers = new ChannelHandler[3];
        handlers[0] = new DelimiterBasedFrameDecoder(1024, delimiter);
        // 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
        handlers[1] = new StringDecoder(Charset.forName("UTF-8"));
        handlers[2] = new Client4DelimiterHandler();
        
        ch.pipeline().addLast(handlers);
      }
    });
    ChannelFuture future = this.bootstrap.connect(host, port).sync();
    return future;
  }
  
  public void release(){
    this.group.shutdownGracefully();
  }
  
  public static void main(String[] args) {
    Client4Delimiter client = null;
    ChannelFuture future = null;
    try{
      client = new Client4Delimiter();
      
      future = client.doRequest("localhost", 9999);
      
      Scanner s = null;
      while(true){
        s = new Scanner(System.in);
        System.out.print("enter message send to server > ");
        String line = s.nextLine();
        future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
        TimeUnit.SECONDS.sleep(1);
      }
    }catch(Exception e){
      e.printStackTrace();
    }finally{
      if(null != future){
        try {
          future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
      if(null != client){
        client.release();
      }
    }
  }
  
}
 
 
 
public class Client4DelimiterHandler extends ChannelHandlerAdapter {
 
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    try{
      String message = msg.toString();
      System.out.println("from server : " + message);
    }finally{
      // 用于释放缓存。避免内存溢出
      ReferenceCountUtil.release(msg);
    }
  }
 
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    System.out.println("client exceptionCaught method run...");
    // cause.printStackTrace();
    ctx.close();
  }
 
}

3.5 使用消息定长协议

利用Netty提供的LengthFieldBasedFrameDecoder等解码器，处理定长消息，从而解决粘包问题。

public class Client4Delimiter {
  
  // 处理请求和处理服务端响应的线程组
  private EventLoopGroup group = null;
  // 服务启动相关配置信息
  private Bootstrap bootstrap = null;
  
  public Client4Delimiter(){
    init();
  }
  
  private void init(){
    group = new NioEventLoopGroup();
    bootstrap = new Bootstrap();
    // 绑定线程组
    bootstrap.group(group);
    // 设定通讯模式为NIO
    bootstrap.channel(NioSocketChannel.class);
  }
  
  public ChannelFuture doRequest(String host, int port) throws InterruptedException{
    this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        // 数据分隔符
        ByteBuf delimiter = Unpooled.copiedBuffer("$E$".getBytes());
        ChannelHandler[] handlers = new ChannelHandler[3];
        handlers[0] = new DelimiterBasedFrameDecoder(1024, delimiter);
        // 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
        handlers[1] = new StringDecoder(Charset.forName("UTF-8"));
        handlers[2] = new Client4DelimiterHandler();
        
        ch.pipeline().addLast(handlers);
      }
    });
    ChannelFuture future = this.bootstrap.connect(host, port).sync();
    return future;
  }
  
  public void release(){
    this.group.shutdownGracefully();
  }
  
  public static void main(String[] args) {
    Client4Delimiter client = null;
    ChannelFuture future = null;
    try{
      client = new Client4Delimiter();
      
      future = client.doRequest("localhost", 9999);
      
      Scanner s = null;
      while(true){
        s = new Scanner(System.in);
        System.out.print("enter message send to server > ");
        String line = s.nextLine();
        future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
        TimeUnit.SECONDS.sleep(1);
      }
    }catch(Exception e){
      e.printStackTrace();
    }finally{
      if(null != future){
        try {
          future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
      if(null != client){
        client.release();
      }
    }
  }
  
}
 
 
 
public class Client4DelimiterHandler extends ChannelHandlerAdapter {
 
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    try{
      String message = msg.toString();
      System.out.println("from server : " + message);
    }finally{
      // 用于释放缓存。避免内存溢出
      ReferenceCountUtil.release(msg);
    }
  }
 
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    System.out.println("client exceptionCaught method run...");
    // cause.printStackTrace();
    ctx.close();
  }
 
}

4. Netty中的粘包问题与解决方案

4.1 Netty中的粘包问题

分析Netty中可能出现的粘包问题。

4.2 Netty中的解决方案

介绍Netty提供的解决粘包问题的方案，如DelimiterBasedFrameDecoder、FixedLengthFrameDecoder等。

5. 注意事项与总结

5.1 注意事项

在处理粘包问题时需要注意的事项。

5.2 总结

总结解决粘包问题的方法及其优缺点，以及在实践中的应用。

通过本文的学习，读者可以更深入地了解Netty中的粘包问题及其解决方案，从而更好地应对实际开发中的网络通信挑战。

netty粘包问题分析