RocketMQ源码：通信协议设计及编解码

本文主要分析RocketMQ通信协议的设计。RocketMQ设计了自己的一个通信协议，用于消息内容和二进制格式之间的转换。

RocketMQ的版本为：4.2.0 release。

一.通信协议的格式

1.length：4字节整数，二三四部分长度总和；

2.header length：4字节整数，第三部分header data长度；

3.header data：存放Json序列化的数据；

4.body data：应用自定义二进制序列化的数据。

二.消息的编码过程

消息的编码是在 RemotingCommand 中 encode 方法中完成的：

public ByteBuffer encode() {
// 1> header length size
int length = 4;
// 2> header data length
byte[] headerData = this.headerEncode();
length += headerData.length;
// 3> body data length
if (this.body != null) {
length += body.length;
}
ByteBuffer result = ByteBuffer.allocate(4 + length);
// 1.先放入消息的总大小
result.putInt(length);
// 2.再放入头部的长度
result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
// 3.接着放入头部数据
result.put(headerData);
// 4.最后放入消息体的数据
if (this.body != null) {
result.put(this.body);
}
result.flip();
return result;
}

对于头部数据 - header data 的编码，在 RemotingCommand 中 headerEncode 方法处理：

private byte[] headerEncode() {
this.makeCustomHeaderToNet();
if (SerializeType.ROCKETMQ == serializeTypeCurrentRPC) {
return RocketMQSerializable.rocketMQProtocolEncode(this);
} else {
return RemotingSerializable.encode(this);
}
}

还有一个细节的地方，保存 header length 的时候，经过了一个 markProtocolType 的处理，作用是将RPC类型和headerData长度编码放到一个byte[4]数组中：

public static byte[] markProtocolType(int source, SerializeType type) {
byte[] result = new byte[4];
result[0] = type.getCode();
result[1] = (byte) ((source >> 16) & 0xFF);
result[2] = (byte) ((source >> 8) & 0xFF);
result[3] = (byte) (source & 0xFF);
return result;
}

记住这个方法，后面从消息数据解码消息头长度的时候，就不会看的很迷茫。

三.消息的解码过程

消息的解码是在类 RemotingCommand 中 decode方法中完成的：

public static RemotingCommand decode(final ByteBuffer byteBuffer) {
int length = byteBuffer.limit();// 获取byteBuffer的总长度
int oriHeaderLen = byteBuffer.getInt();// 1.获取前4个字节，组装int类型，该长度为总长度 图中 length
int headerLength = getHeaderLength(oriHeaderLen);// length & 0xFFFFFF 获取消息头的长度，与运算，编码时候的长度即为24位
byte[] headerData = new byte[headerLength];// 保存header data
byteBuffer.get(headerData);// 2.从缓冲区中读取headerLength个字节的数据，这个数据就是报文头部的数据
RemotingCommand cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));
int bodyLength = length - 4 - headerLength;// 报文体的数据，减去了第二、三部分的长度
byte[] bodyData = null;
if (bodyLength > 0) {
bodyData = new byte[bodyLength];
byteBuffer.get(bodyData);// 获取消息体的数据
}
cmd.body = bodyData;
return cmd;
}

对于头部数据 - header data 的解码，在 RemotingCommand 中 headerDecode 方法处理：

private static RemotingCommand headerDecode(byte[] headerData, SerializeType type) {
switch (type) {
case JSON:
RemotingCommand resultJson = RemotingSerializable.decode(headerData, RemotingCommand.class);
resultJson.setSerializeTypeCurrentRPC(type);
return resultJson;
case ROCKETMQ:
RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(headerData);
resultRMQ.setSerializeTypeCurrentRPC(type);
return resultRMQ;
default:
break;
}
return null;
}

解码出 header length - 消息头长度，在 getHeaderLength 中：

public static int getHeaderLength(int length) {
return length & 0xFFFFFF;// 为什么是和高位的24位与，可以参考编码时的方法：markProtocolType
}

通过代码分析，发现RocketMQ的协议设计不是很复杂，只要我们耐心一步步跟进去，看到最里面的代码。"哦，原来是这样!"