[11] 解码原理解析

2022-06-20 08:32:23 阅读：118 来源： 互联网

摘自《Netty 即时聊天实战与底层原理》

1. 粘包与拆包

1.1 为什么要粘包

首先你得了解一下 TCP/IP 协议，在用户数据量非常小的情况下，比如 1 字节，该 TCP 数据包的有效载荷非常低，传递 100 个字节的数据，需要 100 次 TCP 传送、100 次 ACK，在应用及时性要求不高的情况下，将这 100 个有效数据拼接成一个数据包，就会缩短到一个 TCP 数据包，以及一个 ACK，提高了有效载荷，也节省了带宽。

在非极端情况下，有可能两个数据包拼接成了一个数据包，也有可能一个半的数据包拼接成一个数据包，也有可能两个半的数据包拼接成一个数据包。

1.2 为什么要拆包

拆包和粘包是相对的，一端粘了包，另外一端就需要将粘过的包拆开。举个例子，发送端将 3 个数据包粘成 2 个数据包发送到接收端，接收端就需要根据应用协议将两个数据包重新组装成 3 个数据包。

还有一种情况就是用户数据包超过了 MSS(最大报文长度)，那么这个数据包在发送的时候必须拆分成几个数据包，接收端收到这些数据包之后需要将这些数据包粘合之后再拆开。

1.3 拆包的原理

在没有 Netty 的情况下，用户如果自己需要拆包，基本原理就是不断地从 TCP 缓冲区读取数据，每次读取完都需要判断是否为一个完整地数据包。

如果当前读取的数据不足以拼接成一个完整的业务数据包，那就保留该数据，继续从 TCP 缓冲区中读取，直到得到一个完整的数据包；
如果当前读取得到的数据加上已经读取的数据足够拼接成一个数据包，那就将已经读取的数据拼接上本次读取的数据，构成一个完成的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下次读到的数据尝试拼接。

1.4 Netty 中拆包的基类

Netty 中的拆包原理也如上所述，内部会有一个累加器，每次读到的数据都会不断累加，然后尝试对累加的数据机进行拆包，拆成一个完整的业务数据包，这个基类叫做 ByteToMessageDecoder，下面我们详细分析这个类。

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter { /*...*/ }

ByteToMessageDecoder 中定义了两个累加器：

public static final Cumulator MERGE_CUMULATOR = /*...*/;
public static final Cumulator COMPOSITE_CUMULATOR = /*...*/;

在默认情况下，会使用 MERGE_CUMULATOR。

private Cumulator cumulator = MERGE_CUMULATOR;

MERGE_CUMULATOR 的原理是每次都将读取到的数据通过内存拷贝的方式，拼接到一个大的字节容器中，这个字节容器在 ByteToMessageDecoder 中被叫做 cumulation。

ByteBuf cumulation;

下面我们看一下 MERGE_CUMULATOR 是如何将读取的数据累加到字节容器里的。

public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
    @Override
    public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        final ByteBuf buffer;
        if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
                || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
            // Expand cumulation (by replace it) when either there is not more room
            // in the buffer or if the refCnt is greater then 1 which may happen when
            // the user use slice().retain() or duplicate().retain() or if its read-only.
            buffer = expandCumulation(alloc, cumulation, in.readableBytes());
        } else {
            buffer = cumulation;
        }
        buffer.writeBytes(in);
        in.release();
        return buffer;
    }
};

Netty 中 ByteBuf 的抽象，使得累加非常简单，通过一个简单的 API 调用 buffer.writeBytes(in) 便可以将新数据累加到字节容器中，为了防止字节容器不够，在累加之前还进行了扩容处理。

static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
    ByteBuf oldCumulation = cumulation;
    cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
    cumulation.writeBytes(oldCumulation);
    oldCumulation.release();
    return cumulation;
}

扩容也是一个内存拷贝操作，新增的大小即新读取数据的大小。

2. 拆包抽象

