Netty5_ByteToMessageDecoder_源码解析

欢迎你们关注个人微博 http://weibo.com/hotbain 会将发布的开源项目技术贴经过微博通知你们，但愿你们可以互勉共进！谢谢！也很但愿可以获得你们对我博文的反馈，写出更高质量的文章！！

ByteToMessageDecoder在Netty中起着很大的做用，用来解决半包字节累积问题。粘贴部分重要代码(固然自己方法不是很

public abstract class ByteToMessageDecoder extends ChannelHandlerAdapter {

    ByteBuf cumulation;
    private boolean singleDecode;
    private boolean first;

    protected ByteToMessageDecoder() {
        if (getClass().isAnnotationPresent(Sharable.class)) {//由于每个ByteToMessageDecoder都有针对某个socket的累积对象
//故是一个不能够共享的对象类型
            throw new IllegalStateException("@Sharable annotation is not allowed");
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            RecyclableArrayList out = RecyclableArrayList.newInstance();
            try {
                ByteBuf data = (ByteBuf) msg;
                first = cumulation == null;
                if (first) {
                    cumulation = data;
                } else {
                    //缓冲区的大小没有超过须要写入的数据的大小
                    if (cumulation.writerIndex() > cumulation.maxCapacity() - data.readableBytes()) {
                        expandCumulation(ctx, data.readableBytes());
                    }
                    cumulation.writeBytes(data);//将数据写入到积累对象中
                    data.release();//释放bytebuffer(heap或者direct)--经过引用的方式进行释放缓冲区
                }
                //收集完毕以后解析收集到的字符串
                callDecode(ctx, cumulation, out);
            } catch (DecoderException e) {
                throw e;
            } catch (Throwable t) {
                throw new DecoderException(t);
            } finally {
//若是累积对象中没有数据了(由于全部发送的数据刚恰好n个msg)
                if (cumulation != null && !cumulation.isReadable()) {
                    cumulation.release(); 
                    cumulation = null;
                }
                int size = out.size();
                decodeWasNull = size == 0;
                //针对解析后的out结果，逐个调用message
                for (int i = 0; i < size; i ++) {
                    ctx.fireChannelRead(out.get(i));
                }
                out.recycle();
            }
        } else {
            ctx.fireChannelRead(msg);
        }
    }

    private void expandCumulation(ChannelHandlerContext ctx, int readable) {
        ByteBuf oldCumulation = cumulation;//新的容量=旧的容量+可读取的数量  ---在此处的扩展和初次的分配都是经过同一个allocator进行分配的
        cumulation = ctx.alloc().buffer(oldCumulation.readableBytes() + readable);
        cumulation.writeBytes(oldCumulation);//复制的过程
        oldCumulation.release();//释放老的对象
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (cumulation != null && !first) {//cumulation可读的数据为0那么就不会调用紧跟着的代码段
            cumulation.discardSomeReadBytes();//若是存在半包得话，那么就释放没必要要的空间（有时间的话，咱们会将一个Netty中ByteBuf的构造)
        }
        if (decodeWasNull) {
            decodeWasNull = false;
            if (!ctx.channel().config().isAutoRead()) {
                ctx.read();
            }
        }
        ctx.fireChannelReadComplete();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        RecyclableArrayList out = RecyclableArrayList.newInstance();
        try {//若是channel不活动了得话，对累积对象进行解码。
            if (cumulation != null) {
                callDecode(ctx, cumulation, out);
                decodeLast(ctx, cumulation, out);
            } else {
                decodeLast(ctx, Unpooled.EMPTY_BUFFER, out);
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Exception e) {
            throw new DecoderException(e);
        } finally {
            if (cumulation != null) {
                cumulation.release();
                cumulation = null;
            }
            int size = out.size();
            for (int i = 0; i < size; i ++) {
                ctx.fireChannelRead(out.get(i));
            }
            ctx.fireChannelInactive();
            out.recycle();
        }
    }

    /**
     * Called once data should be decoded from the given {@link ByteBuf}. This method will call
     * {@link #decode(ChannelHandlerContext, ByteBuf, List)} as long as decoding should take place.
     *
     * @param ctx           the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
     * @param in            the {@link ByteBuf} from which to read data
     * @param out           the {@link List} to which decoded messages should be added
     */
    protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        try {
            while (in.isReadable()) {
                int outSize = out.size();
                int oldInputLength = in.readableBytes();
                decode(ctx, in, out);//特别注意： 调用完若是msg被解析出来的话，那么累积对象的readableBytes必定会发生变化
                // 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.
                //
                // See https://github.com/netty/netty/issues/1664
                if (ctx.isRemoved()) {//若是此handler被移除
                    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;
                }
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable cause) {
            throw new DecoderException(cause);
        }
    }
}

由此总结以下：

1. 累积对象主要是承载socket接收的字节数的。每一次decode，当有msg生成的时候，readableBytes都会减少

2. 当一个socket处于inactive时，会对累积对象的数据进行解析。而后释放累积对象

3. 当当前的累积对象不能承载数据的时候，须要进行扩展(调用Allocator建立个新的bytebuf，而后copy一下数据)

4. 该handler是一个有状态的handler，须要记忆每个socket的发送的字节.而后再去decode成msg对象