netty源码阅读之解码之抽象解码器ByteMessageDecoder

- 日理万妓 2022-05-10 03:28 474阅读 0赞

ByteMessageDecoder是所有解码器的基类，它主要通过以下步骤进行解码：

1、累加字节流

2、调用子类的decode方法进行解析

3、将解析到的ByteBuf向下传播

我们看ByteMessageDecoder的channelRead方法，这个方法就是解码的开始：

@Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof ByteBuf) {
                CodecOutputList out = CodecOutputList.newInstance();
                try {
                    ByteBuf data = (ByteBuf) msg;
                    first = cumulation == null;
                    if (first) {
                        cumulation = data;
                    } else {
                        cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
                    }
                    callDecode(ctx, cumulation, out);
                } catch (DecoderException e) {
                    throw e;
                } catch (Throwable t) {
                    throw new DecoderException(t);
                } finally {
                    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.
                        // See https://github.com/netty/netty/issues/4275
                        numReads = 0;
                        discardSomeReadBytes();
                    }
    
                    int size = out.size();
                    decodeWasNull = !out.insertSinceRecycled();
                    fireChannelRead(ctx, out, size);
                    out.recycle();
                }
            } else {
                ctx.fireChannelRead(msg);
            }
        }

通过if (msg instanceof ByteBuf)判断是否ByteBuf，只有ByteBuf才进行解码，否则调用ctx.fireChannelRead向下传播

### 一、累加字节流 ###

cumulation的定义是一个ByteBuf：

ByteBuf cumulation;

如果cumulation为空，就直接赋值，否则就累加：

first = cumulation == null;
                    if (first) {
                        cumulation = data;
                    } else {
                        cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
                    }

我们看看这个累加器cumulator的定义：

private Cumulator cumulator = MERGE_CUMULATOR;

然后：

/**
         * Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
         */
        public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
            @Override
            public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
                ByteBuf buffer;
                if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
                        || cumulation.refCnt() > 1) {
                    // 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().
                    //
                    // See:
                    // - https://github.com/netty/netty/issues/2327
                    // - https://github.com/netty/netty/issues/1764
                    buffer = expandCumulation(alloc, cumulation, in.readableBytes());
                } else {
                    buffer = cumulation;
                }
                buffer.writeBytes(in);
                in.release();
                return buffer;
            }
        };

这是一个匿名内部类的实现，看

cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()

我们把它转换一下其实就是：cumulation.writerIndex()+ in.readableBytes() > cumulation.maxCapacity()

如果原来的数据加上读进来的数据超过了最大容量，那就增加容量：

buffer = expandCumulation(alloc, cumulation, in.readableBytes());

增加容量的方法也很简单，每次只增加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;
        }

然后把把值赋给buffer，

否则直接赋值：buffer = cumulation;

然后往buffer里面写in进来的数据，最后释放in：

buffer.writeBytes(in);
                in.release();

## ##

### 二、调用子类的decode方法进行解析 ###

进入callDecode方法：

protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
            try {
                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.
                        //
                        // See:
                        // - https://github.com/netty/netty/issues/4635
                        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.
                    //
                    // See https://github.com/netty/netty/issues/1664
                    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;
                    }
                }
            } catch (DecoderException e) {
                throw e;
            } catch (Throwable cause) {
                throw new DecoderException(cause);
            }
        }

首先看while (in.isReadable())证明这是一个不停地循环，然后看这一段：

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

out是什么，是一个list，是解析得到的对象的list，在我们不停地循环中，如果解析到对象了，那就往下传播。解析的过程在下面：

decode(ctx, in, out);

然后看定义：

/**
         * Decode the from one {@link ByteBuf} to an other. This method will be called till either the input
         * {@link ByteBuf} has nothing to read when return from this method or till nothing was read from the input
         * {@link ByteBuf}.
         *
         * @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
         * @throws Exception    is thrown if an error accour
         */
        protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
    
        /**

这是一个留给子类去实现的方法，in进去，如果解析到对象，就放在out里面，out的size就不为零。

我们把现在的out的size和之前保存的比较：

if (outSize == out.size()) {
                        if (oldInputLength == in.readableBytes()) {
                            break;
                        } else {
                            continue;
                        }
                    }

如果相同，证明没有解析到对象，没有解析到对象分两种情况：1、数据长度不够长，不足解析出来  2、数据长度够，解析出来了，但是解析出来的东西不够合成一个对象。

我们在decode之前把in的readableBytes保存下来：

int oldInputLength = in.readableBytes();

如果解析过后，长度和之前一样，那就说明不足以解析，直接跳出while循环，等你下次够数据了再来。如果和之前不一样，那就说明够解析，只是没有解析为一个对象而已，那就继续解析。好好体会。

好了，继续看下面 ：

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

走到这里，一定是解析出了一个对象，因为out的size和之前不同了，那么如果解析出了一个对象，数据长度和之前的一样，这个就有问题了：解析出了对象数据却没有减少，就抛出异常。

最后，如果是只解析一个对象就够了，那就断开：

if (isSingleDecode()) {
                        break;
                    }

在这些解析之前和之后还要判断ctx是不是被移除了，这里不一一赘述，看源码：

// 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.
                        //
                        // See:
                        // - https://github.com/netty/netty/issues/4635
                        if (ctx.isRemoved()) {
                            break;
                        }

### 三、将解析到的ByteBuf向下传播 ###

ByteMessageDecoder的channalRead最后这一步就是把解析到的数据继续往下传播：

finally {
                    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.
                        // See https://github.com/netty/netty/issues/4275
                        numReads = 0;
                        discardSomeReadBytes();
                    }
    
                    int size = out.size();
                    decodeWasNull = !out.insertSinceRecycled();
                    fireChannelRead(ctx, out, size);
                    out.recycle();
                }

看fireChannelRead(ctx,out,size)方法：

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

把得到的对象一个个往下传播。。。

下一篇文章我们将解析一个最简单的解码器。