TCP粘包和拆包、编码解码器

ゝ一纸荒年。 2022-04-25 02:40 430阅读 0赞

在进行Netty学习时，发现，如果客户端连续不断的向服务端发送数据包时，服务端接收的数据会出现两个数据包粘在一起的情况，这就是TCP协议中经常会遇到的粘包以及拆包的问题。

我们都知道TCP属于传输层的协议，传输层除了有TCP协议外还有UDP协议。那么UDP是否会发生粘包或拆包的现象呢？答案是不会。UDP是基于报文发送的，从UDP的帧结构可以看出，在UDP首部采用了16bit来指示UDP数据报文的长度，因此在应用层能很好的将不同的数据报文区分开，从而避免粘包和拆包的问题。而TCP是基于字节流的，虽然应用层和TCP传输层之间的数据交互是大小不等的数据块，但是TCP把这些数据块仅仅看成一连串无结构的字节流，没有边界；另外从TCP的帧结构也可以看出，在TCP的首部没有表示数据长度的字段，基于上面两点，在使用TCP传输数据时，才有粘包或者拆包现象发生的可能。

> Tcp是个“流”协议，所谓流就是没有界限的一串数据。可以类比一下水流，没有分极限。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分。所以一个业务数据可能被TCP拆分成多个包进行发送，也有可能把多个小的数据包封装成一个大的数据包发送。这就是TCP的拆包和粘包。

### 粘包、拆包表现形式 ###

现在假设客户端向服务端连续发送了两个数据包，用packet1和packet2来表示，那么服务端收到的数据可以分为三种，现列举如下：

第一种情况，接收端正常收到两个数据包，即没有发生拆包和粘包的现象，此种情况不在本文的讨论范围内。

![20190430170834698.png][]

第二种情况，接收端只收到一个数据包，由于TCP是不会出现丢包的，所以这一个数据包中包含了发送端发送的两个数据包的信息，这种现象即为粘包。这种情况由于接收端不知道这两个数据包的界限，所以对于接收端来说很难处理。

![20190430170859818.png][]

第三种情况，这种情况有两种表现形式，如下图。接收端收到了两个数据包，但是这两个数据包要么是不完整的，要么就是多出来一块，这种情况即发生了拆包和粘包。这两种情况如果不加特殊处理，对于接收端同样是不好处理的。

![20190430170918439.png][]

![20190430170941582.png][]

**粘包、拆包发生原因**

发生TCP粘包或拆包有很多原因，现列出常见的几点，可能不全面，欢迎补充，

*  1、要发送的数据大于TCP发送缓冲区剩余空间大小，将会发生拆包。
 *  2、待发送数据大于MSS（最大报文长度），TCP在传输前将进行拆包。
 *  3、要发送的数据小于TCP发送缓冲区的大小，TCP将多次写入缓冲区的数据一次发送出去，将会发生粘包。
 *  4、接收数据端的应用层没有及时读取接收缓冲区中的数据，将发生粘包。

**粘包、拆包解决办法**

通过以上分析，我们清楚了粘包或拆包发生的原因，那么如何解决这个问题呢？解决问题的关键在于如何给每个数据包添加边界信息，常用的方法有如下几个：

>  *  1、发送端给每个数据包添加包首部（类似于UDP），首部中应该包含数据包的长度，这样接收端在接收到数据后，通过读取包首部的长度字段，便知道每一个数据包的实际长度了。
>  *  2、**消息定长：**发送端将每个数据包封装为固定长度（不够的可以通过补0填充），这样接收端每次从接收缓冲区中读取固定长度的数据就自然而然的把每个数据包拆分开来。
>  *  3、**包尾添加分隔符：**可以在数据包之间设置边界，如添加特殊符号，这样，接收端通过这个边界就可以将不同的数据包拆分开。

### Nagle算法 ###

nagle算法用于处理小报文段（微小分组）的发送问题

>     nagle算法的核心思想是允许网络中最多只能有一个小分组被发送，而待发送的其它小分组会被重新分组成一个”较大的”小分组，等收到上一个小分组的应答后再发送

nagle算法可以减少网络中微小分组的数量，比如客户端需要依次向服务器发送大小为1,2,3,1,2字节的5个分组。

