1 解码概述

在Dubbo服务消费方（客户端）和服务提供方（服务端）进行网络通信时，服务提供方会通过socket把需要发送的内容序列化为二进制流后发出。接着二进制流通过网络流向服务提供方。服务提供方接收到该请求后会解析该请求包，对接收到的数据进行反序列化后对请求进行处理。

服务提供方接收到请求后解析请求包（即Dubbo协议的内容）的过程即服务提供方的解码过程。

2 解码原理解析

2.1 解码入口

服务提供方解析请求包最终是借助于 InternalDecoder 的 decode 方法来实现的。具体实现代码如下所示。

protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception {

    ChannelBuffer message = new NettyBackedChannelBuffer(input);

    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);

    // decode object.
    do {
        int saveReaderIndex = message.readerIndex();
        Object msg = codec.decode(channel, message);
        if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
            message.readerIndex(saveReaderIndex);
            break;
        } else {
            //is it possible to go here ?
            if (saveReaderIndex == message.readerIndex()) {
                throw new IOException("Decode without read data.");
            }
            if (msg != null) {
                out.add(msg);
            }
        }
    } while (message.readable());
}

2.2 解码实现细节

在 InternalDecoder 的 decode 方法内部再调用了具体解码实现来进行解码。如 ExchangeCodec 的 decode 方法，实现如下所示。

public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
    // 将dubbo协议头读取到数据header
    int readable = buffer.readableBytes();
    byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
    buffer.readBytes(header);
    // 解析dubbo协议帧数据部分
    return decode(channel, buffer, readable, header);
}

@Override
protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
    // check magic number.
    if (readable > 0 && header[0] != MAGIC_HIGH
        || readable > 1 && header[1] != MAGIC_LOW) {
        int length = header.length;
        if (header.length < readable) {
            header = Bytes.copyOf(header, readable);
            buffer.readBytes(header, length, readable - length);
        }
        for (int i = 1; i < header.length - 1; i++) {
            if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
                buffer.readerIndex(buffer.readerIndex() - header.length + i);
                header = Bytes.copyOf(header, i);
                break;
            }
        }
        return super.decode(channel, buffer, readable, header);
    }
    // check length.
    if (readable < HEADER_LENGTH) {
        return DecodeResult.NEED_MORE_INPUT;
    }

    // get data length.
    int len = Bytes.bytes2int(header, 12);

    // When receiving response, how to exceed the length, then directly construct a response to the client.
    // see more detail from https://github.com/apache/dubbo/issues/7021.
    Object obj = finishRespWhenOverPayload(channel, len, header);
    if (null != obj) {
        return obj;
    }

    int tt = len + HEADER_LENGTH;
    if (readable < tt) {
        return DecodeResult.NEED_MORE_INPUT;
    }

    // limit input stream.
    ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);

    try {
        return decodeBody(channel, is, header);
    } finally {
        if (is.available() > 0) {
            try {
                if (logger.isWarnEnabled()) {
                    logger.warn(TRANSPORT_SKIP_UNUSED_STREAM, "", "", "Skip input stream " + is.available());
                }
                StreamUtils.skipUnusedStream(is);
            } catch (IOException e) {
                logger.warn(TRANSPORT_SKIP_UNUSED_STREAM, "", "", e.getMessage(), e);
            }
        }
    }
}

而对dubbo协议body进行解析则是通过 decodeBody 方法，实现如下所示。

protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
    byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
    // get request id.
    long id = Bytes.bytes2long(header, 4);
    if ((flag & FLAG_REQUEST) == 0) {
        // decode response.
        Response res = new Response(id);
        if ((flag & FLAG_EVENT) != 0) {
            res.setEvent(true);
        }
        // get status.
        byte status = header[3];
        res.setStatus(status);
        try {
            if (status == Response.OK) {
                Object data;
                if (res.isEvent()) {
                    byte[] eventPayload = CodecSupport.getPayload(is);
                    if (CodecSupport.isHeartBeat(eventPayload, proto)) {
                        // heart beat response data is always null;
                        data = null;
                    } else {
                        data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload);
                    }
                } else {
                    data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(channel, res, id));
                }
                res.setResult(data);
            } else {
                res.setErrorMessage(CodecSupport.deserialize(channel.getUrl(), is, proto).readUTF());
            }
        } catch (Throwable t) {
            res.setStatus(Response.CLIENT_ERROR);
            res.setErrorMessage(StringUtils.toString(t));
        }
        return res;
    } else {
        // decode request.
        Request req;
        try {
            Object data;
            if ((flag & FLAG_EVENT) != 0) {
                byte[] eventPayload = CodecSupport.getPayload(is);
                if (CodecSupport.isHeartBeat(eventPayload, proto)) {
                    // heart beat response data is always null;
                    req = new HeartBeatRequest(id);
                    ((HeartBeatRequest) req).setProto(proto);
                    data = null;
                } else {
                    req = new Request(id);
                    data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload);
                }
                req.setEvent(true);
            } else {
                req = new Request(id);
                data = decodeRequestData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto));
            }
            req.setData(data);
        } catch (Throwable t) {
            // bad request
            req = new Request(id);
            req.setBroken(true);
            req.setData(t);
        }
        req.setVersion(Version.getProtocolVersion());
        req.setTwoWay((flag & FLAG_TWOWAY) != 0);
        return req;
    }
}