RocketMQ消息存储和查询实战
RocketMQ 作为⼀款优秀的分布式消息中间件，可以为业务⽅提供⾼性能低延迟的稳定可靠的消息服务。其核⼼优势是可靠的消费存储、消息发送的⾼性能和低延迟、强⼤的消息堆积能⼒和消息处理能⼒。
从存储⽅式来看，主要有⼏个⽅⾯：
⽂件系统
分布式KV存储
关系型数据库
从效率上来讲，⽂件系统⾼于KV存储，KV存储⼜⾼于关系型数据库。因为直接操作⽂件系统肯定是最快的，那么业界主流的消息队列中间件，如RocketMQ 、RabbitMQ 、kafka 都是采⽤⽂件系统的⽅式来存储消息。
今天，我们就从它的存储⽂件⼊⼿，来探索⼀下 RocketMQ 消息存储的机制。
⼀、CommitLog
CommitLog，消息存储⽂件，所有主题的消息都存储在 CommitLog ⽂件中。
我们的业务系统向 RocketMQ 发送⼀条消息，不管在中间经历了多么复杂的流程，最终这条消息会被持久化到CommitLog⽂件。
我们知道，⼀台Broker服务器只有⼀个CommitLog⽂件(组)，RocketMQ会将所有主题的消息存储在同⼀个⽂件中，这个⽂件中就存储着⼀条条Message，每条Message都会按照顺序写⼊。
也许有时候，你会希望看看这个 CommitLog ⽂件中，存储的内容到底长什么样⼦？
当然，我们需要先往 CommitLog ⽂件中写⼊⼀些内容，所以先来看⼀个消息发送的例⼦。
public static void main(String[] args) throws Exception {
MQProducer producer = getProducer();
for (int i = 0;i<10;i++){
Message message = new Message();
message.setTopic("topic"+i);
message.setBody(("清幽之地的博客").getBytes());
SendResult sendResult = producer.send(message);
}
producer.shutdown();
}
我们向10个不同的主题中发送消息，如果只有⼀台Broker机器，它们会保存到同⼀个CommitLog⽂件中。此时，这个⽂件的位置处于
C:/Users/shiqizhen/store/commitlog/00000000000000000000。
rabbitmq rocketmq kafka区别
这个⽂件我们不能直接打开，因为它是⼀个⼆进制⽂件，所以我们需要通过程序来读取它的字节数组。
public static ByteBuffer read(String path)throws Exception{
File file = new File(path);
FileInputStream fin = new FileInputStream(file);
byte[] bytes = new byte[(int)file.length()];
ByteBuffer buffer = ByteBuffer.wrap(bytes);
return buffer;
}
如上代码，可以通过传⼊⽂件的路径，读取该⽂件所有的内容。为了⽅便下⼀步操作，我们把读取到的字节数组转换为java.nio.ByteBuffer对象。
在解析之前，我们需要弄明⽩两件事：
消息的格式，即⼀条消息包含哪些字段；
每个字段所占的字节⼤⼩。
在上⾯的图中，我们已经看到了消息的格式，包含了19个字段。关于字节⼤⼩，有的是 4 字节，有的是 8 字节，我们不再⼀⼀赘述，直接看代码。
/**
* commitlog ⽂件解析
* @param byteBuffer
* @return
* @throws Exception
*/
public static MessageExt decodeCommitLog(ByteBuffer byteBuffer)throws Exception {
MessageExt msgExt = new MessageExt();
MessageExt msgExt = new MessageExt();
// 1 TOTALSIZE
int storeSize = Int();
msgExt.setStoreSize(storeSize);
if (storeSize<=0){
return null;
}
// 2 MAGICCODE
// 3 BODYCRC
int bodyCRC = Int();
msgExt.setBodyCRC(bodyCRC);
/
/ 4 QUEUEID
int queueId = Int();
msgExt.setQueueId(queueId);
// 5 FLAG
int flag = Int();
msgExt.setFlag(flag);
// 6 QUEUEOFFSET
long queueOffset = Long();
msgExt.setQueueOffset(queueOffset);
// 7 PHYSICALOFFSET
long physicOffset = Long();
msgExt.setCommitLogOffset(physicOffset);
// 8 SYSFLAG
int sysFlag = Int();
msgExt.setSysFlag(sysFlag);
// 9 BORNTIMESTAMP
long bornTimeStamp = Long();
msgExt.setBornTimestamp(bornTimeStamp);
// 10 BORNHOST
int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16;
byte[] bornHost = new byte[bornhostIPLength];
<(bornHost, 0, bornhostIPLength);
int port = Int();
msgExt.setBornHost(new ByAddress(bornHost), port));
// 11 STORETIMESTAMP
long storeTimestamp = Long();
msgExt.setStoreTimestamp(storeTimestamp);
// 12 STOREHOST
int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16; byte[] storeHost = new byte[storehostIPLength];
<(storeHost, 0, storehostIPLength);
port = Int();
msgExt.setStoreHost(new ByAddress(storeHost), port));
// 13 RECONSUMETIMES
int reconsumeTimes = Int();
msgExt.setReconsumeTimes(reconsumeTimes);
// 14 Prepared Transaction Offset
long preparedTransactionOffset = Long();
msgExt.setPreparedTransactionOffset(preparedTransactionOffset);
// 15 BODY
int bodyLen = Int();
if (bodyLen > 0) {
byte[] body = new byte[bodyLen];
<(body);
<(body);
msgExt.setBody(body);
}
// 16 TOPIC
byte topicLen = ();
byte[] topic = new byte[(int) topicLen];
<(topic);
msgExt.setTopic(new String(topic, CHARSET_UTF8));
// 17 properties
short propertiesLength = Short();
if (propertiesLength > 0) {
byte[] properties = new byte[propertiesLength];
<(properties);
String propertiesString = new String(properties, CHARSET_UTF8);
Map<String, String> map = string2messageProperties(propertiesString);
}
int msgIDLength = storehostIPLength + 4 + 8;
ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
String msgId = createMessageId(byteBufferMsgId, StoreHostBytes(), CommitLogOffset());
msgExt.setMsgId(msgId);
return msgExt;
}
public static void main(String[] args) throws Exception {
String filePath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000";
ByteBuffer buffer = read(filePath);
List<MessageExt> messageList = new ArrayList<>();
while (true){
MessageExt message = decodeCommitLog(buffer);
if (message==null){
break;
}
messageList.add(message);
}
for (MessageExt ms:messageList) {
System.out.println("主题:"+ms.getTopic()+" 消息:"+
new Body())+"队列ID:"+ms.getQueueId()+" 存储地址:"+ms.getStoreHost());
}
}
运⾏这段代码，我们就可以直接看到CommitLog⽂件中的内容：
主题:topic0 消息:RocketMQ消息存储和查询实战队列ID:1存储地址:/192.168.44.1:10911
主题:topic1 消息:RocketMQ消息存储和查询实战队列ID:0存储地址:/192.168.44.1:10911
主题:topic2 消息:RocketMQ消息存储和查询实战队列ID:1存储地址:/192.168.44.1:10911
主题:topic3 消息:RocketMQ消息存储和查询实战队列ID:0存储地址:/192.168.44.1:10911
主题:topic4 消息:RocketMQ消息存储和查询实战队列ID:3存储地址:/192.168.44.1:10911
主题:topic5 消息:RocketMQ消息存储和查询实战队列ID:1存储地址:/192.168.44.1:10911
主题:topic6 消息:RocketMQ消息存储和查询实战队列ID:2存储地址:/192.168.44.1:10911
主题:topic7 消息:RocketMQ消息存储和查询实战队列ID:3存储地址:/192.168.44.1:10911
主题:topic8 消息:RocketMQ消息存储和查询实战队列ID:2存储地址:/192.168.44.1:10911
主题:topic9 消息:RocketMQ消息存储和查询实战队列ID:0存储地址:/192.168.44.1:10911
不⽤过多的⽂字描述，通过上⾯这些代码，相信你对CommitLog⽂件就有了更进⼀步的了解。
此时，我们再考虑另外⼀个问题：
CommitLog ⽂件保存了所有主题的消息，但我们消费时，更多的是订阅某⼀个主题进⾏消费。RocketMQ是怎么样进⾏⾼效的检索消息的呢？
⼆、ConsumeQueue
为了解决上⾯那个问题，RocketMQ引⼊了ConsumeQueue消费队列⽂件。
在继续往下说ConsumeQueue之前，我们必须先了解到另外⼀个概念，即MessageQueue。
我们知道，在发送消息的时候，要指定⼀个Topic。那么，在创建Topic的时候，有⼀个很重要的参数MessageQueue。简单来说，就是你这个Topic对应了多少个队列，也就
是⼏个MessageQueue，默认是4个。那它的作⽤是什么呢？
它是⼀个数据分⽚的机制。⽐如我们的Topic⾥⾯有100条数据，该Topic默认是4个队列，那么每个队列中⼤约25条数据。然后，这些MessageQueue是和Broker绑定在⼀起的，就是说每个MessageQueue都可能处于不同的Broker机器上，这取决于你的队列数量和Broker集。
我们来看上⾯的图⽚，Topic名称为order的主题，⼀共有4个MessageQueue，每个⾥⾯都有25条数据。因为在笔者的本地环境只有⼀个Broker，所以它们的brokerName都是指向同⼀台机器。
既然MessageQueue是多个，那么在消息发送的时候，势必要通过某种⽅式选择⼀个队列。默认的情况下，就是通过轮询来获取⼀个消息队列。
public MessageQueue selectOneMessageQueue() {
int index = AndIncrement();
int pos = Math.abs(index) % ssageQueueList.size();
if (pos < 0)
pos = 0;
(pos);
}
当然，RocketMQ还有⼀个故障延迟机制，在选择消息队列的时候会复杂⼀些，我们今天先不讨论。
说完了MessageQueue，我们接着来看ConsumerQueue。上⾯我们说，它是为了⾼效检索主题消息的。
ConsumerQueue也是⼀组组⽂件，它的位置在C:/Users/shiqizhen/store/consumequeue。该⽬录下⾯是以Topic命名的⽂件夹，然后再下⼀级是以MessageQueue队列ID命名的⽂件夹，最后才是⼀个或多个⽂件。
这样分层之后，RocketMQ⾄少可以得到以下⼏个讯息：
先通过主题名称，可以定位到具体的⽂件夹；
然后根据消息队列ID到具体的⽂件；
最后根据⽂件内容，到具体的消息。
那么，这个⽂件⾥⾯存储的⼜是什么内容呢？
为了加速ConsumerQueue的检索速度和节省磁盘空间，⽂件中不会存储消息的全量消息。其存储的格式如下：
同样的，我们先写⼀段代码，按照这个格式输出⼀下ConsumerQueue⽂件的内容。
public static void main(String[] args)throws Exception {
String path = "C:\\Users\\shiqizhen\\store\\consumequeue\\order\\0\\00000000000000000000";
ByteBuffer buffer = read(path);
while (true){
long offset = Long();
long size = Int();
long code = Long();
if (size==0){
break;
}
System.out.println("消息长度:"+size+" 消息偏移量:" +offset);
}
System.out.println("--------------------------");
}
在前⾯，我们已经向order这个主题中写了100条数据，所以在这⾥它的order#messagequeue#0⾥⾯有25条记录。
消息长度:173消息偏移量:2003
消息长度:173消息偏移量:2695
消息长度:173消息偏移量:3387
消息长度:173消息偏移量:4079
消息长度:173消息偏移量:4771
消息长度:173消息偏移量:5463
消息长度:173消息偏移量:6155
消息长度:173消息偏移量:6847
消息长度:173消息偏移量:7539
消息长度:173消息偏移量:8231
消息长度:173消息偏移量:8923
消息长度:173消息偏移量:9615
消息长度:173消息偏移量:10307
消息长度:173消息偏移量:10999
消息长度:173消息偏移量:11691
消息长度:173消息偏移量:12383
消息长度:173消息偏移量:13075
消息长度:173消息偏移量:13767
消息长度:173消息偏移量:14459
消息长度:173消息偏移量:15151
消息长度:173消息偏移量:15843
消息长度:173消息偏移量:16535
消息长度:173消息偏移量:17227
消息长度:173消息偏移量:17919
消息长度:173消息偏移量:18611
--------------------------
细⼼的朋友，肯定发现了。上⾯输出的结果中，消息偏移量的差值等于 = 消息长度 * 队列长度。
现在我们通过ConsumerQueue已经知道了消息的长度和偏移量，那么查消息就⽐较容易了。
public static MessageExt getMessageByOffset(ByteBuffer commitLog,long offset,int size) throws Exception {    ByteBuffer slice = commitLog.slice();
slice.position((int)offset);
slice.limit((int) (offset+size));
MessageExt message = CommitLogTest.decodeCommitLog(slice);
return message;
}
然后，我们可以依靠这种⽅法，来实现通过ConsumerQueue获取消息的具体内容。