*  在没有开启nagle算法的情况下，这些小分组会被依次发送（不需要等待上一个小分组的应答，因为没启动nagle），总共发送的报文段（分组）个数为5。
 *  当开启nagle算法时，客户端首先发送大小为1字节的第一个分组，随后其它分组到达发送缓冲区，由于上一个分组的应答还没有收到，所以TCP会先缓存新来的这4个小分组，并将其重新分组，组成一个大小为8(2+3+1+2)字节的”较大的”小分组。当第一个小分组的应答收到后，客户端将这个8字节的分组发送。总共发送的报文段（分组）个数为2

可以看到，当传输数据存在大量交互数据时，nagle算法可以有效减少网络中的报文段个数。

所以，Nagle算法会产生TCP粘包半包问题

## Netty粘包和拆包解决方案 ##

Netty提供了多个解码器，可以进行分包的操作，分别是：

**\* LineBasedFrameDecoder （换行）**

LineBasedFrameDecoder是回车换行解码器，如果用户发送的消息以回车换行符作为消息结束的标识，则可以直接使用Netty的LineBasedFrameDecoder对消息进行解码，只需要在初始化Netty服务端或者客户端时将LineBasedFrameDecoder正确的添加到ChannelPipeline中即可，不需要自己重新实现一套换行解码器。  
       LineBasedFrameDecoder的工作原理是它依次遍历ByteBuf中的可读字节，判断看是否有“\\n”或者“\\r\\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。防止由于数据报没有携带换行符导致接收到ByteBuf无限制积压，引起系统内存溢出。

**\* DelimiterBasedFrameDecoder（添加特殊分隔符报文来分包）**

DelimiterBasedFrameDecoder是分隔符解码器，用户可以指定消息结束的分隔符，它可以自动完成以分隔符作为码流结束标识的消息的解码。 回车换行解码器实际上是一种特殊的DelimiterBasedFrameDecoder解码器。

**\* FixedLengthFrameDecoder（使用定长的报文来分包）**

FixedLengthFrameDecoder是固定长度解码器，它能够按照指定的长度对消息进行自动解码，开发者不需要考虑TCP的粘包/拆包等问题，非常实用。  
       对于定长消息，如果消息实际长度小于定长，则往往会进行补位操作，它在一定程度上导致了空间和资源的浪费。但是它的优点也是非常明显的，编解码比较简单，因此在实际项目中仍然有一定的应用场景。

### 实例： ###

客户端（数据尾部以回车结束）：

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;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class LineBaseClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf msg = null;
            String request = "Mark,Lison,Peter,James,Deer"
                    + System.getProperty("line.separator");
            for(int i=0;i<100;i++){
                msg = Unpooled.buffer(request.length());
                msg.writeBytes(request.getBytes());
                ctx.writeAndFlush(msg);
            }
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

import io.netty.bootstrap.Bootstrap;
    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.LineBasedFrameDecoder;
    
    import java.net.InetSocketAddress;
    
    
    public class LineBaseEchoClient {
    
        private final String host;
    
        public LineBaseEchoClient(String host) {
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        .remoteAddress(new InetSocketAddress(host,LineBaseEchoServer.PORT))/*配置要连接服务器的ip地址和端口*/
                        .handler(new ChannelInitializerImp());
                ChannelFuture f = b.connect().sync();
                System.out.println("已连接到服务器.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                //	加分割符：系统回车符
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new LineBaseClientHandler());
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new LineBaseEchoClient("127.0.0.1").start();
        }
    }

服务器端：

import io.netty.bootstrap.ServerBootstrap;
    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.LineBasedFrameDecoder;
    
    import java.net.InetSocketAddress;
    
    public class LineBaseEchoServer {
    
        public static final int PORT = 9998;
    
        public static void main(String[] args) throws InterruptedException {
            LineBaseEchoServer lineBaseEchoServer = new LineBaseEchoServer();
            System.out.println("服务器即将启动");
            lineBaseEchoServer.start();
        }
    
        public void start() throws InterruptedException {
            final LineBaseServerHandler serverHandler = new LineBaseServerHandler();
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
                b.group(group)/*将线程组传入*/
                    .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                    .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                    /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                    所以下面这段代码的作用就是为这个子channel增加handle*/
                    .childHandler(new ChannelInitializerImp());
                ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
                System.out.println("服务器启动完成，等待客户端的连接和数据.....");
                f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
            } finally {
                group.shutdownGracefully().sync();/*优雅关闭线程组*/
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));  // 以系统定义的换行符来区分
                ch.pipeline().addLast(new LineBaseServerHandler());
            }
        }
    }

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    @ChannelHandler.Sharable
    public class LineBaseServerHandler extends ChannelInboundHandlerAdapter {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf in = (ByteBuf)msg;
            String request = in.toString(CharsetUtil.UTF_8);
            System.out.println("Server Accept["+request
                    +"] and the counter is:"+counter.incrementAndGet());
            String resp = "Hello,"+request+". Welcome to Netty World!"
                    + System.getProperty("line.separator");
            ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }

