1 | Netty源码：粘包与半包处理

「爱情、让人受尽委屈。」 2023-07-01 06:26 50阅读 0赞

> TCP是个流协议，流是一串没有界限的数据。TCP会根据TCP缓冲区的实际情况对包进行划分。因此造成一个完整的业务包，会被TCP分成多个包、把多个包封装成一个大的包进行发送。

## 粘包与拆包现象 ##

![format_png][]

1.  服务端分两次读取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包；
2.  服务端一次接收到了两个数据包，D1和D2粘合在一起，被称为TCP粘包；
3.  服务端分两次读取到了两个数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，这被称为TCP拆包；
4.  服务端分两次读取到了两个数据包，第一次读取到了D1包的部分内容D1\_1，第二次读取到了D1包的剩余内容D1\_2和D2包的整包。

## 产生原因 ##

1.  应用程序write写入的字节**大小/大于**套接口发送缓冲区大小；
2.  进行MSS大小的TCP分段；
3.  以太网帧的payload大于MTU进行IP分片。

![format_png 1][]

对于Linux，发送缓冲区的默认值为：`16384`。可使用下面命令查看：

# 接收
    cat /proc/sys/net/ipv4/tcp_rmem
    # min   default max
    # 4096	87380	6291456 (单位：byte)
    # 4K		85K		6M
    # 发送(单位：byte)
    cat /proc/sys/net/ipv4/tcp_wmem
    # min   default max
    # 4096	16384	4194304  (单位：byte)
    # 4K		16K		4M

*数据来自百度云的云服务器*：

![format_png 2][]

对于MacOS，可参考：`sysctl net.inet.tcp`，但是好像没找到与linux类似的参数。

## 如何解决 ##

![format_png 3][]

## Netty如何解决 ##

![format_png 4][]

Netty中主要是在收到数据后，对数据进行处理解码处理时，根据不同的策略，进行了拆包操作，然后将得到的完整的业务数据包传递给下个处理逻辑。分割前后的逻辑主要在`ByteToMessageDecoder`这个类中。它的继承如下：

![format_png 5][]

每次从TCP缓冲区读到数据都会调用其`channelRead()`方法。这个函数的处理逻辑是：

1.  用累加器`cumulator`将新读入的数据(`ByteBuf`)存储到`cumulation`中；
2.  调用解码器

![format_png 6][]

### 累加器 ###

存在两个累加器，`MERGE_CUMULATOR`和`COMPOSITE_CUMULATOR`。默认的是前者，即:`private Cumulator cumulator = MERGE_CUMULATOR;`。

`MERGE_CUMULATOR`会先判断是否需要扩容，然后再将收到的msg拷贝到`cumulation`中。

/**
        * Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
        */
    public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
        
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        
            try {
        
                final int required = in.readableBytes();
                if (required > cumulation.maxWritableBytes() ||
                        (required > cumulation.maxFastWritableBytes() && cumulation.refCnt() > 1) ||
                        cumulation.isReadOnly()) {
        
                    // Expand cumulation (by replacing it) under the following conditions:
                    // - cumulation cannot be resized to accommodate the additional data
                    // - cumulation can be expanded with a reallocation operation to accommodate but the buffer is
                    //   assumed to be shared (e.g. refCnt() > 1) and the reallocation may not be safe.
                    return expandCumulation(alloc, cumulation, in);
                }
                return cumulation.writeBytes(in);
            } finally {
        
                // We must release in in all cases as otherwise it may produce a leak if writeBytes(...) throw
                // for whatever release (for example because of OutOfMemoryError)
                in.release();
            }
        }
    };

扩容的过程是先得到一个能够容纳下原数据+当前数据的收集器，然后将原数据和当前数据依次拷贝进入收集器，最后释放旧的收集器里面的数据。

