【netty系列-09】深入理解和解决tcp的粘包拆包

Netty系列整体栏目

一，tcp层的粘包分包问题

在前面了解完整个netty的基本组件和使用之后，本篇文章讲解一个关于网络编程的重点，就是在netty中是如何处理这种tcp层面的粘包和半包问题。

依旧得回归下图，在网络通信编程中，数据要从客户端发送到另一个对端，都需要从客户端的应用层，将数据封装成报文，往下层层封装，然后通过以太网等将数据发送给对端，对端接收到数据之后，将数据从物理层往上层层解析，最终数据解析到应用层，解析后获取到客户端发送的数据。由于在操作系统层面呢，操作系统内部将tcp层以下的协议全部封装好，将内部所有的细节以及实现封装成一个个socket，让开发者只需要更加的关注与应用层的开发，通过操作socket实现与对端的通信。

在前面的nio中讲到，reactor反应堆模式的三大特性分别是：Selector、SocketChannel和Buffer ，并且netty是基于nio实现的，所以不管是在原生的nio中，还是在基于nio实现的netty中，都离不开这个 Buffer ，而在本篇文章中要讲解的这个粘包和半包问题，就是由于这个Buffer缓冲区导致的。如下图，Buffer又有读buffer和写buffer，由于tcp的全双工的特性，因此底层是实现了同时读写的功能

1，通过代码直观的表现出粘包的问题

1.1，服务端代码实现

何为粘包，顾名思义，就是多个包黏贴在一起了。接下来通过一段代码来表现出粘包的问题，还是那套配方，编写服务端主启动类，客户端主启动类，然后就是一个个由用户自定义实现的一些 handler 。服务端主启动类的配置如下

/**
 * @author zhenghuisheng
 * @date 2024/9/1 21:35
 * 粘包服务端主启动类
 */
public class StickPackageServer {
    private static Integer port = 8888;
    public static void main(String[] args) {
        // 创建自定义事件组，一个线程循环的处理事件，类似与nio的selector
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        try{
            //创建服务端主启动类
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(loopGroup)    //绑定组
                    .channel(NioServerSocketChannel.class)
                    .localAddress(port)         //绑定端口
                    .childHandler(new ChannelInitializer<SocketChannel>() { //初始化channel,将事件加入
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new StickPackageServerHandler());     //将事件加入到管道中
                        }
                    });
            //完成绑定,内部如果异步实现bind，因此需要阻塞拿到返回结果
            ChannelFuture future = bootstrap.bind().sync();
            //关闭future时也需要阻塞，内部也采用的是异步操作
            future.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                //处理中断异常
                loopGroup.shutdownGracefully().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

服务端中处理事件的Handler如下，这边主要统计客户端发送了多少个报文过来

@Slf4j
public class StickPackageServerHandler extends ChannelInboundHandlerAdapter {
    private AtomicInteger counter = new AtomicInteger(0);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf byteBuf = (ByteBuf)msg;
        String request = byteBuf.toString(CharsetUtil.UTF_8);
        log.info("服务端接收到请求数量为" + counter.incrementAndGet());
        String resp = request + "成功请求";
        ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
    }
}

1.2，客户端代码实现

首先是客户端主启动类的代码实现，将要处理事件的 StickPackageClientHandler 加入

/**
 * @author zhenghuisheng
 * @date 2024/9/1 21:38
 */
public class StickPackageClient {
    private static Integer port = 8888;
    private static String host = "127.0.0.1";
    public static void main(String[] args) {
        // 创建自定义事件组，一个线程循环的处理事件，类似与nio的selector
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        try{
            //客户端只需要用bootStrap
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(loopGroup)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host,port))    //和服务器不一样，这里只需要连接服务器地址即可
                    .handler(new ChannelInitializer<SocketChannel>() {  //和服务端不同，服务端使用的childHandler客户端只需要具体的handler即可
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new StickPackageClientHandler());
                        }
                    });
            //完成绑定,内部如果异步实现bind，因此需要阻塞拿到返回结果
            ChannelFuture future = bootstrap.connect().sync();
            //关闭future时也需要阻塞，内部也采用的是异步操作
            future.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

