通過這三個文件徹底搞懂Rocketmq的存儲原理
RocketMQ是阿里開發的一個高性能的消息隊列，支持各種消息類型，而且支持事務消息，可以說是現在的很多系統中的香餑餑了，所以呢，怎么使用大家肯定是要學習的。
我們作為一個有夢想的程序員，在學習一門技術的時候，肯定是不能光知其然，這是遠遠不夠的，我們必須要知其所以然，這樣才能在面試的時候侃侃而談，啊呸，不對，這樣我們才能在工作中遇到問題的時候，理性的去思考如何解決問題。
我們知道RocketMQ的架構是producer、NameServer、broker、Consumer，producer是生產消息的，NameServer是路由中心，負責服務的注冊發現以及路由管理這些。
Consumer是屬于消費消息的，broker則屬于真正的存儲消息，以及進行消息的持久化，也就是存儲消息的文件和索引消息的文件都在broker上。
消息隊列的主要作用是解耦異步削峰，也就意味著消息隊列中的存儲功能是必不可少的，而隨著時代的發展，業務量的增加也對消息隊列的存儲功能的強度的要求越來越高了。
也就是說你不能光性能好，你得存儲的消息也得足夠支撐我的業務量，你只能存儲100MB的消息，我這系統每分鐘的消息業務量可能500MB了，那肯定不夠使啊，那還削個啥的峰啊，峰來了你自己都頂不住。
RocketMQ憑借其強大的存儲能力和強大的消息索引能力，以及各種類型消息和消息的特性脫穎而出，于是乎，我們這些有夢想的程序員學習RocketMQ的存儲原理也變得尤為重要。
而要說起這個存儲原理，則不得不說的就是RocketMQ的消息存儲文件commitLog文件，消費方則是憑借著巧妙的設計Consumerqueue文件來進行高性能并且不混亂的消費，還有RocketMQ的強大的支持消息索引的特性，靠的就是indexfile索引文件。
我們這篇文章就從這commitLog、Consumerqueue、indexfile這三個神秘的文件說起，搞懂這三個文件，RocketMQ的核心就被你掏空了。
先上個圖，寫入commitLog文件時commitLog和Consumerqueue、indexfile文件三者的關系：
01 Commitlog文件
- 大小和命名規則
RocketMQ中的消息存儲文件放在${ROCKET_HOME}/store 目錄下，當生產者發送消息時，broker會將消息存儲到Commit文件夾下，文件夾下面會有一個commitLog文件，但是并不是意味著這個文件叫這個，文件命名是根據消息的偏移量來決定的。
文件有自己的生成規則，每個commitLog文件的大小是1G，一般情況下第一個 CommitLog 的起始偏移量為 0，第二個 CommitLog 的起始偏移量為 1073741824 (1G = 1073741824byte)。
也正是因為該文件的文件名字規則，所以也可以更好的知道消息處于哪個文件中，假設物理偏移量是1073741830，則相對的偏移量是6(6 = 1073741830 - 1073741824)，于是判斷出該消息位于第二個commitLog文件上，下面要說的Consumerqueue文件和indexfile文件都是通過偏移量來計算出消息位于哪個文件，進行更為精準的定位，減少了IO次數。
- 文件存儲規則和特點
commitLog文件的最大的一個特點就是消息的順序寫入，隨機讀寫，關于commitLog的文件的落盤有兩種，一種是同步刷盤，一種是異步刷盤，可通過 flushDiskType 進行配置。
在寫入commitLog的時候內部會有一個mappedFile內存映射文件，消息是先寫入到這個內存映射文件中，然后根據刷盤策略寫到硬盤中，對于producer的角度來說就是，同步就是當消息真正的寫到硬盤的時候才會給producer返回成功，而異步就是當消息到達內存的時候就返回成功了，然后異步的去刷盤。
跑題了，最大的特點順序寫入，所有的topic的消息都存儲到commitLog文件中，順序寫入可以充分的利用磁盤順序減少了IO爭用數據存儲的性能，kafka也是通過硬盤順序存盤的。
大家都常說硬盤的速度比內存慢，其實這句話也是有歧義的，當硬盤順序寫入和讀取的時候，速度不比內存慢，甚至比內存速度快，這種存儲方式就好比數組，我們如果知道數組的下標，則可以直接通過下標計算出位置，找到內存地址，眾所周知，數組的讀取是很快的，但是數組的缺點在于插入數據比較慢，因為如果在中間插入數據需要將后面的數據往后移動。
而對于數組來說，如果我們只會順序的往后添加，數組的速度也是很快的，因為數組沒有后續的數據的移動，這一操作很耗時。
回到RocketMQ中的commitLog文件，也是同樣的道理，順序的寫入文件也就不需要太多的去考慮寫入的位置，直接找到文件往后放就可以了，而取數據的時候，也是和數組一樣，我們可以通過文件的大小去精準的定位到哪一個文件，然后再精準的定位到文件的位置。
當然，至于這個索引位置就是靠下面的Consumerqueue文件和indexfile文件來找到消息的位置的，也就是索引地址。
哦對了，數組的元素大小是一樣的，并不意味這commitLog文件的各個消息存儲空間一樣。
- 簡單看下源碼
這部分源碼在DefaultMessageStore.putMessage。
- @Override
- public PutMessageResult putMessage(MessageExtBrokerInner msg) {
- if (this.shutdown) {
- log.warn("message store has shutdown, so putMessage is forbidden");
- return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
- }
- // 從節點不允許寫入
- if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
- long value = this.printTimes.getAndIncrement();
- if ((value % 50000) == 0) {
- log.warn("message store is slave mode, so putMessage is forbidden ");
- }
- return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
- }
- // store是否允許寫入
- if (!this.runningFlags.isWriteable()) {
- long value = this.printTimes.getAndIncrement();
- if ((value % 50000) == 0) {
- log.warn("message store is not writeable, so putMessage is forbidden " + this.runningFlags.getFlagBits());
- }
- return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
- } else {
- this.printTimes.set(0);
- }
- // topic過長
- if (msg.getTopic().length() > Byte.MAX_VALUE) {
- log.warn("putMessage message topic length too long " + msg.getTopic().length());
- return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
- }
- // 消息附加屬性過長
- if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
- log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
- return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
- }
- if (this.isOSPageCacheBusy()) {
- return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
- }
- long beginTime = this.getSystemClock().now();
- // 添加消息到commitLog
- PutMessageResult result = this.commitLog.putMessage(msg);
- long eclipseTime = this.getSystemClock().now() - beginTime;
- if (eclipseTime > 500) {
- log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}", eclipseTime, msg.getBody().length);
- }
- this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);
- if (null == result || !result.isOk()) {
- this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
- }
- return result;
- }
中間的commitLog.putMessage就是負責實現消息寫入commitLog文件，這個太長了，我就不給大家截了。
大致流程就是組裝消息，放入屬性，然后通過MappedFile對象寫入文件，緊接著根據刷盤策略刷盤，最后進行主從同步。
02 consumerQueue文件
RocketMQ是分為多個topic，消息所屬主題，屬于消息類型，每一個topic有多個queue，每個queue放著不同的消息，在同一個消費者組下的消費者，可以同時消費同一個topic下的不同queue隊列的消息。不同消費者下的消費者，可以同時消費同一個topic下的相同的隊列的消息。而同一個消費者組下的消費者，不可以同時消費不同topic下的消息。
而每個topic下的queue隊列都會對應一個Consumerqueue文件，例如Topic中有三個隊列，每個隊列中的消息索引都會有一個編號，編號從0開始，往上遞增。并由此一個位點offset的概念，有了這個概念，就可以對Consumer端的消費情況進行隊列定義。
消息消費完成后，需要將消費進度存儲起來，即前面提到的offset。廣播模式下，同消費組的消費者相互獨立，消費進度要單獨存儲;集群模式下，同一條消息只會被同一個消費組消費一次，消費進度會參與到負載均衡中，故消費進度是需要共享的。
消費進度，也就是由Broker管理每一個消費者消費Topic的進度，包含正常提交消費進度和重置消費進度，消費進度管理的目的是保證消費者在正常運行狀態、重啟、異常關閉等狀態下都能準確續接“上一次”未處理的消息。
在RocketMQ中，實現的消費語義叫“至少投遞一次”，也就是所有的消息至少有一次機會消費不用擔心會丟消息。用戶需要實現消費冪等來避免重復投遞對業務實際數據的影響。
冪等是啥應該不用我多說了吧，親愛的你們肯定知道了。
如上圖所示，消費者一般在兩種情況下“上報”消費進度，消費成功后(包含正常消費成功、重試消費成功)和重置消費進度。
而消費進度的標準就是Consumerqueue文件，這個文件中存儲的是投遞到某一個messagequeue中的位置信息。
比如我們知道消息存儲到commitLog文件中，一個消費者A對應著消費messagequeueA這個隊列，但是無法確定在commitLog文件中該隊列中的消息的位置，于是就有了ConsumerqueueA這個文件，這個文件對應一個messagequeueA，消費者A便可以通過ConsumerqueueA來確定自己的消費進度，獲取消息在commitLog文件中的具體的offset和大小。
- 存放位置和結構
consumequeue存放在store文件里面，里面的consumequeue文件里面按照topic排放，然后每個topic默認4個隊列，里面存放的consumequeue文件。
ConsumeQueue中并不需要存儲消息的內容，而存儲的是消息在CommitLog中的offset。也就是說ConsumeQueue其實是CommitLog的一個索引文件。
consumequeue是定長結構，每個記錄固定大小20個字節，單個consumequeue文件默認包含30w個條目，所以單個文件大小大概6M左右。
很顯然，Consumer消費消息的時候，要讀2次：先讀ConsumeQueue得到offset，再通過offset找到CommitLog對應的消息內容。
- ConsumeQueue的作用
消費者通過broker保存的offset(offsetTable.offset json文件中保存的ConsumerQueue的下標)可以在ConsumeQueue中獲取消息，從而快速的定位到commitLog的消息位置，由于每個消息的大小是不一樣的，也可以通過size獲取到消息的大小，從而讀取完整的消息。
過濾tag是也是通過遍歷ConsumeQueue來實現的(先比較hash(tag)符合條件的再到具體消息比較tag)。
- offsetTable.offset
和commitLog的offset不是一回事，這個offset是ConsumeQueue文件的(已經消費的)下標/行數，可以直接定位到ConsumeQueue并找到commitlogOffset從而找到消息體原文。這個offset是消息消費進度的核心，不同的消費模式，保存地址不同。
廣播模式：DefaultMQPushConsumer的BROADCASTING模式，各個Consumer沒有互相干擾，使用LoclaFileOffsetStore，把Offset存儲在Consumer本地。
集群模式：DefaultMQPushConsumer的CLUSTERING模式，由Broker端存儲和控制Offset的值，使用RemoteBrokerOffsetStore。
- 簡單看下構建過程
在Broker中，構建ComsummerQueue不是存儲完CommitLog就馬上同步構建的，而是通過一個線程任務異步的去做這個事情。在DefaultMessageStore中有一個ReputMessageService成員，它就是負責構建ComsumerQueue的任務線程。
ReputMessageService繼承自ServiceThread，表明其是一個服務線程，它的run方法很簡單，如下所示：
- public void run() {
- while (!this.isStopped()) {
- try {
- Thread.sleep(1);
- this.doReput(); // 構建ComsumerQueue
- } catch (Exception e) {
- DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
- }
- }
- }
在run方法里，每休息1毫秒就進行一次構建ComsumerQueue的動作。因為必須先寫入CommitLog，然后才能進行ComsumerQueue的構建。那么不排除構建ComsumerQueue的速度太快了，而CommitLog還沒寫入新的消息。這時就需要sleep下，讓出cpu時間片，避免浪費CPU資源。
我們點進去這個doReput()看核心處理邏輯：
- private void doReput() {
- for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {
- SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);// 拿到所有的最新寫入CommitLog的數據
- if (result != null) {
- try {
- this.reputFromOffset = result.getStartOffset();
- for (int readSize = 0; readSize < result.getSize() && doNext; ) {
- DispatchRequest dispatchRequest =
- DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false); // 一條一條的讀消息
- int size = dispatchRequest.getMsgSize();
- if (dispatchRequest.isSuccess()) {
- if (size > 0) {
- DefaultMessageStore.this.doDispatch(dispatchRequest); // 派發消息，進行處理，其中就包括構建ComsumerQueue
- this.reputFromOffset += size;
- readSize += size;
- } else if (size == 0) { //
- this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);
- readSize = result.getSize();
- }
- } else if (!dispatchRequest.isSuccess()) { // 獲取消息異常
- if (size > 0) {
- log.error("[BUG]read total count not equals msg total size. reputFromOffset={}", reputFromOffset);
- this.reputFromOffset += size;
- } else {
- doNext = false;
- if (DefaultMessageStore.this.brokerConfig.getBrokerId() == MixAll.MASTER_ID) {
- this.reputFromOffset += result.getSize() - readSize;
- }
- }
- }
- }
- } finally {
- result.release();
- }
- } else {
- doNext = false;
- }
- }
- }
我在這里省略了一些和構建ComsumerQueue不相干的代碼。
其實在doReput里面就做了三件事：
1、獲取最新寫入到CommitLog中的數據byteBuffer。
2、從byteBuffer中一條條的讀取消息，并派發出去處理。
3、更新reputFromOffset位移。
感興趣的可以打斷點走一遍。
03 indexFile文件
RocketMQ還支持通過MessageID或者MessageKey來查詢消息，使用ID查詢時，因為ID就是用broker+offset生成的(這里msgId指的是服務端的)，所以很容易就找到對應的commitLog文件來讀取消息。
對于用MessageKey來查詢消息，MessageStore通過構建一個index來提高讀取速度。
- 文件結構
indexfile文件存儲在store目錄下的index文件里面，里面存放的是消息的hashcode和index內容，文件由一個文件頭組成：長40字節。500w個hashslot，每個4字節。2000w個index條目，每個20字節。
所以這里我們可以估算每個indexfile的大小為：40+500w4+2000w20個字節，大約400M左右。
- 文件詳細信息
IndexHeader：索引文件頭信息由40個字節組成。
- //8位 該索引文件的第一個消息(Message)的存儲時間(落盤時間)
- this.byteBuffer.putLong(beginTimestampIndex, this.beginTimestamp.get());
- //8位 該索引文件的最后一個消息(Message)的存儲時間(落盤時間)
- this.byteBuffer.putLong(endTimestampIndex, this.endTimestamp.get());
- //8位 該索引文件第一個消息(Message)的在CommitLog(消息存儲文件)的物理位置偏移量(可以通過該物理偏移直接獲取到該消息)
- this.byteBuffer.putLong(beginPhyoffsetIndex, this.beginPhyOffset.get());
- //8位 該索引文件最后一個消息(Message)的在CommitLog(消息存儲文件)的物理位置偏移量
- this.byteBuffer.putLong(endPhyoffsetIndex, this.endPhyOffset.get());
- //4位 該索引文件目前的hash slot的個數
- this.byteBuffer.putInt(hashSlotcountIndex, this.hashSlotCount.get());
- //4位 索引文件目前的索引個數
- this.byteBuffer.putInt(indexCountIndex, this.indexCount.get());
Slot槽位，默認每個文件配置的slot是500萬個，每個slot是4位的整型數據，Slot每個節點保存當前已經擁有多少個index數據了。
- //slot的數據存放位置 40 + keyHash %（500W）* 4
- int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
- //Slot Table
- //4字節
- //記錄該slot當前index，如果hash沖突（即absSlotPos一致）作為下一次該slot新增的前置index
- this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
索引消息內容，消息長度固定為20位。
- //Index Linked list
- //topic+message key的hash值
- this.mappedByteBuffer.putInt(absIndexPos, keyHash);
- //消息在CommitLog的物理文件地址, 可以直接查詢到該消息(索引的核心機制)
- this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
- //消息的落盤時間與header里的beginTimestamp的差值(為了節省存儲空間，如果直接存message的落盤時間就得8bytes)
- this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
- //9、記錄該slot上一個index
- //hash沖突處理的關鍵之處, 相同hash值上一個消息索引的index(如果當前消息索引是該hash值的第一個索引，則prevIndex=0, 也是消息索引查找時的停止條件)，每個slot位置的第一個消息的prevIndex就是0的
- this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
- 再論結構
文件結構slot和indexLinkedList可以理解成java中的HashMap。
哎，你說HashMap我可不困了啊，你可別蒙我，這個我熟，什么負載因子、默認大小、擴容機制、紅黑樹，還有多線程下不安全這些。
乖，我知道你熟悉，你跟著我一起學習，這些當然了如指掌，只需要你了解HashMap的結構和沖突即可。
每放入一個新消息的index進來，首先會取MessageKey的HashCode，然后用Hashcode對slot的總數進行取模，決定該消息key的位置，slot的總數默認是500W個。
只要取hash就必然面臨著hash沖突的問題，indexfile也是采用鏈表結構來解決hash沖突，這一點和HashMap一樣的，不過這個不存在紅黑樹轉換這一說，個人猜測這個的沖突數量也達不到很高的級別，所以進行這方面的設計也沒啥必要，甚至變成了強行增加indexfile的文件結構難度。
還有，在indexfile中的slot中放的是最新的index的指針，因為一般查詢的時候大概率是優先查詢最近的消息。
每個slot中放的指針值是索引在indexfile中的偏移量，也就是后面index的位置，而index中存放的就是該消息在commitlog文件中的offset，每個index的大小是20字節，所以根據當前索引是這個文件中的第幾個偏移量，也就很容易定位到索引的位置，根據前面的固定大小可以很快把真實坐標算出來，以此類推，形成一個鏈表的結構。
- 查詢流程
由于indexHeader，slot，index都是固定大小，所以：
- 公式1：第n個slot在indexFile中的起始位置是這樣:40+(n-1)*4
- 公式2：第s個index在indexFile中的起始位置是這樣:40+5000000*4+(s-1)*20
查詢的傳入值除了key外，還包含一個時間起始值以及截止值。
為啥還要傳時間范圍呢?
一個indexFile寫完一個會繼續寫下一個，僅僅一個key無法定位到具體的indexFile，時間范圍就為了更精確的定位到具體的indexFile，縮小查找的范圍，indexFile文件名是一個時間戳，根據這個日期就可以定位到傳入的日期范圍對應在哪個或者哪些indexFile中，是不是很棒。
好了，我們接著說查詢流程。
key-->計算hash值-->hash值對500萬取余算出對應的slot序號-->根據40+(n-1)*4(公式1)算出該slot在文件中的位置-->讀取slot值，也就是index序號-->根據40+5000000*4+(s-1)*20(公式2)算出該index在文件中的位置-->讀取該index-->將key的hash值以及傳入的時間范圍與index的keyHash值以及timeDiff值進行比對。
不滿足則根據index中的preIndexNo找到上一個index，繼續上一步;滿足則根據index中的phyOffset拿到commitLog中的消息。
為啥比對時還要帶上時間范圍呢?
只比key不行嗎?答案是不行，因為key可能會重復，producer在消息生產時可以指定消息的key，這個key顯然無法保證唯一性，那自動生成的msgId呢?也不能保證唯一，你可以去看看msgId的生成規則。
包括當前機器IP+進程號+MessageClientIDSetter.class.getClassLoader()的hashCode值+消息生產時間與broker啟動時間的差值+broker啟動后從0開始單調自增的int值，前面三項很明顯可能重復，后面兩項一個是時間差，一個是重啟歸零，也可能重復。
- 簡單看下源碼，感興趣的下載源碼去研究。
indexfile的添加消息索引的過程
- public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
- //1. 判斷該索引文件的索引數小于最大的索引數，如果>=最大索引數，IndexService就會嘗試新建一個索引文件
- if (this.indexHeader.getIndexCount() < this.indexNum) {
- //2. 計算該message key的hash值
- int keyHash = indexKeyHashMethod(key);
- //3. 根據message key的hash值散列到某個hash slot里
- int slotPos = keyHash % this.hashSlotNum;
- //4. 計算得到該hash slot的實際文件位置Position
- int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
- try {
- //5. 根據該hash slot的實際文件位置absSlotPos得到slot里的值
- //這里有兩種情況：
- //1). slot=0, 當前message的key是該hash值第一個消息索引
- //2). slot>0, 該key hash值上一個消息索引的位置
- int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
- //6. 數據校驗及修正
- if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {
- slotValue = invalidIndex;
- }
- long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
- timeDiff = timeDiff / 1000;
- if (this.indexHeader.getBeginTimestamp() <= 0) {
- timeDiff = 0;
- } else if (timeDiff > Integer.MAX_VALUE) {
- timeDiff = Integer.MAX_VALUE;
- } else if (timeDiff < 0) {
- timeDiff = 0;
- }
- //7. 計算當前消息索引具體的存儲位置(Append模式)
- int absIndexPos =
- IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
- + this.indexHeader.getIndexCount() * indexSize;
- //8. 存入該消息索引
- 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);
- //9. 關鍵之處：在該key hash slot處存入當前消息索引的位置，下次通過該key進行搜索時
- //會找到該key hash slot -> slot value -> curIndex ->
- //if(curIndex.prevIndex>0) pre index (一直循環 直至該curIndex.prevIndex==0就停止)
- this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
- if (this.indexHeader.getIndexCount() <= 1) {
- this.indexHeader.setBeginPhyOffset(phyOffset);
- this.indexHeader.setBeginTimestamp(storeTimestamp);
- }
- this.indexHeader.incHashSlotCount();
- this.indexHeader.incIndexCount();
- this.indexHeader.setEndPhyOffset(phyOffset);
- this.indexHeader.setEndTimestamp(storeTimestamp);
- return true;
- } catch (Exception e) {
- log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);
- }
- } else {
- log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
- + "; index max num = " + this.indexNum);
- }
- return false;
- }
indexfile的索引搜索源碼
- public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,
- final long begin, final long end, boolean lock) {
- if (this.mappedFile.hold()) {
- //1. 計算該key的hash
- int keyHash = indexKeyHashMethod(key);
- //2. 計算該hash value 對應的hash slot位置
- int slotPos = keyHash % this.hashSlotNum;
- //3. 計算該hash value 對應的hash slot物理文件位置
- int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
- FileLock fileLock = null;
- try {
- //4. 取出該hash slot 的值
- int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
- //5. 該slot value <= 0 就代表沒有該key對應的消息索引,直接結束搜索
- // 該slot value > maxIndexCount 就代表該key對應的消息索引超過最大限制，數據有誤,直接結束搜索
- if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
- || this.indexHeader.getIndexCount() <= 1) {
- } else {
- //6. 從當前slot value 開始搜索
- for (int nextIndexToRead = slotValue; ; ) {
- if (phyOffsets.size() >= maxNum) {
- break;
- }
- //7. 找到當前slot value(也就是index count)物理文件位置
- int absIndexPos =
- IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
- + nextIndexToRead * indexSize;
- //8. 讀取消息索引數據
- int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
- long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
- long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
- //9. 獲取該消息索引的上一個消息索引index(可以看成鏈表的prev 指向上一個鏈節點的引用)
- int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);
- //10. 數據校驗
- if (timeDiff < 0) {
- break;
- }
- timeDiff *= 1000L;
- long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
- boolean timeMatched = (timeRead >= begin) && (timeRead <= end);
- //10. 數據校驗比對 hash值和落盤時間
- if (keyHash == keyHashRead && timeMatched) {
- phyOffsets.add(phyOffsetRead);
- }
- //當prevIndex <= 0 或prevIndex > maxIndexCount 或prevIndexRead == nextIndexToRead 或 timeRead < begin 停止搜索
- if (prevIndexRead <= invalidIndex
- || prevIndexRead > this.indexHeader.getIndexCount()
- || prevIndexRead == nextIndexToRead || timeRead < begin) {
- break;
- }
- nextIndexToRead = prevIndexRead;
- }
- }
- } catch (Exception e) {
- log.error("selectPhyOffset exception ", e);
- } finally {
- this.mappedFile.release();
- }
- }
- }
本文轉載自微信公眾號「Java賊船」
