当前位置: 首页>编程语言>正文

Netty编解码器

概述

网络中数据是以二进制字节流进行传输,编码器的作用是将数据编码为二进制字节流,而解码器的作用是将二进制字节流解码为程序能处理的数据格式,本章节将对Netty编解码器实现原理进行分析。

编码器

Netty编解码器,第1张

Netty编码器有两个抽象基类:MessageToByteEncoder与MessageToMessageEncoder,MessageToByteEncoder会将消息编码为字节,MessageToMessageEncode会将消息编码为另一种消息。通过继承这两个基类可以实现自定义编码器,如Netty内置的编码器:StringEncoder、ObjectEncoder、ProtobufEncoder等。

从类图可以看出MessageToByteEncoder与MessageToMessageEncoder均实现了ChannelOutboundHandler接口,也就是说编码器是一个出站事件处理器,其会处理并传播出站事件。以MessageToByteEncoder为例,分析下其实现原理:

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        // 判断编码器是否能处理该消息
        if (acceptOutboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            // 将Object消息类型转换为实际类型
            I cast = (I) msg;
            // 分配内存
            buf = allocateBuffer(ctx, cast, preferDirect);
            try {
                // 调用子类方法进行编码
                encode(ctx, cast, buf);
            } finally {
                // 如果消息是ByteBuf类型则释放内存
                ReferenceCountUtil.release(cast);
            }

            if (buf.isReadable()) {
                // 传播写事件
                ctx.write(buf, promise);
            } else {
                buf.release();
                ctx.write(Unpooled.EMPTY_BUFFER, promise);
            }
            buf = null;
        } else {
            ctx.write(msg, promise);
        }
    } catch (EncoderException e) {
        throw e;
    } catch (Throwable e) {
        throw new EncoderException(e);
    } finally {
        if (buf != null) {
            buf.release();
        }
    }
}

write方法首先会调用acceptOutboundMessage方法判断是否能处理传入的消息,如果不能处理则将该事件向下传播。acceptOutboundMessage通过TypeParameterMatcher#match方法来进行判断。其中TypeParameterMatcher是在实例化编码器时初始化的,具体判断方法在其内部类ReflectiveMatcher中,其中type为编码器中的泛型:

private static final class ReflectiveMatcher extends TypeParameterMatcher {
    private final Class<?> type;

    ReflectiveMatcher(Class<?> type) {
        this.type = type;
    }

    @Override
    public boolean match(Object msg) {
        return type.isInstance(msg);
    }
}

编码器的核心方法为encode,该方法会将消息编码为ByteBuf。基类MessageToMessageEncode中的encode方法为抽象方法,这里使用了模版方法设计模式,需要由子类实现具体的编码逻辑:

protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;

以ObjectEncoder为例,看下ObjectEncoder这个编码器是怎么编码的:

protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
  // 获取ByteBuf对象写指针索引位置  
  int startIdx = out.writerIndex();

    // 创建输出流
    ByteBufOutputStream bout = new ByteBufOutputStream(out);
    // 写入4个字节进行占位
    bout.write(LENGTH_PLACEHOLDER);
    ObjectOutputStream oout = new CompactObjectOutputStream(bout);
    // 将消息写入到输出流
    oout.writeObject(msg);
    oout.flush();
    oout.close();

    int endIdx = out.writerIndex();
        // 在输入流开始位置写入int类型的消息长度,即前四个字节
    out.setInt(startIdx, endIdx - startIdx - 4);
}

解码器

Netty编解码器,第2张

Netty解码器同样有两个抽象基类:MessageToByteDecoder与MessageToMessageDecoder,MessageToByteDecoder会将字节解码为消息,MessageToMessageDecoder会将消息解码为另一种消息。以MessageToByteDecoder为例,看下具体实现:

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        // 获取CodecOutputList,用于存放解码后的消息
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            ByteBuf data = (ByteBuf) msg;
            first = cumulation == null;
            if (first) {
                // 第一次解码将数据赋值给cumulation
                cumulation = data;
            } else {
                // 不是第一次解码则将数据进行累加到cumulation
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            // 解码
            callDecode(ctx, cumulation, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable t) {
            throw new DecoderException(t);
        } finally {
            // 如果cumulation不可读则,则释放cumulation
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
            } else if (++ numReads >= discardAfterReads) {
                // 如果读次数超过16,则尝试丢弃一些字节
                // 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);
    }
}

CodecOutputList是一个可复用的List,内部维护了一个长度为16的对象数组,用来缓存解码后的消息。CodecOutputList#newInstance方法首先会从对象池中获取,如果获取不到会进行创建。

读取到消息后,首先会将其放入到cumulation这个ByteBuf中,然后尝试解码,每次解码成功就会传播channelRead事件,将消息向下传播。如果消息不能解码,比如出现半包,或者出现粘包(一些字节不足以解码为一条消息),则会被累加到cumulation中等待下次接收到消息后处理。

解码的核心逻辑在callDecode中,该方法会将cumulation中累加的消息进行解码:

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);
    }
}

其中decode是一个模版方法,需要由子类解码器实现具体的解码逻辑,解码后的消息会放入到CodecOutputList中。

解码后会调用fireChannelRead方法传播事件,该方法内部会遍历CodecOutputList集合中解码后的消息,将消息传播给下一个处理器:

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

https://www.xamrdz.com/lan/57q2016125.html

相关文章: