[编织消息框架][设计协议]解决粘包半包(下)

接下来介绍netty如何切割分包

学习目的，了解处理业务，方便以后脱离依赖

读者如果不感兴趣或看不懂可以先忽略，难度比较大

LengthFieldBasedFrameDecoder.class

    public LengthFieldBasedFrameDecoder(
            ByteOrder byteOrder,     //大小端模式 默认大端 ByteOrder BIG_ENDIAN
            int maxFrameLength,      //包Frame netty叫帧概念 最大上限
            int lengthFieldOffset,     //包长度信息偏移多少bytes
            int lengthFieldLength,    //包长度单位为bytes
            int lengthAdjustment,     //包附加信息占多少位，如包尾部checksum 也可以不用设置
            int initialBytesToStrip,//忽悠包头信息，返给上层时会去掉这部份bytes
            boolean failFast)        //解包出错，控制抛异常，默认为true 无需关心这选项

如示例

int maxFrameLength = Short.MAX_VALUE;
int lengthFieldOffset =1;
int lengthFieldLength =2;
new LengthFieldBasedFrameDecoder(ByteOrder.BIG_ENDIAN,maxFrameLength,lengthFieldOffset,lengthFieldLength,0,0,true);

如图：

数据包由低到高是从左到右

红色部份lengthFieldOffset 1byte=8bits 

蓝色部份lengthFieldLength 2byte=16bits = short

粉色部份messageLength 2byte 读取 蓝色部份由于大端模式 高位 0000 0010 等于2 

 图2

消息长度为1byte 

解读netty源码

分四部份

1.netty解码介绍

2.边界判断

3.计算逻辑

4.切割包

第一部份

先看下LengthFieldBasedFrameDecoder继承类之间的关系

ByteToMessageDecoder处理比较复杂，先不考虑

其实解码只要工作方法是

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

//公开调用方法 out 是解码后保存返回，为什么是个数组？原因有可能出现粘包情况多次解码，合并结果一次返回上层业务
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    //调用实现解码方法
    Object decoded = decode(ctx, in);
    if (decoded != null) {
        out.add(decoded);
    }
}

提示：如果自己实现解码 继承ByteToMessageDecoder类，逻辑写在decode(ChannelHandlerContext ctx, ByteBuf in)方法即可

第三部份

    //计算包长度偏移坐标 已读坐标+lengthFieldOffset参数
    int actualLengthFieldOffset = in.readerIndex() + lengthFieldOffset;
    //算出包长度
    long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);
    
    //包实际长度  加上开始忽悠的 lengthFieldEndOffset参数 加上 lengthAdjustment 参数
    frameLength += lengthAdjustment + lengthFieldEndOffset;
    
    //跳过忽悠头部 initialBytesToStrip参数
    in.skipBytes(initialBytesToStrip);
        
    //length 单位是byte 如2 是占一个short长度 4占int长度 8占long长度
    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;
    }

第四部份

//切割实现，直接调用ByteBuf 方法切割
        /**
     * Extract the sub-region of the specified buffer.
     * <p>
     * If you are sure that the frame and its content are not accessed after
     * the current {@link #decode(ChannelHandlerContext, ByteBuf)}
     * call returns, you can even avoid memory copy by returning the sliced
     * sub-region (i.e. <tt>return buffer.slice(index, length)</tt>).
     **/
    protected ByteBuf extractFrame(ChannelHandlerContext ctx, ByteBuf buffer, int index, int length) {
        return buffer.retainedSlice(index, length);
    }
    
    AbstractUnpooledSlicedByteBuf.class
    //深追ByteBuf切割实现，发现返回的ByteBuf只是保留引用，只是简单记录了下offsetStart adjustment writerIndex
      AbstractUnpooledSlicedByteBuf(ByteBuf buffer, int index, int length) {
        super(length);
        checkSliceOutOfBounds(index, length, buffer);

        if (buffer instanceof AbstractUnpooledSlicedByteBuf) {
            this.buffer = ((AbstractUnpooledSlicedByteBuf) buffer).buffer;
            adjustment = ((AbstractUnpooledSlicedByteBuf) buffer).adjustment + index;
        } else if (buffer instanceof DuplicatedByteBuf) {
            this.buffer = buffer.unwrap();
            adjustment = index;
        } else {
            this.buffer = buffer;
            adjustment = index;
        }

        initLength(length);
        writerIndex(length);
    }

第二部份

现在回过头来看边界判断，主要是看处理出错逻辑部份

long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder);
        //解码出错
        if (frameLength < 0) {
            //无法计算包内容时netty策略选择跳过 包头信息跟包长度大小
            in.skipBytes(lengthFieldEndOffset);
            throw new CorruptedFrameException(
                    "negative pre-adjustment length field: " + frameLength);
        }

         frameLength += lengthAdjustment + lengthFieldEndOffset;
        //奇怪的逻辑，上面已加上 lengthFieldEndOffset 除非 lengthAdjustment 参数或 lengthFieldEndOffset参数设置为负值
        //一般没什么复杂情况最好不要设置负值
        if (frameLength < lengthFieldEndOffset) {
            in.skipBytes(lengthFieldEndOffset);
            throw new CorruptedFrameException(
                    "Adjusted frame length (" + frameLength + ") is less " +
                    "than lengthFieldEndOffset: " + lengthFieldEndOffset);
        }
        
        // never overflows because it's less than maxFrameLength
        int frameLengthInt = (int) frameLength;
        //缓冲剩余可读bytes少于包长度，说明出现半包情况，忽略处理
        if (in.readableBytes() < frameLengthInt) {
            return null;
        }
        //如果包长度少于 initialBytesToStrip 参数会抛异常并且跳过包内容
        //没什么特殊情况最好不要设置initialBytesToStrip参数
        if (initialBytesToStrip > frameLengthInt) {
            in.skipBytes(frameLengthInt);
            throw new CorruptedFrameException(
                    "Adjusted frame length (" + frameLength + ") is less " +
                    "than initialBytesToStrip: " + initialBytesToStrip);
        }

        //包长度大于 maxFrameLength
        //看下netty处理策略
        if (frameLength > maxFrameLength) {
            //算出超出范围
            long discard = frameLength - in.readableBytes();
            //记录包长度
            tooLongFrameLength = frameLength;

            //有两种情况处理
            
            //存在粘包情况，直接 in.skipBytes((int) frameLength);跳过当前包 什么也不用记录
            if (discard < 0) {
                // buffer contains more bytes then the frameLength so we can discard all now
                in.skipBytes((int) frameLength);
            } else {
                //当 discard>=0 
                //说明没存在粘包情况，有可能出现半包情况，in.skipBytes(in.readableBytes()); 清空所有接收数据
                //设置discardingTooLongFrame，bytesToDiscard记录超出范围，下次解码时用上
                // Enter the discard mode and discard everything received so far.
                discardingTooLongFrame = true;
                bytesToDiscard = discard;
                in.skipBytes(in.readableBytes());
            }
            failIfNecessary(true);
            return null;
        }
        
        //当上次包出现超过最大包限制，要忽悠上个包半包情况
        if (discardingTooLongFrame) {
            long bytesToDiscard = this.bytesToDiscard;
            //剩余半包长度
            int localBytesToDiscard = (int) Math.min(bytesToDiscard, in.readableBytes());
            //忽略剩余半包长度
            in.skipBytes(localBytesToDiscard);
            //记录忽略已忽略多少bytes
            bytesToDiscard -= localBytesToDiscard;
            this.bytesToDiscard = bytesToDiscard;
            //如果this.bytesToDiscard==0 说明已全部忽略上一个包 会重置 discardingTooLongFrame
            failIfNecessary(false);
        }
        
    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;
            if (!failFast ||
                failFast && firstDetectionOfTooLongFrame) {
                fail(tooLongFrameLength);
            }
        } else {
            // Keep discarding and notify handlers if necessary.
            if (failFast && firstDetectionOfTooLongFrame) {
                fail(tooLongFrameLength);
            }
        }
    }

当大于maxFrameLength 处理核心忽略当前包，如果存在粘包直接忽略，如果存在半包记录剩余包大小bytesToDiscard，下次循环解码时不停跳过bytesToDiscard

关于分包处理

为何需要分包？比如要传送10G大小的文件，如果把数据全读进内存再发送出去，明显是不合理的，这时要把10G大小的数据切割分批读，分批发送出去

为什么不做在应用协议里？因业太多业务上的逻辑，底层是无法控制的，再说是特定的功能，没必要添加支持，看作成业务上一个小功能实现就行了