Netty源代码之FrameDecoder

野性酷女 2022-08-25 08:40 103阅读 0赞

FrameDecoder说白了是解码器，可以将网络数据一帧一帧地解码出来

主要看起messageReceived

public void messageReceived(
                ChannelHandlerContext ctx, MessageEvent e) throws Exception {
    
            Object m = e.getMessage();
            if (!(m instanceof ChannelBuffer)) {
                ctx.sendUpstream(e);
                return;
            }
    
            ChannelBuffer input = (ChannelBuffer) m;
            if (!input.readable()) {
                return;
            }
    
            if (cumulation == null) {
                try {
                    // the cumulation buffer is not created yet so just pass the input to callDecode(...) method
                    callDecode(ctx, e.getChannel(), input, e.getRemoteAddress());
                } finally {
                    updateCumulation(ctx, input);
                }
    
            } else {
                input = appendToCumulation(input);
                try {
                    callDecode(ctx, e.getChannel(), input, e.getRemoteAddress());
                } finally {
                    updateCumulation(ctx, input);
                }
            }
        }

这里这个cumulation是关键，cumulation的是"累计的意思"，即将数据累计起来，首先判断cululation是不是null,是null的话直接调用callDecode方法，不然先将input添加到culutation

callDecode()

private void callDecode(
                ChannelHandlerContext context, Channel channel,
                ChannelBuffer cumulation, SocketAddress remoteAddress) throws Exception {
    
            while (cumulation.readable()) {
                int oldReaderIndex = cumulation.readerIndex();
                Object frame = decode(context, channel, cumulation);
                if (frame == null) {
                    if (oldReaderIndex == cumulation.readerIndex()) {
                        // Seems like more data is required.
                        // Let us wait for the next notification.
                        break;
                    } else {
                        // Previous data has been discarded.
                        // Probably it is reading on.
                        continue;
                    }
                } else if (oldReaderIndex == cumulation.readerIndex()) {
                    throw new IllegalStateException(
                            "decode() method must read at least one byte " +
                            "if it returned a frame (caused by: " + getClass() + ")");
                }
    
                unfoldAndFireMessageReceived(context, remoteAddress, frame);
            }
        }

在unfoldAndFireMessageReceived()里面，会调用Channels.fireMessageReceived（），最后调用ctx.sendUpstream，事件会交给下一个Handler来处理

ChannelHandlerContext context, SocketAddress remoteAddress, Object result) {
            if (unfold) {
                if (result instanceof Object[]) {
                    for (Object r: (Object[]) result) {
                        Channels.fireMessageReceived(context, r, remoteAddress);
                    }
                } else if (result instanceof Iterable<?>) {
                    for (Object r: (Iterable<?>) result) {
                        Channels.fireMessageReceived(context, r, remoteAddress);
                    }
                } else {
                    Channels.fireMessageReceived(context, result, remoteAddress);
                }
            } else {
                Channels.fireMessageReceived(context, result, remoteAddress);
            }
        }

appendToCumulation（）这里可以看出cumulation是CompositeChannelBuffer,当之前还有buffer没处理完时，与这次的buffer合并一起进行解码

protected ChannelBuffer appendToCumulation(ChannelBuffer input) {
            ChannelBuffer cumulation = this.cumulation;
            assert cumulation.readable();
            if (cumulation instanceof CompositeChannelBuffer) {
                // Make sure the resulting cumulation buffer has no more than the configured components.
                CompositeChannelBuffer composite = (CompositeChannelBuffer) cumulation;
                if (composite.numComponents() >= maxCumulationBufferComponents) {
                    cumulation = composite.copy();
                }
            }
    
            this.cumulation = input = ChannelBuffers.wrappedBuffer(cumulation, input);
            return input;
        }

另外一个函数时finally里面的updateCumulation（）

protected ChannelBuffer updateCumulation(ChannelHandlerContext ctx, ChannelBuffer input) {
            ChannelBuffer newCumulation;
            int readableBytes = input.readableBytes();
            if (readableBytes > 0) {
                int inputCapacity = input.capacity();
    
                // If input.readableBytes() == input.capacity() (i.e. input is full),
                // there's nothing to save from creating a new cumulation buffer
                // even if input.capacity() exceeds the threshold, because the new cumulation
                // buffer will have the same capacity and content with input.
                if (readableBytes < inputCapacity && inputCapacity > copyThreshold) {
                    // At least one byte was consumed by callDecode() and input.capacity()
                    // exceeded the threshold.
                    cumulation = newCumulation = newCumulationBuffer(ctx, input.readableBytes());
                    cumulation.writeBytes(input);
                } else {
                    // Nothing was consumed by callDecode() or input.capacity() did not
                    // exceed the threshold.
                    if (input.readerIndex() != 0) {
                        cumulation = newCumulation = input.slice();
                    } else {
                        cumulation = newCumulation = input;
                    }
                }
            } else {
                cumulation = newCumulation = null;
            }
            return newCumulation;
        }

在这个函数里面会对cumulation进行操作修改。这个是在finally，也就是对剩余的数据进行暂时的存取，因为不管是这个buffer比一帧数据大或者小，都不能只是一帧数据，少于一帧不能解码出来，大于一帧势必会有一部分需要暂时存取