private static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf oldCumulation, ByteBuf in) {
        
        ByteBuf newCumulation = alloc.buffer(alloc.calculateNewCapacity(
                oldCumulation.readableBytes() + in.readableBytes(), MAX_VALUE));
        ByteBuf toRelease = newCumulation;
        try {
        
            newCumulation.writeBytes(oldCumulation);
            newCumulation.writeBytes(in);
            toRelease = oldCumulation;
            return newCumulation;
        } finally {
        
            toRelease.release();
        }
    }

`COMPOSITE_CUMULATOR`是将每个新收到的消息，作为一个`Component`存储到收集器`CompositeByteBuf`中的`components`数组中。

/**
        * Cumulate {@link ByteBuf}s by add them to a {@link CompositeByteBuf} and so do no memory copy whenever possible.
        * Be aware that {@link CompositeByteBuf} use a more complex indexing implementation so depending on your use-case
        * and the decoder implementation this may be slower then just use the {@link #MERGE_CUMULATOR}.
        */
    public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
        
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        
            try {
        
                if (cumulation.refCnt() > 1) {
        
                    // Expand cumulation (by replace it) when the refCnt is greater then 1 which may happen when the
                    // user use slice().retain() or duplicate().retain().
                    //
                    // See:
                    // - https://github.com/netty/netty/issues/2327
                    // - https://github.com/netty/netty/issues/1764
                    return expandCumulation(alloc, cumulation, in);
                }
                final CompositeByteBuf composite;
                if (cumulation instanceof CompositeByteBuf) {
        
                    composite = (CompositeByteBuf) cumulation;
                } else {
        
                    composite = alloc.compositeBuffer(MAX_VALUE);
                    composite.addComponent(true, cumulation);
                }
                composite.addComponent(true, in);
                in = null;
                return composite;
            } finally {
        
                if (in != null) {
        
                    // We must release if the ownership was not transferred as otherwise it may produce a leak if
                    // writeBytes(...) throw for whatever release (for example because of OutOfMemoryError).
                    in.release();
                }
            }
        }
    };

### 拆包解码流程 ###

`callDecode()`方法中的`decodeRemovalReentryProtection()`将调用`decode()`方法，其中`decode()`是一个抽象方法，由子类去实现。主要的子类有：

#### `FixedLengthFrameDecoder` ####

里面有一个属性叫`frameLength`，用来表示消息的长度。

A decoder that splits the received ByteBufs by the fixed number of bytes. For example, if you received the following four fragmented packets:
       +---+----+------+----+
       | A | BC | DEFG | HI |
       +---+----+------+----+
    
    A FixedLengthFrameDecoder(3) will decode them into the following three packets with the fixed length:
       +-----+-----+-----+
       | ABC | DEF | GHI |
       +-----+-----+-----+

流程也比较简单，收集器里面的数据长度够`frameLength`，就从收集器中截取`frameLength`byte，然后返回一个新的`ByteBuf`。

@Override
    protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        
        Object decoded = decode(ctx, in);
        if (decoded != null) {
        
            out.add(decoded);
        }
    }
    
    /**
     * Create a frame out of the {@link ByteBuf} and return it.
     *
     * @param   ctx             the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
     * @param   in              the {@link ByteBuf} from which to read data
     * @return  frame           the {@link ByteBuf} which represent the frame or {@code null} if no frame could
     *                          be created.
     */
    protected Object decode(
            @SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        
        if (in.readableBytes() < frameLength) {
        
            return null;// 长度不够，此次decode不产生消息
        } else {
        
            return in.readRetainedSlice(frameLength);
        }
    }

**有一个问题，如果一次收到的数据长度为`2 * frameLength`，且这个数据是最后一个数据，那么是否存在解码出现异常的情况？**

1.  有一个循环  
    ![format_png 7][]
2.  输入结束的时候再次调用解码  
    ![format_png 8][]

#### `LineBasedFrameDecoder` ####

流程是先找到当前消息中的换行符，存在且没有超过最大长度，返回解释到的数据。

#### `DelimiterBasedFrameDecoder` ####

根据特定的字符进行分割，其中如果分割符是行标志，会调用`LineBasedFrameDecoder`进行分割解码。

// decode()方法中
    if (lineBasedDecoder != null) {
        
    	return lineBasedDecoder.decode(ctx, buffer);
    }
    
    // lineBasedDecoder不为空的情况是分割字符是行分割字符
    // 构造方法中
    if (isLineBased(delimiters) && !isSubclass()) {
        
    	lineBasedDecoder = new LineBasedFrameDecoder(maxFrameLength, stripDelimiter, failFast);
    	this.delimiters = null;
    }

判断分割符是否为行分割符的代码如下：

private static boolean isLineBased(final ByteBuf[] delimiters) {
        
    	if (delimiters.length != 2) {
        
    		return false;
    	}
    	ByteBuf a = delimiters[0];
    	ByteBuf b = delimiters[1];
    	if (a.capacity() < b.capacity()) {
        
    		a = delimiters[1];
    		b = delimiters[0];
    	}
    	return a.capacity() == 2 && b.capacity() == 1
    		&& a.getByte(0) == '\r' && a.getByte(1) == '\n'
    		&& b.getByte(0) == '\n';
    }

因为分割字符可能是多个，当数据中存在多个分割字符的情况下，会用分割后得到的数据最短的那个分割字符。如下：

// Try all delimiters and choose the delimiter which yields the shortest frame.
    int minFrameLength = Integer.MAX_VALUE;
    ByteBuf minDelim = null;
    for (ByteBuf delim: delimiters) {
        
    	int frameLength = indexOf(buffer, delim);
    	if (frameLength >= 0 && frameLength < minFrameLength) {
        
    		minFrameLength = frameLength;
    		minDelim = delim;
    	}
    }

For example, if you have the following data in the buffer:  
±-------------+  
| ABC\\nDEF\\r\\n |  
±-------------+

a DelimiterBasedFrameDecoder(Delimiters.lineDelimiter()) will choose ‘\\n’ as the first delimiter and produce two frames:  
±----±----+  
| ABC | DEF |  
±----±----+

rather than incorrectly choosing ‘\\r\\n’ as the first delimiter:  
±---------+  
| ABC\\nDEF |  
±---------+

#### `LengthFieldBasedFrameDecoder` ####

简而言之，就是在数据的头部，放一个专门的长度位，根据长度位来读取后面信息的内容。

这个类比较有意思，注释差不多占了2/5。主要的处理逻辑是`decode()`，但是这个方法100行都不到。注释主要解释了这个类里面几个参数的不同配置，产生不同的处理情况。

![format_png 9][]

情况对应于下表：

<table> 
 <thead> 
  <tr> 
   <th align="center"></th> 
   <th align="center">lengthFieldOffset</th> 
   <th align="center">lengthFieldLength</th> 
   <th align="center">lengthAdjustment</th> 
   <th align="center">initialBytesToStrip</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">0x01</td> 
   <td align="center">0</td> 
   <td align="center">2</td> 
   <td align="center">0</td> 
   <td align="center">0</td> 
  </tr> 
  <tr> 
   <td align="center">0x02</td> 
   <td align="center">0</td> 
   <td align="center">2</td> 
   <td align="center">0</td> 
   <td align="center">2</td> 
  </tr> 
  <tr> 
   <td align="center">0x03</td> 
   <td align="center">0</td> 
   <td align="center">2</td> 
   <td align="center">-2</td> 
   <td align="center">0</td> 
  </tr> 
  <tr> 
   <td align="center">0x04</td> 
   <td align="center">2</td> 
   <td align="center">3</td> 
   <td align="center">0</td> 
   <td align="center">0</td> 
  </tr> 
  <tr> 
   <td align="center">0x05</td> 
   <td align="center">0</td> 
   <td align="center">3</td> 
   <td align="center">2</td> 
   <td align="center">0</td> 
  </tr> 
  <tr> 
   <td align="center">0x06</td> 
   <td align="center">1</td> 
   <td align="center">2</td> 
   <td align="center">1</td> 
   <td align="center">3</td> 
  </tr> 
  <tr> 
   <td align="center">0x07</td> 
   <td align="center">1</td> 
   <td align="center">2</td> 
   <td align="center">-3</td> 
   <td align="center">3</td> 
  </tr> 
 </tbody> 
</table>

##### 0x01 #####

`lengthFieldLength = 2`表示长度位占头部的2 bytes，剩下的都是消息占位，也就是`0x000C(12) + 2 = 14`。

![format_png 10][]

##### 0x02 #####

与`0x01`类似，只是多了`initialBytesToStrip = 2`，解码后的内容截取掉了头部的`initialBytesToStrip`位。也就是解码后的长度为`14 - initialBytesToStrip = 12`。

![format_png 11][]

##### 0x03 #####

这种情况下，长度位的值，表示整个包的长度，包括长度位本身的长度。`lengthAdjustment = -2`表示要将长度位的值加上`lengthAdjustment`,作为消息的长度。

![format_png 12][]

##### 0x04 #####

与`0x01`相比，多了个一个长度位的偏移量`lengthFieldOffset`。所以长度位的前面又可以放一些其他数据。也就是说，真正的消息是从`lengthFieldOffset + lengthFieldLength`后开始。

![format_png 13][]

##### 0x05 #####

与`0x03`对比，只是`lengthAdjustment`的正负不同，也就意味着真实的消息是在长度位后面是有偏移的，而偏移出来的空间，可以用作存放另外一种数据类型。

![format_png 14][]

##### 0x06 #####

在`0x04`、`0x05`的基础上，长度位多了偏移`lengthFieldOffset`，真实的消息的偏移又多加了一个`lengthAdjustment`，然后截掉了头部开始的`initialBytesToStrip`bytes。

![format_png 15][]

##### 0x07 #####

在`0x06`的基础上，`lengthAdjustment`变成负数了，与`0x03`的情况类似。

![format_png 16][]

**整体代码的流程**  
除去异常处理的情况，就是计算整个消息的长度，然后跳过要求跳过的字节数，再从`ByteBuf`中读取消息。如下：

![format_png 17][]

参考：

[《Netty权威指南》][Netty]

[netty源码分析之拆包器的奥秘][netty]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAxOTEyMzAxNDExMjgucG5n?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAxOTEyMzAxNTIwMzAucG5n?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAxOTEyMzAxNTE2NDcucG5n?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTMwOTE1NDAucG5n?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTMwOTE4MzYucG5n?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAxOTEyMzAxNTM4MDgucG5n?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAxOTEyMzAxNTUxNDAucG5n?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExNDQyMjgucG5n?x-oss-process=image/format,png
[format_png 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExNDQ0NDAucG5n?x-oss-process=image/format,png
[format_png 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTEwOTI5NDcucG5n?x-oss-process=image/format,png
[format_png 10]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTAyMDcucG5n?x-oss-process=image/format,png
[format_png 11]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTA2NTIucG5n?x-oss-process=image/format,png
[format_png 12]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTQyNTgucG5n?x-oss-process=image/format,png
[format_png 13]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTUyMjMucG5n?x-oss-process=image/format,png
[format_png 14]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTU4MTMucG5n?x-oss-process=image/format,png
[format_png 15]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMTU2MTcucG5n?x-oss-process=image/format,png
[format_png 16]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExMjA5MTkucG5n?x-oss-process=image/format,png
[format_png 17]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvMjAyMDAxMTExNDMxMDkucG5n?x-oss-process=image/format,png
[Netty]: https://book.douban.com/subject/26373138/
[netty]: https://mp.weixin.qq.com/s/526p5f9fgtZu7yYq5j7LiQ