【Netty】ByteToMessageDecoder源码解析

本节源码分析需求就是要解析一个自定义TCP协议的数据报文进行解码，关于编码解码熟悉网络编程的同学都明白，不清楚的可以稍微查阅一下资料有助于学习为什么要解码以及如何解码。本节不会对具体报文的解析做具体讲解，只对Netty提供的解码器基类ByteToMessageDecoder做一下源码分解，以及如何使用ByteToMessageDecoder开发属于自己的Decoder，接下来我们看看ByteToMessageDecoder的定义。


/*继承ChannelInboundHandlerAdapter
 *	字节到消息的编码器，是Inbound操作，这类解码器处理器都不是共享的，因为需要存没有    
 *  解码完的数据，还有各种状态，是独立的，不能进行共享。
 *	所以在pipeline上需要每次创建新的对象，不是能够进行对象复用。
 */
public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
}

2.类的实现

解码器的ByteToMessageDecoder ，该类继承了ChannelInboundHandlerAdapter ，ChannelInboundHandlerAdapter继承ChannelHandlerAdapter，

ChannelInboundHandlerAdapter实现ChannelInboundHandler接口，也就是说ChannelInboundHandler定义了解码器需要处理的工作（方法）

ChannelInboundHandlerAdapter是一个适配器模式，负责Decoder的扩展。它的实现有很多，简单列举一下：

```
HeartBeatHandler
```
```
MessageToMessageDecoder
```

SimpleChannelInboundHandler（实现抽象了channelRead方法，提供抽象方法channelRead0）

ByteToMessageDecoder
。。。。。。

以上都是比较常用的Decoder或Handler，基于这些基类还定义了很多handler，有兴趣的同学可以跟代码查阅。

3.Decoder工作流程

每当数据到达Server端时，SocketServer通过Reactor模型分配具体的worker线程进行处理数据，处理数据就需要我们的事先定义好的Decoder以及handler，假如我们定义了以下两个对象：

MyDecoder extends ByteToMessageDecoder{} 作为解码器
MyHandler extends SimpleChannelInboundHandler{} 作为解码后的业务处理器

worker线程——〉MyDecoder#channelRead实际就是调用ByteToMessageDecoder#channelRead——〉Cumulator累加器处理——〉

解码器decode处理（MyDecoder需要实现decode方法）——〉Myhandler#channelRead0处理具体的数据（msg），msg是通过MyDecoder#decode方法解码后的数据对象。

4.源码解析

4.1 累加器Cumulator（数据累加缓冲区）

累加器Cumulator的作用是数据累加缓冲区，解决tcp数据包中出现半包和粘包问题。

半包：接收到的byte字节不足一个完整的数据包，

半包处理办法：不足一个完整的数据包先放入累加器不做解码，等待续传的数据包；

粘包：接收到的byte字节数据包中包括其他数据包的数据（靠数据包协议中定义的帧头帧尾标识来识别，多于1个以上的帧头或帧尾数据包为粘包数据），

粘包处理办法：按照数据包帧结构定义去解析，需要结合累加器，解析完一个数据包交给handler去处理，剩下的不足一个数据包长度的字节保存在累加器等待续传的数据包收到之后继续解码。

ByteToMessageDecoder内部定义了Cumulator接口


    /**
     * 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.
         */
        ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
    }

其中在类最开始的时候构建了两个对象，分别是MERGE_CUMULATOR，COMPOSITE_CUMULATOR，代码如下


    /**
     * Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
     */
    public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
        /**
	     *	主要是做一般缓冲区的合并，直接将新的缓冲区拷贝到累加缓冲区中。
	     *	cumulation		累加的缓冲区
	     *  in				新读取的数据
         */
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
            try {
                final ByteBuf buffer;
                //1.如果累加器ByteBuf 剩余可写的capacity不满足当前需要写入的ByteBuf(in)长度，则进行扩容累加器ByteBuf容量，执行expandCumulation方法
                if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
                    || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
                   
                    buffer = expandCumulation(alloc, cumulation, in.readableBytes());
                } else {
                    buffer = cumulation;
                }
                //2.写入累加器并返回更新后的cumulation
                buffer.writeBytes(in);
                return buffer;
            } 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)
                //3.由于是对in的拷贝，in不需要在进行往下的传递，所以应该直接释放,需要release
                in.release();
            }
        }
    };
 
 
    //通过对CompositeByteBuf的累加器的实现，CompositeByteBuf内部使用ComponentList
    //实现对ByteBuf进行追加
    //ComponentList是ArrayList的实现，所以每次Add操作都是一次内存拷贝。
    public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
            ByteBuf buffer;
            try {
                if (cumulation.refCnt() > 1) {
                    buffer = expandCumulation(alloc, cumulation, in.readableBytes());
                    buffer.writeBytes(in);
                } else {
                    CompositeByteBuf composite;
                    if (cumulation instanceof CompositeByteBuf) {
                        composite = (CompositeByteBuf) cumulation;
                    } else {
                        composite = alloc.compositeBuffer(Integer.MAX_VALUE);
                        composite.addComponent(true, cumulation);
                    }
                    composite.addComponent(true, in);
                    in = null;
                    buffer = composite;
                }
                return buffer;
            } finally {
                if (in != null) {
                    //因为是对ByteBuf in的拷贝，所以需要释放
                    in.release();
                }
            }
        }
    };

4.2 状态码说明


    //状态码
    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;	//表示handler正在删除
 
    ByteBuf cumulation;								//核心重点		累加的缓冲区(累加器)
    private Cumulator cumulator = MERGE_CUMULATOR;	//累加器的默认状(默认是合并的累加器)
    private boolean singleDecode;					//是否解码一次
    private boolean decodeWasNull;					//out对象有被添加或者设置，表是有读过了
    private boolean first;							//是否是第一次累加缓冲区，true表示第一次累加缓存区，false表示不是第一次累加缓冲区
   
   /**
     * A bitmask where the bits are defined as
     * 
     *     
{@link #STATE_INIT}

*
{@link #STATE_CALLING_CHILD_DECODE}