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

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

網(wǎng)站建設(shè)哪家好，找創(chuàng)新互聯(lián)！專注于網(wǎng)頁設(shè)計(jì)、網(wǎng)站建設(shè)、微信開發(fā)、小程序定制開發(fā)、集團(tuán)企業(yè)網(wǎng)站建設(shè)等服務(wù)項(xiàng)目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了中山免費(fèi)建站歡迎大家使用！

一、問題思考

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

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

//Broker啟動(dòng)初始化
@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());
//存儲(chǔ)服務(wù)啟動(dòng)
@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默認(rèn)為false即不允許重復(fù)，從CommitLog中的最大偏移量開始轉(zhuǎn)發(fā)，reputMessageService線程類在Broker啟動(dòng)時(shí)啟動(dòng)，主要負(fù)責(zé)構(gòu)建consumeQueue與index文件。

2.ReputMessageService線程類職責(zé)

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

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

在Broker啟動(dòng)時(shí)初始化了dispatcherList, 添加了分別負(fù)責(zé)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文件時(shí)進(jìn)行補(bǔ)位，用0填充，可促使系統(tǒng)實(shí)際分配內(nèi)存起到預(yù)熱作用。

補(bǔ)位代碼

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); //限定每個(gè)條目大小
this.byteBufferIndex.putLong(offset); //寫入消息偏移量
this.byteBufferIndex.putInt(size); //寫入消息長(zhǎng)度
this.byteBufferIndex.putLong(tagsCode); //寫入tag hashcode

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

ConsumeQueue文件中每個(gè)條目占20位。

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

在Broker啟動(dòng)時(shí)初始化了dispatcherList, 添加了分別負(fù)責(zé)ConsumeQueue文件和Index文件構(gòu)建類；這部分關(guān)注Index構(gòu)建。
IndexService初始化時(shí)初始化兩個(gè)參數(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ù)通過守護(hù)線程寫入磁盤；計(jì)算要寫入索引所在hash槽的位置取出原來的值；構(gòu)建索引條目填充數(shù)據(jù)；最后更新索引文件頭部信息。

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

/計(jì)算索引數(shù)據(jù)需要放在哪個(gè)位置
int absIndexPos =
IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
+ this.indexHeader.getIndexCount() * indexSize;
//將hashcode存儲(chǔ)在MappedByteBuffer中
this.mappedByteBuffer.putInt(absIndexPos, keyHash);
//將物理偏移量存儲(chǔ)在MappedByteBuffer中
this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
//落地時(shí)間-當(dāng)前索引的起始時(shí)間差值寫入MappedByteBuffer
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
//記錄前一條hash桶對(duì)應(yīng)的值（Index條目下標(biāo)）；注意此處用于解決Hash沖突
this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
//將當(dāng)前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沖突，在每個(gè)索引條目最后記錄了原來Hash槽里的值，從而形成鏈表結(jié)構(gòu)。

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

當(dāng)前標(biāo)題：RocketMQ中ConsumeQueue文件與Index文件是怎么樣的
網(wǎng)頁地址：http://muchs.cn/article2/iioeic.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供外貿(mào)建站、響應(yīng)式網(wǎng)站、用戶體驗(yàn)、企業(yè)建站、品牌網(wǎng)站設(shè)計(jì)、網(wǎng)站收錄

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)