RocketMq通信协议格式及编解码（源码分析）

一、RocketMq broker服务器与客户端的网络通信是基于netty4.x实现的，重点分析 RocketMq设计的通信协议及对应的编解码开发。

名字解释

编码：将java对象转换成二进制数据，用于放到网络中进行传输

解码：将从网络中读取到的二进制数据转换成相应的java对象

二、Remoting设计的通信协议格式如下（重点理解，能根据通信协议格式来对网络中读取的二进制数据进行编解码）：

协议格式 <length> <header length> <header data> <body data>

1 2 3 4

1、4个字节的int型数据来存储2、3、4的总长度

2、4个字节的int型数据来存储报文头部的字节长度等于3的长度

3、存储报文头部的数据

4、存储报文体的数据

三、在通信过程中，服务器与客户端通过传递 RemotingCommand 对象来进行交互，下面将根据代码实现来分析通信协议的编解码开发

3.1 分析RemotingCommand 类encode()方法，它将按照上述定义的协议格式进行对象的编码操作：

public ByteBuffer encode() {

// 1> header length size

int length = 4;   //表示用4个字节来存储头部长度

// 2> header data length

byte[] headerData = this.buildHeader(); //报文头部的数据

length += headerData.length;   //加上头部报文的字节长度

// 3> body data length

if (this.body != null) {

length += body.length;    //如果报文体body有数据则加上报文体的字节长度

}

ByteBuffer result = ByteBuffer.allocate(4 + length); //分配一个 (4+length)这么大的字节缓冲区，这个缓冲区就用来存储上述协议格式的整个报文的数据

  //下面代码开始往缓冲区存放数据

// length

result.putInt(length); //缓冲区的最开始的4个字节用来存储总的长度length

// header length

result.putInt(headerData.length); //缓冲区接下来4个字节用来存储报文头部的长度

// header data

result.put(headerData); //缓冲区接下来存储报文头部数据

// body data;

if (this.body != null) {

result.put(this.body);   //缓冲区最后用来存储报文体的数据

}

result.flip(); //将缓冲区翻转，用于将ByteBuffer放到网络通道中进行传输

return result;

}

3.2 分析RemotingCommand 类decode()方法，它将按照上述定义的协议格式进行各个报文段的字节数据读取，然后转换成RemotingCommand对象：

public class NettyDecoder extends LengthFieldBasedFrameDecoder {

private static final Logger log = LoggerFactory.getLogger(RemotingHelper.RemotingLogName);

private static final int FRAME_MAX_LENGTH = //

Integer.parseInt(System.getProperty("com.rocketmq.remoting.frameMaxLength", "8388608"));

public NettyDecoder() {

super(FRAME_MAX_LENGTH, 0, 4, 0, 4); //0,4,0,4 每一个0,4,这个是表示存放长度的变量的字节所占的长度，为4个，第二个4表示就是解码之后的数据包跳过的字节数为4,表示就将数据包的头部给去掉了。

}

//此处省略一万字

}

//在调用decode()方法解码之前，会调用上述NettyDecoder 类的decode()方法，在上述构造方法中，会先去掉报文的前4个字节，这4个字节是存储的后面报文的长度.

public static RemotingCommand decode(final ByteBuffer byteBuffer) {

int length = byteBuffer.limit();   //获取字节缓冲区的整个长度，这个长度等于通信协议格式的2、3、4段的总长度

int headerLength = byteBuffer.getInt(); //从缓冲区中读取4个字节的int类型的数据值，这个值就是报文头部的长度

byte[] headerData = new byte[headerLength];

byteBuffer.get(headerData); //接下来从缓冲区中读取headerLength个字节的数据，这个数据就是报文头部的数据

int bodyLength = length - 4 - headerLength;

byte[] bodyData = null;

if (bodyLength > 0) {

bodyData = new byte[bodyLength];

byteBuffer.get(bodyData);   //接下来读取length-4-headerLength 个字节的数据，这个数据就是报文体的数据

}

//接下来将读取到的数据转换成 RemotingCommand 对象

RemotingCommand cmd = RemotingSerializable.decode(headerData, RemotingCommand.class);

cmd.body = bodyData;

return cmd;

}

四、总结

开发任何的socket 长连接的网络程序，涉及服务器与客户端的开发，首先要定义服务端与客户端的通信协议格式，第二根据定义的通信协议格式来进行

传输数据的编解码操作。上述的通信协议格式为常用的通信协议格式。当长连接在设定的间隔时间范围内没有数据传输时，需要按照协议发送心跳包，

心跳包的协议格式也可以按照这种协议格式发送，也可以另外针对发送的心跳包来定义通信协议格式。

RocketMq通信协议格式及编解码 （源码分析）

RocketMq通信协议格式及编解码（源码分析）