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

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

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

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

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

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

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