累加器原理清楚之后，我们回到主流程，目光集中在 channelRead() 方法上，channelRead() 方法是每次从 TCP 缓冲区读到数据都会调用的方法，触发点在 AbstractNioByteChannel 的 read 方法中，里面有一个 while 循环不断读取数据，读取到一次就触发一次channelRead() 方法。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            // 1. 累加数据
            ByteBuf data = (ByteBuf) msg;
            first = cumulation == null;
            if (first) {
                cumulation = data;
            } else {
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            // 2. 将累加的数据传递给业务进行拆包
            callDecode(ctx, cumulation, out);
        } finally {
            // 3. 清理字节容器
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
            } else if (++ numReads >= discardAfterReads) {
                // We did enough reads already try to discard
                // some bytes so we not risk to see a OOME.
                numReads = 0;
                discardSomeReadBytes();
            }
            // 4. 将业务数据包传递给业务解码器处理
            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            fireChannelRead(ctx, out, size);
            out.recycle();
        }
    } else {
        ctx.fireChannelRead(msg);
    }
}

2.1 累加数据

如果当前累加器中没有数据，就直接跳过内存拷贝，将字节容器的指针指向新读取的数据，否则，调用累加器累加数据至字节容器。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    CodecOutputList out = CodecOutputList.newInstance();

    // 1. 累加数据
    ByteBuf data = (ByteBuf) msg;
    first = cumulation == null;
    if (first) {
        cumulation = data;
    } else {
        cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
    }

    // 2. 将累加的数据传递给业务进行拆包
    // 3. 清理字节容器
    // 4. 将业务数据包传递给业务解码器处理
}

2.2 将累加的数据传递给业务进行拆包

到这一步，字节容器里的数据已经是目前未拆包部分的所有数据了。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    CodecOutputList out = CodecOutputList.newInstance();

    // 1. 累加数据

    // 2. 将累加的数据传递给业务进行拆包
    callDecode(ctx, cumulation, out);

    // 3. 清理字节容器
    // 4. 将业务数据包传递给业务解码器处理

}

callDecode(...) 尝试将字节容器的数据拆分成业务数据包，塞到业务数据容器 out 中。

protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {

    while (in.isReadable()) {
        int outSize = out.size();

        if (outSize > 0) {
            fireChannelRead(ctx, out, outSize);
            out.clear();

            // Check if this handler was removed before continuing with decoding.
            // If it was removed, it is not safe to continue to operate on the buffer.
            if (ctx.isRemoved()) {
                break;
            }
            outSize = 0;
        }

        // 记录一下字节容器中有多少字节待拆
        int oldInputLength = in.readableBytes();
        decode(ctx, in, out);

        // Check if this handler was removed before continuing the loop.
        // If it was removed, it is not safe to continue to operate on the buffer.
        if (ctx.isRemoved()) {
            break;
        }

        if (outSize == out.size()) {
            if (oldInputLength == in.readableBytes()) { // 拆包器未读取任何数据
                break;
            } else {                                    // 拆包器已读取部分数据，还需要继续
                continue;
            }
        }

        if (oldInputLength == in.readableBytes()) {
            throw new DecoderException(StringUtil.simpleClassName(getClass()) +
                        ".decode() did not read anything but decoded a message.");
        }

        if (isSingleDecode()) {
            break;
        }
    }

}

笔者将原始代码做了一些精简，在解码之前，先记录一下字节容器中有多少字节待拆，然后调用抽象函数 decode 进行拆包。

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

Netty 中对各种用户协议的支持就体现在这个抽象函数中，传进去的是当前读取到的未被消费的所有数据以及业务协议包容器，所有的拆包器最终都实现了该抽象方法。

业务拆包完成之后，如果发现并没有拆到一个完整的数据包，这个时候又分成两种情况：

拆包器什么数据也没读取，可能数据还不够业务拆包器处理，通过 break 关键字跳出循环；
拆包器已读取部分数据，说明解码器仍然在工作，继续解码。

业务拆包完成之后，如果发现已经解到了数据包，但是发现并没有读取任何数据，这时候就会抛出一个 RuntimeException$DecoderException，告诉你什么数据都没读取到，却解析出一个业务数据包，这是有问题的。

