Java的RocketMQ之消息存儲和查詢原理詳解
一、CommitLog
CommitLog,消息存儲文件,所有主題的消息都存儲在 CommitLog 文件中。
我們的業(yè)務系統(tǒng)向 RocketMQ 發(fā)送一條消息,不管在中間經(jīng)歷了多么復雜的流程,最終這條消息會被持久化到CommitLog文件。
我們知道,一臺Broker服務器只有一個CommitLog文件(組),RocketMQ會將所有主題的消息存儲在同一個文件中,這個文件中就存儲著一條條Message,每條Message都會按照順序?qū)懭搿?/p>

也許有時候,你會希望看看這個 CommitLog 文件中,存儲的內(nèi)容到底長什么樣子?
1、消息發(fā)送
當然,我們需要先往 CommitLog 文件中寫入一些內(nèi)容,所以先來看一個消息發(fā)送的例子。
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個不同的主題中發(fā)送消息,如果只有一臺Broker機器,它們會保存到同一個CommitLog文件中。此時,這個文件的位置處于 C:/Users/shiqizhen/store/commitlog/00000000000000000000。
2、讀取文件內(nèi)容
這個文件我們不能直接打開,因為它是一個二進制文件,所以我們需要通過程序來讀取它的字節(jié)數(shù)組。
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()];
fin.read(bytes);
ByteBuffer buffer = ByteBuffer.wrap(bytes);
return buffer;
}如上代碼,可以通過傳入文件的路徑,讀取該文件所有的內(nèi)容。為了方便下一步操作,我們把讀取到的字節(jié)數(shù)組轉(zhuǎn)換為java.nio.ByteBuffer對象。
3、解析
在解析之前,我們需要弄明白兩件事:
- 消息的格式,即一條消息包含哪些字段;
- 每個字段所占的字節(jié)大小。
在上面的圖中,我們已經(jīng)看到了消息的格式,包含了19個字段。關于字節(jié)大小,有的是 4 字節(jié),有的是 8 字節(jié),我們不再一一贅述,直接看代碼。
/**
* commitlog 文件解析
* @param byteBuffer
* @return
* @throws Exception
*/
public static MessageExt decodeCommitLog(ByteBuffer byteBuffer)throws Exception {
MessageExt msgExt = new MessageExt();
// 1 TOTALSIZE
int storeSize = byteBuffer.getInt();
msgExt.setStoreSize(storeSize);
if (storeSize<=0){
return null;
}
// 2 MAGICCODE
byteBuffer.getInt();
// 3 BODYCRC
int bodyCRC = byteBuffer.getInt();
msgExt.setBodyCRC(bodyCRC);
// 4 QUEUEID
int queueId = byteBuffer.getInt();
msgExt.setQueueId(queueId);
// 5 FLAG
int flag = byteBuffer.getInt();
msgExt.setFlag(flag);
// 6 QUEUEOFFSET
long queueOffset = byteBuffer.getLong();
msgExt.setQueueOffset(queueOffset);
// 7 PHYSICALOFFSET
long physicOffset = byteBuffer.getLong();
msgExt.setCommitLogOffset(physicOffset);
// 8 SYSFLAG
int sysFlag = byteBuffer.getInt();
msgExt.setSysFlag(sysFlag);
// 9 BORNTIMESTAMP
long bornTimeStamp = byteBuffer.getLong();
msgExt.setBornTimestamp(bornTimeStamp);
// 10 BORNHOST
int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16;
byte[] bornHost = new byte[bornhostIPLength];
byteBuffer.get(bornHost, 0, bornhostIPLength);
int port = byteBuffer.getInt();
msgExt.setBornHost(new InetSocketAddress(InetAddress.getByAddress(bornHost), port));
// 11 STORETIMESTAMP
long storeTimestamp = byteBuffer.getLong();
msgExt.setStoreTimestamp(storeTimestamp);
// 12 STOREHOST
int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16;
byte[] storeHost = new byte[storehostIPLength];
byteBuffer.get(storeHost, 0, storehostIPLength);
port = byteBuffer.getInt();
msgExt.setStoreHost(new InetSocketAddress(InetAddress.getByAddress(storeHost), port));
// 13 RECONSUMETIMES
int reconsumeTimes = byteBuffer.getInt();
msgExt.setReconsumeTimes(reconsumeTimes);
// 14 Prepared Transaction Offset
long preparedTransactionOffset = byteBuffer.getLong();
msgExt.setPreparedTransactionOffset(preparedTransactionOffset);
// 15 BODY
int bodyLen = byteBuffer.getInt();
if (bodyLen > 0) {
byte[] body = new byte[bodyLen];
byteBuffer.get(body);
msgExt.setBody(body);
}
// 16 TOPIC
byte topicLen = byteBuffer.get();
byte[] topic = new byte[(int) topicLen];
byteBuffer.get(topic);
msgExt.setTopic(new String(topic, CHARSET_UTF8));
// 17 properties
short propertiesLength = byteBuffer.getShort();
if (propertiesLength > 0) {
byte[] properties = new byte[propertiesLength];
byteBuffer.get(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, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset());
msgExt.setMsgId(msgId);
return msgExt;
}4、輸出消息內(nèi)容
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 String(ms.getBody())+"隊列ID:"+ms.getQueueId()+" 存儲地址:"+ms.getStoreHost());
}
}運行這段代碼,我們就可以直接看到CommitLog文件中的內(nèi)容:
主題:topic0 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic1 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
主題:topic2 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic3 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
主題:topic4 消息:清幽之地的博客 隊列ID:3 存儲地址:/192.168.44.1:10911
主題:topic5 消息:清幽之地的博客 隊列ID:1 存儲地址:/192.168.44.1:10911
主題:topic6 消息:清幽之地的博客 隊列ID:2 存儲地址:/192.168.44.1:10911
主題:topic7 消息:清幽之地的博客 隊列ID:3 存儲地址:/192.168.44.1:10911
主題:topic8 消息:清幽之地的博客 隊列ID:2 存儲地址:/192.168.44.1:10911
主題:topic9 消息:清幽之地的博客 隊列ID:0 存儲地址:/192.168.44.1:10911
不用過多的文字描述,通過上面這些代碼,相信你對CommitLog文件就有了更進一步的了解。
此時,我們再考慮另外一個問題:
CommitLog 文件保存了所有主題的消息,但我們消費時,更多的是訂閱某一個主題進行消費。RocketMQ是怎么樣進行高效的檢索消息的呢 ?
二、ConsumeQueue
為了解決上面那個問題,RocketMQ引入了ConsumeQueue消費隊列文件。
在繼續(xù)往下說ConsumeQueue之前,我們必須先了解到另外一個概念,即MessageQueue。
1、MessageQueue
我們知道,在發(fā)送消息的時候,要指定一個Topic。那么,在創(chuàng)建Topic的時候,有一個很重要的參數(shù)MessageQueue。簡單來說,就是你這個Topic對應了多少個隊列,也就是幾個MessageQueue,默認是4個。那它的作用是什么呢 ?
它是一個數(shù)據(jù)分片的機制。比如我們的Topic里面有100條數(shù)據(jù),該Topic默認是4個隊列,那么每個隊列中大約25條數(shù)據(jù)。 然后,這些MessageQueue是和Broker綁定在一起的,就是說每個MessageQueue都可能處于不同的Broker機器上,這取決于你的隊列數(shù)量和Broker集群。

我們來看上面的圖片,Topic名稱為order的主題,一共有4個MessageQueue,每個里面都有25條數(shù)據(jù)。因為在筆者的本地環(huán)境只有一個Broker,所以它們的brokerName都是指向同一臺機器。
既然MessageQueue是多個,那么在消息發(fā)送的時候,勢必要通過某種方式選擇一個隊列。默認的情況下,就是通過輪詢來獲取一個消息隊列。
public MessageQueue selectOneMessageQueue() {
int index = this.sendWhichQueue.getAndIncrement();
int pos = Math.abs(index) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
return this.messageQueueList.get(pos);
}當然,RocketMQ還有一個故障延遲機制,在選擇消息隊列的時候會復雜一些,我們今天先不討論。
2、ConsumeQueue
說完了MessageQueue,我們接著來看ConsumerQueue。上面我們說,它是為了高效檢索主題消息的。
ConsumerQueue也是一組組文件,它的位置在C:/Users/shiqizhen/store/consumequeue。該目錄下面是以Topic命名的文件夾,然后再下一級是以MessageQueue隊列ID命名的文件夾,最后才是一個或多個文件。

這樣分層之后,RocketMQ至少可以得到以下幾個訊息:
- 先通過主題名稱,可以定位到具體的文件夾;
- 然后根據(jù)消息隊列ID找到具體的文件;
- 最后根據(jù)文件內(nèi)容,找到具體的消息。
那么,這個文件里面存儲的又是什么內(nèi)容呢 ?
3、解析文件
為了加速ConsumerQueue的檢索速度和節(jié)省磁盤空間,文件中不會存儲消息的全量消息。其存儲的格式如下:

同樣的,我們先寫一段代碼,按照這個格式輸出一下ConsumerQueue文件的內(nèi)容。
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 = buffer.getLong();
long size = buffer.getInt();
long code = buffer.getLong();
if (size==0){
break;
}
System.out.println("消息長度:"+size+" 消息偏移量:" +offset);
}
System.out.println("--------------------------");
}在前面,我們已經(jīng)向order這個主題中寫了100條數(shù)據(jù),所以在這里它的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
細心的朋友,肯定發(fā)現(xiàn)了。上面輸出的結果中,消息偏移量的差值等于 = 消息長度 * 隊列長度。
4、查詢消息
現(xiàn)在我們通過ConsumerQueue已經(jīng)知道了消息的長度和偏移量,那么查找消息就比較容易了。
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;
}然后,我們可以依靠這種方法,來實現(xiàn)通過ConsumerQueue獲取消息的具體內(nèi)容。
public static void main(String[] args) throws Exception {
//consumerqueue根目錄
String consumerPath = "C:\\Users\\shiqizhen\\store\\consumequeue";
//commitlog目錄
String commitLogPath = "C:\\Users\\shiqizhen\\store\\commitlog\\00000000000000000000";
//讀取commitlog文件內(nèi)容
ByteBuffer commitLogBuffer = CommitLogTest.read(commitLogPath);
//遍歷consumerqueue目錄下的所有文件
File file = new File(consumerPath);
File[] files = file.listFiles();
for (File f:files) {
if (f.isDirectory()){
File[] listFiles = f.listFiles();
for (File queuePath:listFiles) {
String path = queuePath+"/00000000000000000000";
//讀取consumerqueue文件內(nèi)容
ByteBuffer buffer = CommitLogTest.read(path);
while (true){
//讀取消息偏移量和消息長度
long offset = (int) buffer.getLong();
int size = buffer.getInt();
long code = buffer.getLong();
if (size==0){
break;
}
//根據(jù)偏移量和消息長度,在commitloh文件中讀取消息內(nèi)容
MessageExt message = getMessageByOffset(commitLogBuffer,offset,size);
if (message!=null){
System.out.println("消息主題:"+message.getTopic()+" MessageQueue:"+
message.getQueueId()+" 消息體:"+new String(message.getBody()));
}
}
}
}
}
}運行這段代碼,就可以得到之前測試樣例中,10個主題的所有消息。
消息主題:topic0 MessageQueue:1 消息體:清幽之地的博客
消息主題:topic1 MessageQueue:0 消息體:清幽之地的博客
消息主題:topic2 MessageQueue:1 消息體:清幽之地的博客
消息主題:topic3 MessageQueue:0 消息體:清幽之地的博客
消息主題:topic4 MessageQueue:3 消息體:清幽之地的博客
消息主題:topic5 MessageQueue:1 消息體:清幽之地的博客
消息主題:topic6 MessageQueue:2 消息體:清幽之地的博客
消息主題:topic7 MessageQueue:3 消息體:清幽之地的博客
消息主題:topic8 MessageQueue:2 消息體:清幽之地的博客
消息主題:topic9 MessageQueue:0 消息體:清幽之地的博客
5、消費消息
消息消費的時候,其查找消息的過程也是差不多的。不過值得注意的一點是,ConsumerQueue文件和CommitLog文件可能都是多個的,所以會有一個定位文件的過程,我們來看源碼。
首先,根據(jù)消費進度來查找對應的ConsumerQueue,獲取其文件內(nèi)容。
public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
//ConsumerQueue文件大小
int mappedFileSize = this.mappedFileSize;
//根據(jù)消費進度,找到在consumerqueue文件里的偏移量
long offset = startIndex * CQ_STORE_UNIT_SIZE;
if (offset >= this.getMinLogicOffset()) {
//返回ConsumerQueue映射文件
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
if (mappedFile != null) {
//返回文件里的某一塊內(nèi)容
SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
return result;
}
}
return null;
}然后拿到消息在CommitLog文件中的偏移量和消息長度,獲取消息。
public SelectMappedBufferResult getMessage(final long offset, final int size) {
//commitlog文件大小
int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
//根據(jù)消息偏移量,定位到具體的commitlog文件
MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
if (mappedFile != null) {
//根據(jù)消息偏移量和長度,獲取消息內(nèi)容
int pos = (int) (offset % mappedFileSize);
return mappedFile.selectMappedBuffer(pos, size);
}
return null;
}6、通過 Message Id 查詢
上面我們看到了通過消息偏移量來查找消息的方式,但RocketMQ還提供了其他幾種方式可以查詢消息。
- 通過Message Key 查詢;
- 通過Unique Key查詢;
- 通過Message Id查詢。
在這里,Message Key和Unique Key都是在消息發(fā)送之前,由客戶端生成的。我們可以自己設置,也可以由客戶端自動生成,Message Id是在Broker端存儲消息的時候生成。
Message Id總共 16 字節(jié),包含消息存儲主機地址和在CommitLog文件中的偏移量offset。有源碼為證:
/**
* 創(chuàng)建消息ID
* @param input
* @param addr Broker服務器地址
* @param offset 正在存儲的消息,在Commitlog中的偏移量
* @return
*/
public static String createMessageId(final ByteBuffer input, final ByteBuffer addr, final long offset) {
input.flip();
int msgIDLength = addr.limit() == 8 ? 16 : 28;
input.limit(msgIDLength);
input.put(addr);
input.putLong(offset);
return UtilAll.bytes2string(input.array());
}當我們根據(jù)Message Id向Broker查詢消息時,首先會通過一個decodeMessageId方法,將Broker地址和消息的偏移量解析出來。
public static MessageId decodeMessageId(final String msgId) throws Exception {
SocketAddress address;
long offset;
int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2;
byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength));
byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
ByteBuffer bb = ByteBuffer.wrap(port);
int portInt = bb.getInt(0);
//解析出來Broker地址
address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);
//偏移量
byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
bb = ByteBuffer.wrap(data);
offset = bb.getLong(0);
return new MessageId(address, offset);
}所以通過Message Id查詢消息的時候,實際上還是直接從特定Broker上的CommitLog指定位置進行查詢,屬于精確查詢。
這個也沒問題,但是如果通過 Message Key 和 Unique Key 查詢的時候,RocketMQ 又是怎么做的呢?
三、Index
1、index索引文件
ConsumerQueue消息消費隊列是專門為消息訂閱構建的索引文件,提高根據(jù)主題與消息隊列檢索消息的速度。
另外,RocketMQ引入Hash索引機制,為消息建立索引,它的鍵就是Message Key 和 Unique Key。
那么,我們先看看index索引文件的結構:

為了便于理解,我們還是以代碼的方式,來解析這個文件。
public static void main(String[] args) throws Exception {
//index索引文件的路徑
String path = "C:\\Users\\shiqizhen\\store\\index\\20200506224547616";
ByteBuffer buffer = CommitLogTest.read(path);
//該索引文件中包含消息的最小存儲時間
long beginTimestamp = buffer.getLong();
//該索引文件中包含消息的最大存儲時間
long endTimestamp = buffer.getLong();
//該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
long beginPhyOffset = buffer.getLong();
//該索引文件中包含消息的最大物理偏移量(commitlog文件偏移量)
long endPhyOffset = buffer.getLong();
//hashslot個數(shù)
int hashSlotCount = buffer.getInt();
//Index條目列表當前已使用的個數(shù)
int indexCount = buffer.getInt();
//500萬個hash槽,每個槽占4個字節(jié),存儲的是index索引
for (int i=0;i<5000000;i++){
buffer.getInt();
}
//2000萬個index條目
for (int j=0;j<20000000;j++){
//消息key的hashcode
int hashcode = buffer.getInt();
//消息對應的偏移量
long offset = buffer.getLong();
//消息存儲時間和第一條消息的差值
int timedif = buffer.getInt();
//該條目的上一條記錄的index索引
int pre_no = buffer.getInt();
}
System.out.println(buffer.position()==buffer.capacity());
}我們看最后輸出的結果為true,則證明解析的過程無誤。
2、構建索引
我們發(fā)送的消息體中,包含Message Key 或 Unique Key,那么就會給它們每一個都構建索引。
這里重點有兩個:
- 根據(jù)消息Key計算Hash槽的位置;
- 根據(jù)Hash槽的數(shù)量和Index索引來計算Index條目的起始位置。
將當前 Index條目 的索引值,寫在Hash槽absSlotPos位置上;將Index條目的具體信息(hashcode/消息偏移量/時間差值/hash槽的值),從起始偏移量absIndexPos開始,順序按字節(jié)寫入。
public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
if (this.indexHeader.getIndexCount() < this.indexNum) {
//計算key的hash
int keyHash = indexKeyHashMethod(key);
//計算hash槽的坐標
int slotPos = keyHash % this.hashSlotNum;
int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
//計算時間差值
long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
timeDiff = timeDiff / 1000;
//計算INDEX條目的起始偏移量
int absIndexPos =
IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
+ this.indexHeader.getIndexCount() * indexSize;
//依次寫入hashcode、消息偏移量、時間戳、hash槽的值
this.mappedByteBuffer.putInt(absIndexPos, keyHash);
this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
//將當前INDEX中包含的條目數(shù)量寫入HASH槽
this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
return true;
}
return false;
}這樣構建完Index索引之后,根據(jù)Message Key 或 Unique Key查詢消息就簡單了。
比如我們通過RocketMQ客戶端工具,根據(jù)Unique Key來查詢消息。
adminImpl.queryMessageByUniqKey("order", "FD88E3AB24F6980059FDC9C3620464741BCC18B4AAC220FDFE890007");在Broker端,通過Unique Key來計算Hash槽的位置,從而找到Index索引數(shù)據(jù)。從Index索引中拿到消息的物理偏移量,最后根據(jù)消息物理偏移量,直接到CommitLog文件中去找就可以了。
到此這篇關于Java的RocketMQ之消息存儲和查詢原理詳解的文章就介紹到這了,更多相關RocketMQ消息存儲和查詢原理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
Java SpringBoot啟動指定profile的8種方式詳解
這篇文章主要介紹了spring boot 如何指定profile啟動的操作,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-09-09
java自帶的MessageDigest實現(xiàn)文本的md5加密算法
這篇文章主要介紹了java自帶的MessageDigest實現(xiàn)文本的md5加密算法,需要的朋友可以參考下2015-12-12
Elasticsearch 映射 fielddata 工作原理解析
在 Elasticsearch 中,fielddata 是一種在內(nèi)存中構建的索引,用于加速某些類型的查詢,特別是聚合和排序操作,下面通過本文給大家介紹Elasticsearch 映射 fielddata的相關知識,感興趣的朋友一起看看吧2025-06-06