随后定义这个 StickPackageClientHandler 事件的具体实现，通过实现这个 channelActive 方法来触发所需要执行的动作。这里主要是定义好对应的数据，内部将对应的数据封装成报文发送给对端，然后循环10次往服务端发送数据。

/**
 * @author zhenghuisheng
 * @date 2024/9/1 21:38
 */
@Slf4j
public class StickPackageClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private AtomicInteger counter = new AtomicInteger(0);

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        log.info("接收到的请求数量为：" + counter.incrementAndGet());
    }

    //事件被触发后所执行的动作
    @Override
    public void channelActive(ChannelHandlerContext ctx) {

        //定义请求内容
        String request = "abcdefghijklmnopqrstuvwxyz" + System.getProperty("line.separator");
        final ByteBufAllocator byteBufAllocator = ctx.alloc();
        ByteBuf msg = null;

        //给服务器发送10个报文
        for(int i=0;i<10;i++){
            msg = byteBufAllocator.buffer(request.length());
            msg.writeBytes(request.getBytes());
            ctx.writeAndFlush(msg);
        }
    }
}

随后先启动服务端，然后再启动客户端，打印的结果如下。

INFO com.zhs.netty.netty.stickypackage.StickPackageServerHandler - 服务端请求数量为1

按理来说服务端也应该接收到10个报文，但是打印日志显示服务端只接收到一个报文，因此猜想而知就是数据在传输过程中，为了提升整个系统的吞吐量，某个流程将这10个报文封装成了一个包发送给了服务端，导致服务端只接收到了1个数据包

2，导致粘包拆包因素以及解决方案

根据上面的例子，在客户端中，发送了10个包，按理来说就是发送的一个报文对应一个封包，应该会有10个包，而在实际的打印日志中只有一个包，说明就产生了粘包的情况，就是将多个报文包粘在一起了。

拆包因素就是和粘包的相反，粘包是因为每个数据报文太小，而将多个包合成一个数据包。拆包就是因为单个包的报文太大了，如单个包的大小为2000字节，超过了tcp最大1460字节包大小，游戏需要分成两段报文包发送给对端，即一个数据包多次发送。这样就需要合并两个报文包下面的同一段报文。

这就是经典的生产者消费者问题了，客户端的写buffer对应的就是生产者，服务端的读buffer对应的就是消费者

生产者为了提高整个系统的效率，以IPv4为例，tcp每个报文最大的长度是1460个字节，假设客户端这边连续上传10个这种100字节的数据报文，总大小也在1000字节，那么客户端就会认为这10个报文我一次就可以发送给对端，那么这10个包就被黏在一起了 。

当然在tcp层中，tcp的粘包拆包问题并不是其本身的缺陷，而是内部的一种机制，就是说tcp内部是不知道应用层每个包的大小，多少个包之类的，也不会对包与包之间的边界处理，因此只能在应用层或者通过相关协议去做一些限制。

2.1. 各个包之间没有设置边界感导致

举个例子，就像我们平常时开发的批量删除中，如果要删除多个id，一个是直接用数组将id从前端传给后端，但是现在不考虑这种方式；另一种是通过逗号或者其他的分隔符拼接成字符串将数据传给后端，如下面这段这种格式，这样后台先去解析这段字符串，然后再将对应的id进行删除。对于后台来讲，整个数据的边界感就是逗号 ,

"1,2,3,4,5,6"

但是在netty中，buffer缓冲区在合并包时是没有边界感的，就是不能像我们认为一样手动的去添加这种分隔符等，因此这就可能出现多个包粘成一个包的情况，最后不能对这些请求做出正确的响应。就像上面的这段代码，按理来说会有10个报文，并且在客户端这边接收到10个响应，但是最终在客户端这边只接收到了一个响应数据。由于客户端这边没有给实际的边界感，当服务端接收到数据时，也不能根据对应的边界做处理，只能将整包一起处理并响应。

2.2. 服务端度缓冲区数据处理导致

在服务段的readBuf中读取到数据时，如果在应用层没有设置边界，那么服务端也可能会根据滑动串口读取固定的数据，那么也可能会使得数据出现粘包情况。并且如果服务出现阻塞情况，所有数据都挤压在一起，那么也会导致出现粘包的情况

即使说通过设置 TCP_NODELAY 这个在客户端那边无延迟的情况，就是来一个报文立马发送给对端，也可能因为收到服务端这边阻塞或者没设这边界的情况，出现粘包。即禁用 Nagle 算法

