# 精尽 Netty 源码解析 —— Codec 之 ByteToMessageDecoder（一）Cumulator
# 1. 概述
在 [《精尽 Netty 源码解析 —— ChannelHandler（一）之简介》](http://svip.iocoder.cn/Netty/ChannelHandler-1-intro) 中，我们看了 ChannelHandler 的核心类图，如下：[](http://static.iocoder.cn/images/Netty/2018_10_01/01.png)核心类图
- **绿框**部分，我们可以看到，Netty 基于 ChannelHandler 实现了读写的数据( 消息 )的编解码。
> Codec( 编解码 ) = Encode( 编码 ) + Decode( 解码 )。
- 图中有五个和 Codec 相关的类，整理如下：
- 😈 ，实际应该是六个，漏画了 MessageToMessageDecoder 类。
- ByteToMessageCodec ，ByteToMessageDecoder + MessageByteEncoder 的
组合
。
- ByteToMessageDecoder ，将字节**解码**成消息。
- MessageByteEncoder ，将消息**编码**成字节。
- MessageToMessageCodec ，MessageToMessageDecoder + MessageToMessageEncoder 的
组合
。
- MessageToMessageDecoder ，将消息**解码**成另一种消息。
- MessageToMessageEncoder ，将消息**编码**成另一种消息。
------
而本文，我们来分享 ByteToMessageDecoder 部分的内容。
# 2. ByteToMessageDecoder 核心类图
[](http://static.iocoder.cn/images/Netty/2018_12_01/01.png)核心类图
ByteToMessageDecoder 本身是个**抽象**类，其下有多个子类，笔者简单整理成三类，可能不全哈：
- **绿框**部分 FrameDecoder ：消息帧( Frame )解码器。也就是说该类解码器，用于处理 TCP 的**粘包**现象，将网络发送的字节流解码为具有确定含义的消息帧。之后的解码器再将消息帧解码为实际的 POJO 对象。 如下图所示：[decode](http://static.iocoder.cn/images/Netty/2018_12_01/02.png)
- 黄框
部分，将字节流使用
指定序列化方式
反序列化成
消息
，例如：XML、JSON 等等。
- 对于该类解码器，不处理 TCP 的**粘包**现象，所以需要搭配 FrameDecoder 一起使用。
- 蓝框
部分，将字节流
解压
，主要涉及相关压缩算法，例如：GZip、BZip 等等。
- 对于该类解码器，不处理 TCP 的**粘包**现象，所以需要搭配 FrameDecoder 一起使用。
# 3. 为什么要粘包拆包
😈 因为有些朋友不了解粘包和拆包的概念和原理，这里引用笔者的基友【闪电侠】在 [《netty 源码分析之拆包器的奥秘》](https://www.jianshu.com/p/dc26e944da95) 对这块的描述。
## 3.1 为什么要粘包
> 首先你得了解一下 TCP/IP 协议，在用户数据量非常小的情况下，极端情况下，一个字节，该 TCP 数据包的有效载荷非常低，传递 100 字节的数据，需要 100 次TCP传送， 100 次ACK，在应用及时性要求不高的情况下，将这 100 个有效数据拼接成一个数据包，那会缩短到一个TCP数据包，以及一个 ack ，有效载荷提高了，带宽也节省了。
>
> 非极端情况，有可能**两个**数据包拼接成一个数据包，也有可能**一个半**的数据包拼接成一个数据包，也有可能**两个半**的数据包拼接成一个数据包。
## 3.2 为什么要拆包
> 拆包和粘包是相对的，一端粘了包，另外一端就需要将粘过的包拆开。举个栗子，发送端将三个数据包粘成两个TCP数据包发送到接收端，接收端就需要根据应用协议将两个数据包重新组装成三个数据包。
>
> 还有一种情况就是用户数据包超过了 mss(最大报文长度)，那么这个数据包在发送的时候必须拆分成几个数据包，接收端收到之后需要将这些数据包粘合起来之后，再拆开。
## 3.3 拆包的原理
> 数据，每次读取完都需要判断是否是一个完整的数据包：
>
> 1. 如果当前读取的数据不足以拼接成一个完整的业务数据包，那就保留该数据，继续从tcp缓冲区中读取，直到得到一个完整的数据包。
> 2. 如果当前读到的数据加上已经读取的数据足够拼接成一个数据包，那就将已经读取的数据拼接上本次读取的数据，够成一个完整的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下次读到的数据尝试拼接。
# 4. Cumulator
Cumulator ，是 ByteToMessageDecoder 的**内部**接口。中文翻译为“累加器”，用于将读取到的数据进行累加到一起，然后再尝试**解码**，从而实现**拆包**。
也是因为 Cumulator 的累加，所以能将不完整的包累加到一起，从而完整。当然，累加的过程，没准又进入了一个不完整的包。所以，这是一个不断累加，不断解码拆包的过程。
------
Cumulator 接口，代码如下：
```
/**
* ByteBuf 累积器接口
*
* Cumulate {@link ByteBuf}s.
*/
public interface Cumulator {
/**
* Cumulate the given {@link ByteBuf}s and return the {@link ByteBuf} that holds the cumulated bytes.
* The implementation is responsible to correctly handle the life-cycle of the given {@link ByteBuf}s and so
* call {@link ByteBuf#release()} if a {@link ByteBuf} is fully consumed.
*
* @param alloc ByteBuf 分配器
* @param cumulation ByteBuf 当前累积结果
* @param in 当前读取( 输入 ) ByteBuf
* @return ByteBuf 新的累积结果
*/
ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
}
```
- 对于 `Cumulator#cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in)` 方法，将**原有** `cumulation` 累加上**新的** `in` ，返回“新”的 ByteBuf 对象。
- 如果 `in` 过大，超过 `cumulation` 的空间上限，使用 `alloc` 进行扩容后再累加。
------
Cumulator 有两个实现类，代码如下：
```
public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
// ... 省略代码
}
public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
// ... 省略代码
}
```
两者的累加方式不同，我们来详细解析。
## 4.1 MERGE_CUMULATOR
`MERGE_CUMULATOR` 思路是，不断使用**老的** ByteBuf 累积。如果空间不够，扩容出**新的** ByteBuf ，再继续进行累积。代码如下：
```
// ByteToMessageDecoder.java
/**
* Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
*/
1: public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
2:
3: @Override
4: public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
5: final ByteBuf buffer;
6: if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes() // 超过空间大小，需要扩容
7: || cumulation.refCnt() > 1 // 引用大于 1 ，说明用户使用了 slice().retain() 或 duplicate().retain() 使refCnt增加并且大于 1 ，
8: // 此时扩容返回一个新的累积区ByteBuf，方便用户对老的累积区ByteBuf进行后续处理。
9: || cumulation.isReadOnly()) { // 只读，不可累加，所以需要改成可写
10: // Expand cumulation (by replace it) when either there is not more room in the buffer
11: // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
12: // duplicate().retain() or if its read-only.
13: //
14: // See:
15: // - https://github.com/netty/netty/issues/2327
16: // - https://github.com/netty/netty/issues/1764
17: // 扩容，返回新的 buffer
18: buffer = expandCumulation(alloc, cumulation, in.readableBytes());
19: } else {
20: // 使用老的 buffer
21: buffer = cumulation;
22: }
23: // 写入 in 到 buffer 中
24: buffer.writeBytes(in);
25: // 释放输入 in
26: in.release();
27: // 返回 buffer
28: return buffer;
29: }
30:
31: };
```
- 获取 `buffer` 对象。
- 第 6 至 9 行：如下三个条件，满足任意，需要进行扩容。
- ① 第 6 行：
```
cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
```
，超过空间大小，需要扩容。
- 这个比较好理解。
- ② 第 7 行：
```
cumulation.refCnt() > 1
```
，引用大于 1 ，说明用户使用了
```
ByteBuf#slice()#retain()
```
或
```
ByteBuf#duplicate()#retain()
```
方法，使
```
refCnt
```
增加并且大于 1 。
- 关于这块，在【第 11 行】的英文注释，也相应的提到。
- ③ 第 9 行：只读，不可累加，所以需要改成可写。
- 这个比较好理解。
- 【需要扩容】第 18 行：调用 `ByteToMessageDecoder#expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable)` **静态**方法，扩容，并返回新的，并赋值给 `buffer` 。代码如下：
```
static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
// 记录老的 ByteBuf 对象
ByteBuf oldCumulation = cumulation;
// 分配新的 ByteBuf 对象
cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
// 将老的数据，写入到新的 ByteBuf 对象
cumulation.writeBytes(oldCumulation);
// 释放老的 ByteBuf 对象
oldCumulation.release();
// 返回新的 ByteBuf 对象
return cumulation;
}
```
- 标准的扩容，并复制老数据的过程。胖友自己看下注释噢。
- 【无需扩容】第 21 行：`buffer` 直接使用的 `cumulation` 对象。
- 第 24 行：写入
```
in
```
到
```
buffer
```
中，进行累积。
- 第 26 行：释放 `in` 。
- 第 28 行：返回 `buffer` 。
## 4.2 COMPOSITE_CUMULATOR
`COMPOSITE_CUMULATOR` 思路是，使用 CompositeByteBuf ，组合新输入的 ByteBuf 对象，从而避免内存拷贝。代码如下：
```
// ByteToMessageDecoder.java
/**
* 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}.
*
* 相比 MERGE_CUMULATOR 来说：
*
* 好处是，内存零拷贝
* 坏处是，因为维护复杂索引，所以某些使用场景下，慢于 MERGE_CUMULATOR
*/
1: public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
2:
3: @Override
4: public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
5: ByteBuf buffer;
6: // 和 MERGE_CUMULATOR 的情况类似
7: if (cumulation.refCnt() > 1) {
8: // Expand cumulation (by replace it) when the refCnt is greater then 1 which may happen when the user
9: // use slice().retain() or duplicate().retain().
10: //
11: // See:
12: // - https://github.com/netty/netty/issues/2327
13: // - https://github.com/netty/netty/issues/1764
14: buffer = expandCumulation(alloc, cumulation, in.readableBytes());
15: buffer.writeBytes(in);
16: in.release();
17: } else {
18: CompositeByteBuf composite;
19: // 原来是 CompositeByteBuf 类型，直接使用
20: if (cumulation instanceof CompositeByteBuf) {
21: composite = (CompositeByteBuf) cumulation;
22: // 原来不是 CompositeByteBuf 类型，创建，并添加到其中
23: } else {
24: composite = alloc.compositeBuffer(Integer.MAX_VALUE);
25: composite.addComponent(true, cumulation);
26: }
27: // 添加 in 到 composite 中
28: composite.addComponent(true, in);
29: // 赋值给 buffer
30: buffer = composite;
31: }
32: // 返回 buffer
33: return buffer;
34: }
35:
36: };
```
- 第 7 至 16 行：`cumulation.refCnt() > 1` 成立，和 `MERGE_CUMULATOR` 的情况一致，创建一个新的 ByteBuf 对象。这样，再下一次 `#cumulate(...)` 时，就会走【第 22 至 26 行】的情况。
- 获得
```
composite
```
对象
- 第 19 至 21 行：如果原来**就是** CompositeByteBuf 类型，直接使用。
- 第 22 至 26 行：如果原来**不是** CompositeByteBuf 类型，创建 CompositeByteBuf 对象，并添加 `cumulation` 到其中。
- 第 28 行：添加 `in` 到 `composite` 中，避免内存拷贝。
## 4.3 对比
关于 `MERGE_CUMULATOR` 和 `COMPOSITE_CUMULATOR` 的对比，已经写在 `COMPOSITE_CUMULATOR` 的**头上**的注释。
默认情况下，ByteToMessageDecoder 使用 `MERGE_CUMULATOR` 作为累加器。
# 5. ByteToMessageDecoder
`io.netty.handler.codec.ByteToMessageDecoder` ，继承 ChannelInboundHandlerAdapter 类，**抽象基类**，负责将 Byte 解码成 Message 。
> 老艿艿：ByteToMessageDecoder 的细节比较多，建议胖友理解如下小节即可：
>
> - [5.1 构造方法](https://svip.iocoder.cn/Netty/Codec-1-1-ByteToMessageDecoder-core-impl/#)
> - [5.2 channelRead](https://svip.iocoder.cn/Netty/Codec-1-1-ByteToMessageDecoder-core-impl/#)
> - [5.3 callDecode](https://svip.iocoder.cn/Netty/Codec-1-1-ByteToMessageDecoder-core-impl/#)
> - [5.4 channelReadComplete](https://svip.iocoder.cn/Netty/Codec-1-1-ByteToMessageDecoder-core-impl/#)
## 5.1 构造方法
```
private static final byte STATE_INIT = 0;
private static final byte STATE_CALLING_CHILD_DECODE = 1;
private static final byte STATE_HANDLER_REMOVED_PENDING = 2;
/**
* 已累积的 ByteBuf 对象
*/
ByteBuf cumulation;
/**
* 累计器
*/
private Cumulator cumulator = MERGE_CUMULATOR;
/**
* 是否每次只解码一条消息，默认为 false 。
*
* 部分解码器为 true ，例如：Socks4ClientDecoder
*
* @see #callDecode(ChannelHandlerContext, ByteBuf, List)
*/
private boolean singleDecode;
/**
* 是否解码到消息。
*
* WasNull ，说明就是没解码到消息
*
* @see #channelReadComplete(ChannelHandlerContext)
*/
private boolean decodeWasNull;
/**
* 是否首次读取，即 {@link #cumulation} 为空
*/
private boolean first;
/**
* A bitmask where the bits are defined as
*
* - {@link #STATE_INIT}
* - {@link #STATE_CALLING_CHILD_DECODE}
* - {@link #STATE_HANDLER_REMOVED_PENDING}
*
*
* 解码状态
*
* 0 - 初始化
* 1 - 调用 {@link #decode(ChannelHandlerContext, ByteBuf, List)} 方法中，正在进行解码
* 2 - 准备移除
*/
private byte decodeState = STATE_INIT;
/**
* 读取释放阀值
*/
private int discardAfterReads = 16;
/**
* 已读取次数。
*
* 再读取 {@link #discardAfterReads} 次数据后，如果无法全部解码完，则进行释放，避免 OOM
*/
private int numReads;
protected ByteToMessageDecoder() {
// 校验，不可共享
ensureNotSharable();
}
```
属性比较简单，胖友自己看注释。
## 5.2 channelRead
`#channelRead(ChannelHandlerContext ctx, Object msg)` 方法，读取到新的数据，进行解码。代码如下：
```
1: @Override
2: public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3: if (msg instanceof ByteBuf) {
4: // 创建 CodecOutputList 对象
5: CodecOutputList out = CodecOutputList.newInstance();
6: try {
7: ByteBuf data = (ByteBuf) msg;
8: // 判断是否首次
9: first = cumulation == null;
10: // 若首次，直接使用读取的 data
11: if (first) {
12: cumulation = data;
13: // 若非首次，将读取的 data ，累积到 cumulation 中
14: } else {
15: cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
16: }
17: // 执行解码
18: callDecode(ctx, cumulation, out);
19: } catch (DecoderException e) {
20: throw e; // 抛出异常
21: } catch (Exception e) {
22: throw new DecoderException(e); // 封装成 DecoderException 异常，抛出
23: } finally {
24: // cumulation 中所有数据被读取完，直接释放全部
25: if (cumulation != null && !cumulation.isReadable()) {
26: numReads = 0; // 重置 numReads 次数
27: cumulation.release(); // 释放 cumulation
28: cumulation = null; // 置空 cumulation
29: // 读取次数到达 discardAfterReads 上限，释放部分的已读
30: } else if (++ numReads >= discardAfterReads) {
31: // We did enough reads already try to discard some bytes so we not risk to see a OOME.
32: // See https://github.com/netty/netty/issues/4275
33: numReads = 0; // 重置 numReads 次数
34: discardSomeReadBytes(); // 释放部分的已读
35: }
36:
37: // 解码消息的数量
38: int size = out.size();
39: // 是否解码到消息
40: decodeWasNull = !out.insertSinceRecycled();
41:
42: // 触发 Channel Read 事件。可能是多条消息
43: fireChannelRead(ctx, out, size);
44:
45: // 回收 CodecOutputList 对象
46: out.recycle();
47: }
48: } else {
49: // 触发 Channel Read 事件到下一个节点
50: ctx.fireChannelRead(msg);
51: }
52: }
```
- 第 48 至 51 行：消息的类型**不是** ByteBuf 类，直接触发 Channel Read 事件到下一个节点。也就说，不进行解码。
- 第 3 行：消息的类型**是** ByteBuf 类，开始解码。
- 第 5 行：创建 CodecOutputList 对象。CodecOutputList 的简化代码如下：
```
/**
* Special {@link AbstractList} implementation which is used within our codec base classes.
*/
final class CodecOutputList extends AbstractList