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

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

一、問題思考

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

二、ConsumeQueue/Index構(gòu)建概覽1.調(diào)用鏈條

//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());
//存儲服務(wù)啟動
@4 DefaultMessageStore#start()
//允許重復(fù)轉(zhuǎn)發(fā)reputFromOffset設(shè)置為CommitLog的提交指針
if (this.getMessageStoreConfig().isDuplicationEnable()) {this.reputMessageService.setReputFromOffset(this.commitLog.getConfirmOffset());
} else {
//不允許重復(fù)轉(zhuǎn)發(fā)reputFromOffset設(shè)置為CommitLog內(nèi)存中最大偏移量 this.reputMessageService.setReputFromOffset(this.commitLog.getMaxOffset();
}
this.reputMessageService.start();

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

2.ReputMessageService線程類職責

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

三、ConsumeQueue構(gòu)建流程及數(shù)據(jù)結(jié)構(gòu)

在Broker啟動時初始化了dispatcherList, 添加了分別負責ConsumeQueue文件和Index文件構(gòu)建類；這部分關(guān)注ConsumeQueue構(gòu)建。

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

1.ConsumeQueue文件構(gòu)建流程調(diào)用鏈

@1 DefaultMessageStore#putMessagePositionInfo
@2 ConsumeQueue#putMessagePositionInfoWrapper

流程圖

小結(jié)：ConsumeQueue構(gòu)建主要流程為構(gòu)建ConsumeQueue數(shù)據(jù)結(jié)構(gòu)并將其寫入fileChannel落盤；第一次創(chuàng)建ConsumeQueue文件時進行補位，用0填充，可促使系統(tǒng)實際分配內(nèi)存起到預(yù)熱作用。

補位代碼

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數(shù)據(jù)結(jié)構(gòu)數(shù)據(jù)結(jié)構(gòu)代碼

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

數(shù)據(jù)結(jié)構(gòu)圖示

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

四、Index構(gòu)建流程及數(shù)據(jù)結(jié)構(gòu)

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

this.dispatcherList = new LinkedList<>();
//構(gòu)建ConsumeQueue
this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
//構(gòu)建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文件構(gòu)建流程調(diào)用鏈條

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

流程圖

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

2.Index數(shù)據(jù)結(jié)構(gòu)寫入索引數(shù)據(jù)到mappedByteBuffer代碼

/計算索引數(shù)據(jù)需要放在哪個位置
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桶對應(yīng)的值（Index條目下標）；注意此處用于解決Hash沖突
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
//將當前index中包含的條目數(shù)量存入到Hash槽中，將覆蓋原先的值
this.mappedByteBuffer.putInt(absSlotPos,this.indexHeader.getIndexCount());

小結(jié)：上述代碼描述索引條目的構(gòu)建流程及數(shù)據(jù)結(jié)構(gòu)，通過記錄上一條沖突的槽值形成鏈表結(jié)構(gòu)。

Index數(shù)據(jù)結(jié)構(gòu)圖示

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

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