2.3 清理字节容器

业务拆包完成之后，只是从字节容器中取走了数据，但是这部分空间对于字节容器来说依然被保留着，而字节容器每次累加字节数据的时候都是将字节数据追加到尾部，如果不对字节容器做清理，那么时间一长就会出现 OOM 问题。

在正常情况下，其实每次读取完数据，Netty 就会在下面这个方法中将字节容器清理，只不过，在发送端发送数据过快时，channelReadComplete 可能会很久才被调用一次。

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    numReads = 0;
    discardSomeReadBytes();
    if (decodeWasNull) {
        decodeWasNull = false;
        if (!ctx.channel().config().isAutoRead()) {
            ctx.read();
        }
    }
    ctx.fireChannelReadComplete();
}

这里顺带插一句，如果一次数据读取完毕之后（可能接收端一边收，发送端一边发，这里的读取完毕指接收端在某个时间不再接收数据），发现仍然没有拆到一个完整的用户数据包，即使该 Channel 的设置为非自动读取，也会触发一次读取操作 ctx.read()，该操作会重新向 Selector 注册 op_read 事件，以便下一次能在读取到数据后拼接成一个完整的数据包。

所以为了防止发送端发送数据过快，Netty 会在每次读取到一次数据，业务拆包之后，对字节容器做清理。清理字节容器部分的代码如下。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    // 1. 累加数据
    // 2. 将累加的数据传递给业务进行拆包

    // 3. 清理字节容器
    if (cumulation != null && !cumulation.isReadable()) {
        numReads = 0;
        cumulation.release();
        cumulation = null;
    } else if (++ numReads >= discardAfterReads) {
        // We did enough reads already try to discard
        // some bytes so we not risk to see a OOME.
        numReads = 0;
        discardSomeReadBytes();
    }

    // 4. 将业务数据包传递给业务解码器处理

}

2.4 将业务数据包传递给业务解码器处理

以上三个步骤完成之后，就可以将拆成的数据包丢到业务解码器中处理了，代码如下。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    // 1. 累加数据
    // 2. 将累加的数据传递给业务进行拆包
    // 3. 清理字节容器

    // 4. 将业务数据包传递给业务解码器处理
    int size = out.size();
    decodeWasNull = !out.insertSinceRecycled();
    fireChannelRead(ctx, out, size);
    out.recycle();

}

期间用一个成员变量 decodeWasNull 来标识本次读取数据是否拆到一个业务数据包，然后调用 fireChannelRead(...) 将拆到的业务数据包都传递到后续的 Handler。

static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
    for (int i = 0; i < numElements; i ++) {
        ctx.fireChannelRead(msgs.getUnsafe(i));
    }
}

这样，就可以把一个个完整的业务数据包传递到后续的业务解码器进行解码，随后处理业务逻辑。

3. 行拆包器

下面以一个具体的例子来看看 Netty 自带的拆包器是如何拆包的。

这个类叫做 LineBasedFrameDecoder，是基于行分隔符的拆包器，它可以同时处理 \n 和 \r\n 两种类型的行分隔符，核心方法都在继承的 decode 方法中。

@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);
    }
}

