0%

dubbo协议报文消息格式

Posted on In

“谷神不死，是谓玄牝。
玄牝之门，是谓天地根。
绵绵若存，用之不勤。”1

dubbo默认采用netty进行网络传输，协议中对字节流的处理在com.alibaba.dubbo.remoting.exchange.codec.ExchangeCodec类中，包含了对request请求的编码和解码，response响应的编码和解码。
dubbo协议采用固定长度的消息头（16字节）和不定长度的消息体来进行数据传输，消息头定义了netty在IO线程处理时需要的信息，协议的报文格式如下：
dubbo_message_protocol_header

消息头详解

协议头是16字节的定长数据:

1
2byte magic:类似java字节码文件里的魔数，用来判断是不是dubbo协议的数据包。魔数是常量0xdabb
2
1byte 的消息标志位:16-20序列id,21 event,22 two way,23请求或响应标识
3
1byte 状态，当消息类型为响应时，设置响应状态。24-31位。状态位, 设置请求响应状态，dubbo定义了一些响应的类型。具体类型见com.alibaba.dubbo.remoting.exchange.Response
4
8byte 消息ID,long类型，32-95位。每一个请求的唯一识别id（由于采用异步通讯的方式，用来把请求request和返回的response对应上）
5
4byte 消息长度，96-127位。消息体 body 长度, int 类型，即记录Body Content有多少个字节。

解码过程 ExchangeCodec->encode-> encodeRequest

1
public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
2
        if (msg instanceof Request) {
3
            encodeRequest(channel, buffer, (Request) msg);
4
        } else if (msg instanceof Response) {
5
            encodeResponse(channel, buffer, (Response) msg);
6
        } else {
7
            super.encode(channel, buffer, msg);
8
        }
9
}
1
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
2
     Serialization serialization = getSerialization(channel);
3
     // header.
4
     byte[] header = new byte[HEADER_LENGTH];
5
     // set magic number.
6
     Bytes.short2bytes(MAGIC, header);
7
8
     // set request and serialization flag.
9
     header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
10
11
     if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
12
     if (req.isEvent()) header[2] |= FLAG_EVENT;
13
14
     // set request id.
15
     Bytes.long2bytes(req.getId(), header, 4);
16
17
     // encode request data.
18
     int savedWriteIndex = buffer.writerIndex();
19
     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
20
     ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
21
     ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
22
     if (req.isEvent()) {
23
         encodeEventData(channel, out, req.getData());
24
     } else {
25
         encodeRequestData(channel, out, req.getData());
26
     }
27
     out.flushBuffer();
28
     if (out instanceof Cleanable) {
29
         ((Cleanable) out).cleanup();
30
     }
31
     bos.flush();
32
     bos.close();
33
     int len = bos.writtenBytes();
34
     checkPayload(channel, len);
35
     Bytes.int2bytes(len, header, 12);
36
37
     // write
38
     buffer.writerIndex(savedWriteIndex);
39
     buffer.writeBytes(header); // write header.
40
     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
41
 }

消息体详解

实现源码在DubboCodec.encodeRequestData(Channel channel, ObjectOutput out, Object data)：

1
protected void encodeRequestData(Channel channel, ObjectOutput out, Object data) throws IOException {
2
    RpcInvocation inv = (RpcInvocation) data;
3
4
    out.writeUTF(inv.getAttachment(Constants.DUBBO_VERSION_KEY, DUBBO_VERSION));
5
    out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
6
    out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
7
8
    out.writeUTF(inv.getMethodName());
9
    out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
10
    Object[] args = inv.getArguments();
11
    if (args != null)
12
        for (int i = 0; i < args.length; i++) {
13
            out.writeObject(encodeInvocationArgument(channel, inv, i));
14
        }
15
    out.writeObject(inv.getAttachments());
16
}

消息体的内容如下：
1、dubbo版本号
2、invoke的路径
3、invoke的provider端暴露的服务的版本号
4、调用的方法名称
5、参数类型描述符
6、遍历请求参数值并编码
7、dubbo请求的attachments

1:老子《道德经》第六章，老子故里，中国鹿邑。