### **定长发送：** ###

**客户端：**

private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                //消息定长
                ch.pipeline().addLast(
                        new FixedLengthFrameDecoder(
                                FixedLengthEchoServer.RESPONSE.length())); //服务器放松数据的长度
                ch.pipeline().addLast(new FixedLengthClientHandler());
            }
        }

**服务器端：**

private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                //消息定长
                ch.pipeline().addLast(
                        new FixedLengthFrameDecoder(
                                FixedLengthEchoClient.REQUEST.length()));//客户端发送数据的长度
                ch.pipeline().addLast(new FixedLengthServerHandler());
            }
        }

# 编码解码器 #

当你通过Netty发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会被**解码**：从字节转换为另一种格式（比如java对象）；如果是出站消息，它会被**编码**成字节。

Netty提供了一系列实用的编码解码器，他们都实现了ChannelInboundHadnler或者ChannelOutcoundHandler接口。在这些类中，channelRead方法已经被重写了。以入站为例，对于每个从入站Channel读取的消息，这个方法会被调用。随后，它将调用由已知解码器所提供的decode()方法进行解码，并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。

Netty提供了丰富的解码器抽象基类，我们可以很容易的实现这些基类来自定义解码器。下面是解码器的一个类型：

*  解码字节到消息：ByteToMessageDecoder
 *  解码消息到消息：MessageToMessageDecoder

码器负责解码“入站”数据从一种格式到另一种格式，解码器处理入站数据是抽象ChannelInboundHandler的实现。实践中使用解码器很简单，就是将入站数据转换格式后传递到ChannelPipeline中的下一个ChannelInboundHandler进行处理；这样的处理时很灵活的，我们可以将解码器放在ChannelPipeline中，重用逻辑。

**ByteToMessageDecoder**

通常你需要将消息从字节解码成消息或者从字节解码成其他的序列化字节。这是一个常见的任务，Netty提供了抽象基类，我们可以使用它们来实现。Netty中提供的ByteToMessageDecoder可以将字节消息解码成POJO对象，下面列出了ByteToMessageDecoder两个主要方法：

>  *  decode(ChannelHandlerContext, ByteBuf, List<Object>)，这个方法是唯一的一个需要自己实现的抽象方法，作用是将ByteBuf数据解码成其他形式的数据。
>  *  decodeLast(ChannelHandlerContext, ByteBuf, List<Object>)，实际上调用的是decode(...)。

看下面ToIntegerDecoder的实现代码：

public class ToIntegerDecoder extends ByteToMessageDecoder {  
    	    @Override  
    	    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {  
    	        if(in.readableBytes() >= 4){  
    	            out.add(in.readInt());  
    	        }  
    	    }  
    	}

** MessageToMessageDecoder**

将消息对象转成消息对象可是使用MessageToMessageDecoder，它是一个抽象类，需要我们自己实现其decode(...)。message-to-message同上面讲的byte-to-message的处理机制一样，看下图：

public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> {  
          @Override  
          protected void decode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {  
             out.add(String.valueOf(msg));  
         }  
     }

### **编码器** ###

解码器的功能正好相反。Netty 提供了一组类，用于帮助你编写具有以下功能的编码器：

*  将消息编码为字节；MessageToByteEncoder
 *  将消息编码为消息：MessageToMessageEncoder<T>，T代表源数据的类型

[20190430170834698.png]: /images/20220210/f86c808171394bf3b68524e1f31f6288.png
[20190430170859818.png]: /images/20220210/13e8facc00694c488d875d84a94efe4c.png
[20190430170918439.png]: /images/20220210/8aa38338ad3d441d88394d0af7174719.png
[20190430170941582.png]: /images/20220210/6555e85c87ce4ae695114edf4412e842.png