Netty 中自带的拆包器都是基于如上这种模板的，其实可以加一层，把这层模板抽取出来，不知道为什么 Netty 没有这么做。我们接着跟进去，代码比较长，我们还是分模块来剖析。

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置
    final int eol = findEndOfLine(buffer);

    // 2. 非 discarding 模式的处理
    if (!discarding) {
        if (eol >= 0) {
            final ByteBuf frame;
            final int length = eol - buffer.readerIndex();
            final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;

            if (length > maxLength) {
                buffer.readerIndex(eol + delimLength);
                fail(ctx, length);
                return null;
            }

            if (stripDelimiter) {
                frame = buffer.readRetainedSlice(length);
                buffer.skipBytes(delimLength);
            } else {
                frame = buffer.readRetainedSlice(length + delimLength);
            }

            return frame;
        } else {
            final int length = buffer.readableBytes();
            if (length > maxLength) {
                discardedBytes = length;
                buffer.readerIndex(buffer.writerIndex());
                discarding = true;
                offset = 0;
                if (failFast) {
                    fail(ctx, "over " + discardedBytes);
                }
            }
            return null;
        }
    } else {
        if (eol >= 0) {
            final int length = discardedBytes + eol - buffer.readerIndex();
            final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
            buffer.readerIndex(eol + delimLength);
            discardedBytes = 0;
            discarding = false;
            if (!failFast) {
                fail(ctx, length);
            }
        } else {
            discardedBytes += buffer.readableBytes();
            buffer.readerIndex(buffer.writerIndex());
        }
        return null;
    }
}

3.1 找到换行符的位置

LineBasedFrameDecoder

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置
    final int eol = findEndOfLine(buffer);
    // ...
}

/**
 * Returns the index in the buffer of the end of line found.
 * Returns -1 if no end of line was found in the buffer.
 */
private int findEndOfLine(final ByteBuf buffer) {
    int totalLength = buffer.readableBytes();
    int i = buffer.forEachByte(buffer.readerIndex() + offset, totalLength - offset, ByteProcessor.FIND_LF);
    if (i >= 0) {
        offset = 0;
        if (i > 0 && buffer.getByte(i - 1) == '\r') {
            i--;
        }
    } else {
        offset = totalLength;
    }
    return i;
}

for 循环遍历，找到第一个 \n 的位置，如果 \n 前面的字符为 \r，那么返回 \r 的位置。

3.2 !discarding 模式的处理

接下来，Netty 会判断当前拆包是否处于 discarding 模式，用一个成员变量来标识。

/** True if we're discarding input because we're already over maxLength.  */
private boolean discarding;

第一次拆包不在 discarding 模式中（后面会说何为 !discarding 模式），于是进入以下环节。

a. 找到行分隔符的处理