.option(ChannelOption.TCP_NODELAY, true)  // 禁用 Nagle 算法

2.3，粘包拆包解决方案

既然知道粘包的主要原因是tcp对包与包之间的边界无感知，那么解决方案就呼之欲出了，那就是加边界呗。

2.3.1，分隔符设置边界

可以直接加换行符，也可以自定义边界。如以换行符为边界的代码如下，只需要在加pipeline之前，加一个 LineBasedFrameDecoder 的对象即可，服务端和客户端都需要加上这句 **.addLast(new LineBasedFrameDecoder(26)) **。服务端和客户端两边都要加以下这段代码

socketChannel.pipeline()
	.addLast(new LineBasedFrameDecoder(26)) 		//设置边界
	.addLast(new StickPackageServerHandler());     //将事件加入到管道中

也可以自定义边界分隔符，每个报文之间通过这个 @_ 设置边界符

ByteBuf delimiter = Unpooled.copiedBuffer("@_".getBytes());
socketChannel.pipeline()
	.addLast(new DelimiterBasedFrameDecoder(26,delimiter))
    .addLast(new StickPackageClientHandler());

如果设置了自定义边界，在客户端对应的handler中，也需要将发送的内容后面拼接一个 @_ ，这样服务端在接收到数据之后可以直接根据这个自定义的边界获取以及处理相关数据了

String request = "abcdefghijklmnopqrstuvwxyz" + "@_";

服务端这边打印的日志详情如下，通过这种边界设置对应文本的方式确实解决了这种粘包的问题

[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为1
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为2
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为3
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为4
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为5
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为6
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为7
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为8
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为9
[nioEventLoopGroup-2-2] INFO c.r.w.c.n.s.StickPackageServerHandler - [channelRead,20] - 服务端接收到请求数量为10

2.3.2，固定长度解码器

使用分隔符设置边界确实可以解决粘包问题，但是只适用于一些文本类型的消息，如果是使用这种二进制流数据，那么上面的加字符分割的方式就不好使了，那么就可以服务端和客户端两边约定好定长的数据格式进行分界了。

在服务端的pipeline中加上一个 FixedLengthFrameDecoder 类，并将长度设置为客户端报文请求长度

new FixedLengthFrameDecoder(FixedLengthEchoClient.REQUEST.length())

服务端中通过下面这段代码将响应数的长度放回给客户端

ctx.writeAndFlush(Unpooled.copiedBuffer(FixedLengthEchoServer.RESPONSE.getBytes()))

客户端中也加上这个 FixedLengthFrameDecoder 定长实现类，设置响应段的长度

new FixedLengthFrameDecoder(FixedLengthEchoServer.RESPONSE.length())

客户端的handler中在发送数据时，通过一下两行代码发送数据

msg = Unpooled.buffer(FixedLengthEchoClient.REQUEST.length());
msg.writeBytes(FixedLengthEchoClient.REQUEST.getBytes());

2.3.3，使用长度字段解码器

这个看起来和上面那个好像，上面那个是固定长度的解码器，但是这个使用的是动态的长度字段解码器，就是每个包的大小都告诉服务端，服务端根据一些内部的偏移量等去解析数据

pipeline.addLast(new LengthFieldBasedFrameDecoder(
    1024,   // 最大帧长度
    0,      // 长度字段的偏移量
    4,      // 长度字段的字节数
    0,      // 长度字段的调整值
    4       // 跳过长度字段的字节数
));

如使下面这段示例，基于长度字段进行拆分帧，并且在发送消息时，在消息头自动加上4个字节长度

// 解码器: 基于长度字段拆分帧
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
// 编码器: 在发送消息时，自动在消息前加上长度字段
ch.pipeline().addLast(new LengthFieldPrepender(4));

在客户端的handler中，通过 Unpooled.copiedBuffer 将数据封装成 ByteBuf 传递

String message = "Hello from client";
ByteBuf buf = Unpooled.copiedBuffer(message.getBytes());

服务端在响应的时候也需要加上这段，将数据响应给客户端

ByteBuf response = Unpooled.copiedBuffer("Message received".getBytes());
ctx.writeAndFlush(response);