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的协议设计不是很复杂，只要我们耐心一步步跟进去，看到最里面的代码。"哦，原来是这样!"