```java
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置

    // 2. !discarding 模式的处理
    if (!discarding) {
        if (eol >= 0) {                            // --- 找到行分隔符的处理
            // (1) 计算分隔符和包长度
            final ByteBuf frame;
            final int length = eol - buffer.readerIndex();
            final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;

            // (2) 丢失异常数据
            if (length > maxLength) {
                buffer.readerIndex(eol + delimLength);
                fail(ctx, length);
                return null;
            }
            // (3) 取包的时候是否包括分隔符
            if (stripDelimiter) {
                frame = buffer.readRetainedSlice(length);
                buffer.skipBytes(delimLength);
            } else {
                frame = buffer.readRetainedSlice(length + delimLength);
            }

            return frame;
        } else {                                   // --- 未找到行分隔符的处理
            // ...
        }
    }

    // ...

}

新建一个帧，计算当前包的长度和分隔符的长度（因为有两种分隔符）；
判断一下需要拆包的长度是否大于该拆包器允许的最大长度（maxLength），这个参数在构造函数中被传递进去，如果超出允许的最大长度，则将这段数据抛弃，返回 null；
将一个完整的数据包取出，如果构造本拆包器的时候，指定 stripDelimiter 为 false，则解析出来的数据包包含分隔符。默认是不包含分隔符的。

b. 未找到行分隔符的处理

没有找到对应的行分隔符，说明字节容器没有足够的数据拼接成一个完整的业务数据包，则进入如下流程处理：

 /** Whether or not to throw an exception as soon as we exceed maxLength. */
private final boolean failFast;

private int discardedBytes;

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置
    final int eol = findEndOfLine(buffer);

    // 2. !discarding 模式的处理
    if (!discarding) {
        if (eol >= 0) {                            // --- 找到行分隔符的处理
            // ...
        } else {                                   // --- 未找到行分隔符的处理
            final int length = buffer.readableBytes();
            if (length > maxLength) {
                discardedBytes = length;
                buffer.readerIndex(buffer.writerIndex());
                discarding = true;
                offset = 0;
                if (failFast) {
                    fail(ctx, "over " + discardedBytes);
                }
            }
            return null;
        }
    }
    // ...

首先取得当前字节容器的可读字节数，接着判断一下是否已经超过允许的最大长度，如果没有超过，则直接返回 null，字节容器中的数据没有任何改变，否则就需要进入 discarding 模式。

使用一个成员变量 discardedBytes 来表示已经丢弃了多少数据，然后将字节容器的读指针移到写指针，意味着丢弃这一部分数据。设置成员变量 discarding 为 true，表示当前处理 discarding 模式。如果设置了 failFast，那么直接抛出异常；默认情况下 failFast 为 false，即安静地丢弃数据。

3.3 discarding 模式的处理

如果解包的时候处在 discarding 模式，也会有两种情况发生。

a. 找到行分隔符的处理

在 discarding 模式下，如果找到分隔符，那么可以将分隔符之前的数据都丢弃。

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置
    final int eol = findEndOfLine(buffer);

    // 2. !discarding 模式的处理
    if (!discarding) {
        // ...
    } else {
        // 3. discarding 模式的处理
        if (eol >= 0) {                            // --- 找到行分隔符的处理
            final int length = discardedBytes + eol - buffer.readerIndex();
            final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
            buffer.readerIndex(eol + delimLength);
            discardedBytes = 0;
            discarding = false;
            if (!failFast) {
                fail(ctx, length);
            }
        } else {                                   // --- 未找到行分隔符的处理
            // ...
        }
        return null;
    }
}

计算出分割符的长度之后，直接就把分隔符之前的数据全部丢弃，当然丢弃的字符也包括分隔符，经过这么一次丢弃，后面就有可能是正常的数据包，下一次解包的时候就会进入正常的解包流程。

b. 未找到行分隔符的处理

这种情况比较简单，因为当前还处在 discarding 模式，没有找到行分隔符意味着当前一个完整的数据包还没丢弃完，当前读取的数据是丢弃的一部分，所以直接丢弃。

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
    // 1. 找到换行符的位置
    final int eol = findEndOfLine(buffer);

    // 2. 非 discarding 模式的处理
    if (!discarding) {
        // ...
    } else {
        if (eol >= 0) {                            // --- 找到行分隔符的处理
            // ...
        } else {                                   // --- 未找到行分隔符的处理
            discardedBytes += buffer.readableBytes();
            buffer.readerIndex(buffer.writerIndex());
        }
        return null;
    }
}

4. 特定分隔符拆包

这个类叫做 DelimiterBasedFrameDecoder，可以传递给它一个分隔符列表，数据包会按照分隔符列表进行拆分，读者可以完全根据行拆包器的思路去分析这个 DelimiterBasedFrameDecoder，这里不再赘述。

5. LengthFieldBasedFrameDecoder

之所以 Netty 的拆包能如此强大，就是因为 Netty 将具体如何拆包抽象出一个 decode 方法，不同的拆包器实现不同的 decode 方法，就能实现不同协议的拆包。

接下来要介绍的就是最常见的通用拆包器 LengthFieldBasedFrameDecoder，下面我们深入介绍一下其用法及原理。

5.1 用法

a. 基于长度的拆包

b. 基于长度的截断拆包

c. 基于偏移长度的拆包

d. 基于可调整长度的拆包

e. 基于偏移可调整长度的截断拆包

f. 基于偏移可调整变异长度的截断拆包

5.2 源码

a. 构造函数

关于 LengthFieldBasedFrameDecoder 的构造函数，我们只需要看一个就够了。


/**
 * Creates a new instance.
 *
 * @param byteOrder
 *        the {@link ByteOrder} of the length field
 * @param maxFrameLength
 *        the maximum length of the frame.  If the length of the frame is
 *        greater than this value, {@link TooLongFrameException} will be
 *        thrown.
 * @param lengthFieldOffset
 *        the offset of the length field
 * @param lengthFieldLength
 *        the length of the length field
 * @param lengthAdjustment
 *        the compensation value to add to the value of the length field
 * @param initialBytesToStrip
 *        the number of first bytes to strip out from the decoded frame
 * @param failFast
 *        If <tt>true</tt>, a {@link TooLongFrameException} is thrown as
 *        soon as the decoder notices the length of the frame will exceed
 *        <tt>maxFrameLength</tt> regardless of whether the entire frame
 *        has been read.  If <tt>false</tt>, a {@link TooLongFrameException}
 *        is thrown after the entire frame that exceeds <tt>maxFrameLength</tt>
 *        has been read.
 */
public LengthFieldBasedFrameDecoder(ByteOrder byteOrder,
        int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,
        int lengthAdjustment, int initialBytesToStrip, boolean failFast) {

    this.byteOrder = byteOrder;
    this.maxFrameLength = maxFrameLength;
    this.lengthFieldOffset = lengthFieldOffset;
    this.lengthFieldLength = lengthFieldLength;
    this.lengthAdjustment = lengthAdjustment;
    lengthFieldEndOffset = lengthFieldOffset + lengthFieldLength;
    this.initialBytesToStrip = initialBytesToStrip;
    this.failFast = failFast;
}

构造函数做的事很简单，只是把传入的参数简单地保存在 field 里，这里的大多数 field 在前面已经阐述过了，剩下的几个补充说明如下：

byteOrder 表示字节流表示的数据用大端还是小端，用于长度域的读取；
lengthFieldEndOffset 表示紧跟长度域字段后面的第一个字节在整个数据包中的偏移量；
failFast，如果为 true，则表示读取到长度域，它的值超过 maxFrameLength，就抛出一个 TooLongFrameException，而为 false 则表示只有当真正读取完长度域的值表示的字节之后，才会抛出 TooLongFrameException。默认情况下设置为 true，建议不要修改，否则可能会造成内存溢出。

b. 拆包入口

通过前面内容的介绍，我们已经知道，具体的拆包协议只需要实现以下代码即可。

decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)

其中，in 表示到目前为止还未拆的数据，拆完之后的包添加到 out 这个 list 中，即可实现包向下传递。

第一层实现比较简单，代码如下：

@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);
    }
}

重载的 protected 方法 decode 做真正的拆包动作，下面详细分析一下这个重量级函数。

c. 获取待拆包的大小

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    // ...

    // 如果当前可读字节还未到达长度域的偏移，说明肯定是读不到长度域的，则直接不读
    if (in.readableBytes() < lengthFieldEndOffset) {
        return null;
    }
    // 拿到长度域的实际字节偏移
    int actualLengthFieldOffset = in.readerIndex() + lengthFieldOffset;
    // 拿到实际的未调整过的包长度
    long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);
    // 如果拿到的长度为负数，则直接跳过长度域并抛出异常
    if (frameLength < 0) {
        failOnNegativeLengthField(in, frameLength, lengthFieldEndOffset);
    }
    // 调整包的长度，后面统一做拆分
    frameLength += lengthAdjustment + lengthFieldEndOffset;

    // ...
}

上面这段内容有一个扩展点 getUnadjustedFrameLength(...)，如果长度域代表的值表达的含义不是正常的 int、short 等基本类型，则可以重写这个函数。

protected long getUnadjustedFrameLength(ByteBuf buf, int offset, int length, ByteOrder order) {
    buf = buf.order(order);
    long frameLength;
    switch (length) {
    case 1:
        frameLength = buf.getUnsignedByte(offset);
        break;
    case 2:
        frameLength = buf.getUnsignedShort(offset);
        break;
    case 3:
        frameLength = buf.getUnsignedMedium(offset);
        break;
    case 4:
        frameLength = buf.getUnsignedInt(offset);
        break;
    case 8:
        frameLength = buf.getLong(offset);
        break;
    default:
        throw new DecoderException("unsupported lengthFieldLength: " + 
                    lengthFieldLength + " (expected: 1, 2, 3, 4, or 8)");
    }
    return frameLength;
}

比如，有的“奇葩”的长度域虽然是 4 字节，比如 0x1234，但是它的含义是十进制的，即长度就是十进制的 1234，那么覆盖这个函数即可实现“奇葩”长度域拆包。

d. 长度校验

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    // ...

    // 整个数据包的长度还没有长度域长，直接抛出异常
    if (frameLength < lengthFieldEndOffset) {
        in.skipBytes(lengthFieldEndOffset);
        throw new CorruptedFrameException(
           "Adjusted frame length (" + frameLength + ") is less " +
              "than lengthFieldEndOffset: " + lengthFieldEndOffset);
    }

    // 数据包长度超出最大包长度，进入 discarding 模式
    if (frameLength > maxFrameLength) {
        long discard = frameLength - in.readableBytes();
        tooLongFrameLength = frameLength;

        if (discard < 0) {
            // 当前可读字节已到达 frameLength，直接跳过 frameLength 字节
            // 丢弃之后，后面有可能就是一个合法的数据包
            // buffer contains more bytes then the frameLength so we can discard all now
            in.skipBytes((int) frameLength);
        } else {
            // 当前可读字节未到达 frameLength，说明后面未读到的字节也需要丢弃
            // 进入 discarding 模式，先把当前积累的字节全部丢弃
            // Enter the discard mode and discard everything received so far.
            discardingTooLongFrame = true;
            // bytesToDiscard 表示还需要丢弃多少字节
            bytesToDiscard = discard;
            in.skipBytes(in.readableBytes());
        }

        failIfNecessary(true);
        return null;
    }

    // ...
}

最后，调用 failIfNecessary() 判断是否需要抛出异常。

private void failIfNecessary(boolean firstDetectionOfTooLongFrame) {
    // 不需要再丢弃后面的未读字节，就开始重置丢弃状态
    if (bytesToDiscard == 0) {
        // Reset to the initial state and tell the handlers that
        // the frame was too large.
        long tooLongFrameLength = this.tooLongFrameLength;
        this.tooLongFrameLength = 0;
        discardingTooLongFrame = false;
        // 如果没有设置快速失败，或者设置了快速失败并且是第一次检测到大包错误
        // 抛出异常，让 Handler 去处理
        if (!failFast || firstDetectionOfTooLongFrame) {
            fail(tooLongFrameLength);
        }
    } else {
        // 如果设置了快速失败，并且是第一次检测到大包错误，则抛出异常，让 Handler 去处理
        // Keep discarding and notify handlers if necessary.
        if (failFast && firstDetectionOfTooLongFrame) {
            fail(tooLongFrameLength);
        }
    }
}

前面我们知道 failFast 默认为 true，而这里 firstDetectionOfTooLongFrame 为 true，所以，第一次检测到大包错误肯定会抛出异常。

下面是抛出异常的代码：

private void fail(long frameLength) {
    if (frameLength > 0) {
        throw new TooLongFrameException(
                        "Adjusted frame length exceeds " + maxFrameLength +
                        ": " + frameLength + " - discarded");
    } else {
        throw new TooLongFrameException(
                        "Adjusted frame length exceeds " + maxFrameLength +
                        " - discarding");
    }
}

e. discarding 模式的处理

LengthFieldBasedFrameDecoder#decode 方法的入口处还有一段代码在我们前面的分析中被省略掉了。放到这一节中的目的是承接上一节，更加容易读懂 discarding 模式的处理。

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {

    if (discardingTooLongFrame) {
        long bytesToDiscard = this.bytesToDiscard;
        int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
        in.skipBytes(localBytesToDiscard);
        bytesToDiscard -= localBytesToDiscard;
        this.bytesToDiscard = bytesToDiscard;
        failIfNecessary(false);
    }

    // ...
}

如上代码所示，如果当前处在 discarding 模式，先计算需要丢弃多少字节，取当前还需可丢弃字节和可读字节的最小值。丢弃之后，进入 failIfNecessary，对照这个函数看，默认情况下是不会继续抛出异常的；而如果设置了 failFast 为 false，那么等丢弃完之后，才会抛出异常，读者可自行分析。

f. 跳过指定字节长度

discarding 模式的处理及长度的校验都通过后，进入跳过指定字节长度这个环节。

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    // ...

    int frameLengthInt = (int) frameLength;
    if (in.readableBytes() < frameLengthInt) {
    	return null;
    }

    if (initialBytesToStrip > frameLengthInt) {
    	in.skipBytes((int) frameLength);
    	throw new CorruptedFrameException(
    		"Adjusted frame length (" + frameLength + ") is less " +
    			"than initialBytesToStrip: " + initialBytesToStrip);
    }
    in.skipBytes(initialBytesToStrip);

    // ...
}

先验证当前是否已经读到足够的字节，如果读到了，则在下一步抽取一个完整的数据包之前，需要根据 initialBytesToStrip 的设置来跳过某些字节（见开篇）。当然，条i过的字节不能大于数据包的长度，否则就抛出 CorruptedFrameException。

g. 抽取 frame

protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    // ...

    // extract frame
    int readerIndex = in.readerIndex();
    int actualFrameLength = frameLengthInt - initialBytesToStrip;
    ByteBuf frame = extractFrame(ctx, in, readerIndex, actualFrameLength);
    in.readerIndex(readerIndex + actualFrameLength);
    return frame;
}

protected ByteBuf extractFrame(ChannelHandlerContext ctx, ByteBuf buffer, int index, int length) {
    return buffer.retainedSlice(index, length);
}

到了最后，抽取数据包其实就很简单了，拿到当前累积数据的读指针，然后拿到待抽取数据包的实际长度进行抽取，抽取之后，移动读指针。

抽取的过程就是简单地调用了一下 ByteBuf 地 retainedSliceapi，该 API 无内存拷贝开销。

从真正抽取数据包来看，传入的参数为 int 类型，所以，可以判断，在自定义协议中，如果你的长度域是 8 字节的，那么前面 4 字节基本是没有用的。

6. 总结

Netty 中的拆包过程其实和你自己去拆包的过程一样，只不过它将拆包过程中逻辑比较独立的部分抽象出来变成几个不同层次的类，方便各种协议的扩产。我们平时写代码过程中，也必须培养这种抽象能力，这样你的编码水平才会不断地调高；
如果你使用了 Netty，并且二进制协议是基于长度的，那么考虑使用 LengthFieldBasedFrameDecoder吧，通过调整各种参数，一定会满足你的需求。
LengthFieldBasedFrameDecoder 地拆包包括合法参数校验、异常包处理，以及最后调用 ByteBuf 的 retainedSlice 来实现无内存拷贝的拆包。

标签：11,字节,buffer,解码,ctx,int,拆包,解析,数据包
来源： https://www.cnblogs.com/liujiaqi1101/p/16392046.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

[11] 解码原理解析

1. 粘包与拆包

1.1 为什么要粘包

1.2 为什么要拆包

1.3 拆包的原理

1.4 Netty 中拆包的基类

2. 拆包抽象

2.1 累加数据

2.2 将累加的数据传递给业务进行拆包

2.3 清理字节容器

2.4 将业务数据包传递给业务解码器处理

3. 行拆包器

3.1 找到换行符的位置

3.2 !discarding 模式的处理

a. 找到行分隔符的处理

b. 未找到行分隔符的处理

3.3 discarding 模式的处理

a. 找到行分隔符的处理

b. 未找到行分隔符的处理

4. 特定分隔符拆包

5. LengthFieldBasedFrameDecoder

5.1 用法

a. 基于长度的拆包

b. 基于长度的截断拆包

c. 基于偏移长度的拆包

d. 基于可调整长度的拆包

e. 基于偏移可调整长度的截断拆包

f. 基于偏移可调整变异长度的截断拆包

5.2 源码

a. 构造函数

b. 拆包入口

c. 获取待拆包的大小

d. 长度校验

e. discarding 模式的处理

f. 跳过指定字节长度

g. 抽取 frame

6. 总结