RocketMQ中ConsumeQueue文件與Index文件是怎么樣的

這篇文章主要為大家展示了“RocketMQ中ConsumeQueue文件與Index文件是怎么樣的”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“RocketMQ中ConsumeQueue文件與Index文件是怎么樣的”這篇文章吧。

一、問題思考

消息消費時先從ConsumeQueue中獲取物理偏移量，再根據物理偏移量從commitLog中獲取具體消息；消息檢索時會用到索引文件，其中值得思考的問題：
1.ConsumeQueue構建流程是怎樣的？
2.ConsumeQueue數據結構是怎樣的？
3.Index索引文件構建流程怎樣的？
4.Index數據結構時怎么樣的？

二、ConsumeQueue/Index構建概覽1.調用鏈條

//Broker啟動初始化
@1 BrokerStartup#main
start(createBrokerController(args))
boolean initResult = controller.initialize()
@2 BrokerController#initialize
this.messageStore = new DefaultMessageStore
@3 DefaultMessageStore#DefaultMessageStore()
this.reputMessageService = new ReputMessageService();
this.dispatcherList = new LinkedList<>();
this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
//存儲服務啟動
@4 DefaultMessageStore#start()
//允許重復轉發reputFromOffset設置為CommitLog的提交指針
if (this.getMessageStoreConfig().isDuplicationEnable()) {this.reputMessageService.setReputFromOffset(this.commitLog.getConfirmOffset());
} else {
//不允許重復轉發reputFromOffset設置為CommitLog內存中最大偏移量 this.reputMessageService.setReputFromOffset(this.commitLog.getMaxOffset();
}
this.reputMessageService.start();

小結：@1中分別為dispatcherList添加了CommitLogDispatcherBuildConsumeQueue
和CommitLogDispatcherBuildIndex；
@4中duplicationEnable默認為false即不允許重復，從CommitLog中的最大偏移量開始轉發，reputMessageService線程類在Broker啟動時啟動，主要負責構建consumeQueue與index文件。

2.ReputMessageService線程類職責

小結：ReputMessageServicee根據構建進度reputFromOffset查找可構建的消息數據，然后逐條解析組成構建請求，并構建consumeQueue和index文件構建；如果當前Broker為Master并且長輪詢模式上通過消息到達監聽器通知客戶端。

三、ConsumeQueue構建流程及數據結構

在Broker啟動時初始化了dispatcherList, 添加了分別負責ConsumeQueue文件和Index文件構建類；這部分關注ConsumeQueue構建。

this.dispatcherList = new LinkedList<>();
//構建ConsumeQueue
this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
//構建Index
this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
public void doDispatch(DispatchRequest req) {
for (CommitLogDispatcher dispatcher : this.dispatcherList) {
dispatcher.dispatch(req);
}
}

1.ConsumeQueue文件構建流程調用鏈

@1 DefaultMessageStore#putMessagePositionInfo
@2 ConsumeQueue#putMessagePositionInfoWrapper

流程圖

小結：ConsumeQueue構建主要流程為構建ConsumeQueue數據結構并將其寫入fileChannel落盤；第一次創建ConsumeQueue文件時進行補位，用0填充，可促使系統實際分配內存起到預熱作用。

補位代碼

private void fillPreBlank(final MappedFile mappedFile, final long untilWhere) {
ByteBuffer byteBuffer = ByteBuffer.allocate(CQ_STORE_UNIT_SIZE);
byteBuffer.putLong(0L);
byteBuffer.putInt(Integer.MAX_VALUE);
byteBuffer.putLong(0L);
int until = (int) (untilWhere % this.mappedFileQueue.getMappedFileSize());
for (int i = 0; i < until; i += CQ_STORE_UNIT_SIZE) {
mappedFile.appendMessage(byteBuffer.array());
}
}

2.ConsumeQueue數據結構數據結構代碼

this.byteBufferIndex.flip();
this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE); //限定每個條目大小
this.byteBufferIndex.putLong(offset); //寫入消息偏移量
this.byteBufferIndex.putInt(size); //寫入消息長度
this.byteBufferIndex.putLong(tagsCode); //寫入tag hashcode

數據結構圖示

ConsumeQueue文件中每個條目占20位。

四、Index構建流程及數據結構

在Broker啟動時初始化了dispatcherList, 添加了分別負責ConsumeQueue文件和Index文件構建類；這部分關注Index構建。
IndexService初始化時初始化兩個參數Hash槽數量hashSlotNum=5000000，索引的最大數量maxIndexNum=5000000 * 4=20000000。

this.dispatcherList = new LinkedList<>();
//構建ConsumeQueue
this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
//構建Index
this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
public void doDispatch(DispatchRequest req) {
for (CommitLogDispatcher dispatcher : this.dispatcherList) {
dispatcher.dispatch(req);
}
}
//IndexService初始化
public IndexService(final DefaultMessageStore store) {
this.defaultMessageStore = store;
this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum();
this.indexNum = store.getMessageStoreConfig().getMaxIndexNum();
this.storePath =
StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir());
}

1.Index文件構建流程調用鏈條

@1 DefaultMessageStore#CommitLogDispatcherBuildIndex#dispatch
@2 IdexService#buildIndex

流程圖

小結：Index文件寫入流程概要：先獲取索引文件并將未寫入文件的內存數據通過守護線程寫入磁盤；計算要寫入索引所在hash槽的位置取出原來的值；構建索引條目填充數據；最后更新索引文件頭部信息。

2.Index數據結構寫入索引數據到mappedByteBuffer代碼

/計算索引數據需要放在哪個位置
int absIndexPos =
IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
+ this.indexHeader.getIndexCount() * indexSize;
//將hashcode存儲在MappedByteBuffer中
this.mappedByteBuffer.putInt(absIndexPos, keyHash);
//將物理偏移量存儲在MappedByteBuffer中
this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
//落地時間-當前索引的起始時間差值寫入MappedByteBuffer
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
//記錄前一條hash桶對應的值（Index條目下標）；注意此處用于解決Hash沖突
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
//將當前index中包含的條目數量存入到Hash槽中，將覆蓋原先的值
this.mappedByteBuffer.putInt(absSlotPos,this.indexHeader.getIndexCount());

小結：上述代碼描述索引條目的構建流程及數據結構，通過記錄上一條沖突的槽值形成鏈表結構。

Index數據結構圖示

Index索引有三部分組成IndexHeader、Hash槽位、索引條目；每塊內容和所占大小如圖所示；如果有Hash沖突，在每個索引條目最后記錄了原來Hash槽里的值，從而形成鏈表結構。

以上是“RocketMQ中ConsumeQueue文件與Index文件是怎么樣的”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！