1. 程式人生 > >RocketMQ(十):資料儲存模型設計與實現

RocketMQ(十):資料儲存模型設計與實現

  訊息中介軟體,說是一個通訊元件也沒有錯,因為它的本職工作是做訊息的傳遞。然而要做到高效的訊息傳遞,很重要的一點是資料結構,資料結構設計的好壞,一定程度上決定了該訊息元件的效能以及能力上限。

 

1. 訊息中介軟體的實現方式概述

  訊息中介軟體實現起來自然是很難的,但我們可以從某些角度,簡單了說說實現思路。

  它的最基本的兩個功能介面為:接收訊息的傳送(produce), 訊息的消費(consume). 就像一個郵遞員一樣,經過它與不經過它實質性的東西沒有變化,它只是一箇中介(其他功能效應,咱們拋卻不說)。

  為了實現這兩個基本的介面,我們就得實現兩個最基本的能力:訊息的儲存和查詢。儲存即是接收發送過來的訊息,查詢則包括業務查詢與系統自行查詢推送。

 

我們先來看第一個點:訊息的儲存。

  直接基於記憶體的訊息元件,可以做到非常高效的傳遞,基本上此時的訊息中介軟體就是由幾個記憶體佇列組成,只要保證這幾個佇列的安全性和實時性,就可以工作得很好了。然而基於記憶體則必然意味著能力有限或者成本相當高,所以這樣的設計適用範圍得結合業務現狀做下比對。

  另一個就是基於磁碟的訊息元件,磁碟往往意味著更大的儲存空間,或者某種程度上意味著無限的儲存空間,因為畢竟所有的大資料都是存放在磁碟上的,前提是系統需要協調好各磁碟間的資料關係。然而,磁碟也意味著效能的下降,資料存放起來更麻煩。但rocketmq藉助於作業系統的pagecache和mmap以及順序寫機制,在讀寫效能方面已經非常優化。所以,更重要的是如何設計好磁碟的資料據結構。

 

然後是第二個點:訊息的查詢。

  具體如何查詢,則必然依賴於如何儲存,與上面的原理類似,不必細說。但一般會有兩種消費模型:推送訊息模型和拉取消費模型。即是訊息中介軟體主動向消費者推送訊息,或者是消費者主動查詢訊息中介軟體。二者也各有優劣,推送模型一般可以體現出更強的實時性以及保持比較小的server端儲存空間佔用,但是也帶來了非常大的複雜度,它需要處理各種消費異常、重試、負載均衡、上下線,這不是件小事。而拉取模型則會對訊息中介軟體減輕許多工作,主要是省去了異常、重試、負載均衡類的工作,將這些工作轉嫁到消費者客戶端上。但與此同時,也會對訊息中介軟體提出更多要求,即要求能夠保留足夠長時間的資料,以便所有合法的消費者都可以進行消費。而對於客戶端,則也需要中介軟體提供相應的便利,以便可以實現客戶端的基本訴求,比如消費組管理,上下線管理以及最基本的高效查詢能力。

 

2. rocketmq儲存模型設計概述

  很明顯,rocketmq的初衷就是要應對大資料的訊息傳遞,所以其必然是基於磁碟的儲存。而其效能如上節所述,其利用作業系統的pagecache和mmap機制,讀寫效能非常好,另外他使用順序寫機制,使普通磁碟也能體現出非常高的效能。

  但是,以上幾項,只是為高效能提供了必要的前提。但具體如何利用,還需要從重設計。畢竟,快不是目的,實現需求才是意義。

  rocketmq中主要有四種儲存檔案:commitlog 資料檔案, consumequeue 消費佇列檔案, index 索引檔案, 元資料資訊檔案。最後一個元資料資訊檔案比較簡單,因其資料量小,方便操作。但針對前三個檔案,都會涉及大量的資料問題,所以必然好詳細設計其結構。

  從總體上來說,rocketmq都遵從定長資料結構儲存,定長的最大好處就在於可以快速定位位置,這是其高效能的出發點。定長模型。

  從核心上來說,commitlog檔案儲存了所有原始資料,所有資料想要獲取,都能從或也只能從commitlog檔案中獲取,由於commitlog檔案保持了順序寫的特性,所以其效能非常高。而因資料只有一份,所以也就從根本上保證了資料一致性。

  而根據各業務場景,衍生出了consumequeue和index檔案,即 consumequeue 檔案是為了消費者能夠快速獲取到相應訊息而設計,而index檔案則為了能夠快速搜尋到訊息而設計。從功能上說,consumequeue和index檔案都是索引檔案,只是索引的維度不同。consumequeue 是以topic和queueId維度進行劃分的索引,而index 則是以時間和key作為劃分的索引。有了這兩個索引之後,就可以為各自的業務場景,提供高效能的服務了。具體其如何實現索引,我們稍後再講!

  commitlog vs consumequeue 的儲存模型如下:

 

3. commitlog檔案的儲存結構

  直接順序寫的形式儲存,每個檔案設定固定大小,預設是1G即: 1073741824 bytes. 寫滿一個檔案後,新開一個檔案寫入。檔名就是其儲存的起始訊息偏移量。

  官方描述如下:

CommitLog:訊息主體以及元資料的儲存主體,儲存Producer端寫入的訊息主體內容,訊息內容不是定長的。單個檔案大小預設1G ,檔名長度為20位,左邊補零,剩餘為起始偏移量,比如00000000000000000000代表了第一個檔案,起始偏移量為0,檔案大小為1G=1073741824;當第一個檔案寫滿了,第二個檔案為00000000001073741824,起始偏移量為1073741824,以此類推。訊息主要是順序寫入日誌檔案,當檔案滿了,寫入下一個檔案;

  當給定一個偏移量,要查詢某條訊息時,只需在所有的commitlog檔案中,根據其名字即可知道偏移的資料資訊是否存在其中,即相當於可基於檔案實現一個二分查詢,實際上rocketmq實現得更簡潔,直接一次性查詢即可定位:

    // org.apache.rocketmq.store.CommitLog#getData
    public SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound) {
        int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
        // 1. 先在所有commitlog檔案中查詢到對應所在的 commitlog 分片檔案
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, returnFirstOnNotFound);
        if (mappedFile != null) {
            // 再從該分片檔案中,移動餘數的大小偏移,即可定位到要查詢的訊息記錄了
            int pos = (int) (offset % mappedFileSize);
            SelectMappedBufferResult result = mappedFile.selectMappedBuffer(pos);
            return result;
        }

        return null;
    }
    // 查詢偏移所在commitlog檔案的實現方式:
    // org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)
    // firstMappedFile.getFileFromOffset() / this.mappedFileSize 代表了第一條記錄所處的檔案位置編號
    // offset / this.mappedFileSize 代表當前offset所處的檔案編號
    // 那麼,兩個編號相減就是當前offset對應的檔案編號,因為第一個檔案編號的相對位置是0
    // 但有個前提:就是每個檔案儲存的大小必須是真實的對應的 offset 大小之差,而實際上consumeQueue根本無法確定它存了多少offset
    // 也就是說,只要檔案定長,offset用於定位 commitlog檔案就是合理的
    int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
    MappedFile targetFile = null;
    try {
        // 所以,此處可以找到 commitlog 檔案對應的 mappedFile
        targetFile = this.mappedFiles.get(index);
    } catch (Exception ignored) {
    }
    if (targetFile != null && offset >= targetFile.getFileFromOffset()
        && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
        return targetFile;
    }
    // 如果快速查詢失敗,則退回到遍歷方式, 使用O(n)的複雜度再查詢一次
    for (MappedFile tmpMappedFile : this.mappedFiles) {
        if (offset >= tmpMappedFile.getFileFromOffset()
            && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
            return tmpMappedFile;
        }
    }

  定位到具體的訊息記錄位置後,如何知道要讀多少資料呢?這實際上在commitlog的資料第1個位元組中標明,只需讀出即可知道。

  具體commitlog的儲存實現如下:

    // org.apache.rocketmq.store.CommitLog.DefaultAppendMessageCallback#doAppend
    ...
    // Initialization of storage space
    this.resetByteBuffer(msgStoreItemMemory, msgLen);
    // 1 TOTALSIZE, 首先將訊息大小寫入
    this.msgStoreItemMemory.putInt(msgLen);
    // 2 MAGICCODE
    this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
    // 3 BODYCRC
    this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
    // 4 QUEUEID
    this.msgStoreItemMemory.putInt(msgInner.getQueueId());
    // 5 FLAG
    this.msgStoreItemMemory.putInt(msgInner.getFlag());
    // 6 QUEUEOFFSET
    this.msgStoreItemMemory.putLong(queueOffset);
    // 7 PHYSICALOFFSET
    this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
    // 8 SYSFLAG
    this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
    // 9 BORNTIMESTAMP
    this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
    // 10 BORNHOST
    this.resetByteBuffer(bornHostHolder, bornHostLength);
    this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
    // 11 STORETIMESTAMP
    this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
    // 12 STOREHOSTADDRESS
    this.resetByteBuffer(storeHostHolder, storeHostLength);
    this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
    // 13 RECONSUMETIMES
    this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
    // 14 Prepared Transaction Offset
    this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
    // 15 BODY
    this.msgStoreItemMemory.putInt(bodyLength);
    if (bodyLength > 0)
        this.msgStoreItemMemory.put(msgInner.getBody());
    // 16 TOPIC
    this.msgStoreItemMemory.put((byte) topicLength);
    this.msgStoreItemMemory.put(topicData);
    // 17 PROPERTIES
    this.msgStoreItemMemory.putShort((short) propertiesLength);
    if (propertiesLength > 0)
        this.msgStoreItemMemory.put(propertiesData);

    final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
    // Write messages to the queue buffer
    byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
    ...

  可以看出,commitlog的儲存還是比較簡單的,因為其主要就是負責將接收到的所有訊息,依次寫入同一檔案中。因為專一所以專業。

 

4. consumequeue檔案的儲存結構

  consumequeue作為消費者的重要依據,同樣起著非常重要的作用。消費者在進行消費時,會使用一些偏移量作為依據(拉取模型實現)。而這些個偏移量,實際上就是指的consumequeue的偏移量(注意不是commitlog的偏移量)。這樣做有什麼好處呢?首先,consumequeue作為索引檔案,它被要求要有非常高的查詢效能,所以越簡單越好。最好是能夠一次性定位到資料!

  如果想一次性定位資料,那麼唯一的辦法是直接使用commitlog的offset。但這會帶來一個最大的問題,就是當我當前訊息消費拉取完成後,下一條訊息在哪裡呢?如果單靠commitlog檔案,那麼,它必然需要將下一條訊息讀入,然後再根據topic判定是不是需要的資料。如此一來,就必然存在大量的commitlog檔案的io問題了。所以,這看起來是非常快速的一個解決方案,最終又變成了非常費力的方案了。

  而使用commitlog檔案的offset,則好了許多。因為consumequeue的檔案儲存格式是一條訊息佔20位元組,即定長。根據這20位元組,你可以找到commitlog的offset. 而因為consumequeue本身就是按照topic/queueId進行劃分的,所以,本次消費完成後,下一次消費的資料必定就在consumequeue的下一位置。如此簡單快速搞得定了。具體consume的儲存格式,如官方描述:

ConsumeQueue:訊息消費佇列,引入的目的主要是提高訊息消費的效能,由於RocketMQ是基於主題topic的訂閱模式,訊息消費是針對主題進行的,如果要遍歷commitlog檔案中根據topic檢索訊息是非常低效的。Consumer即可根據ConsumeQueue來查詢待消費的訊息。其中,ConsumeQueue(邏輯消費佇列)作為消費訊息的索引,儲存了指定Topic下的佇列訊息在CommitLog中的起始物理偏移量offset,訊息大小size和訊息Tag的HashCode值。consumequeue檔案可以看成是基於topic的commitlog索引檔案,故consumequeue資料夾的組織方式如下:topic/queue/file三層組織結構,具體儲存路徑為:$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同樣consumequeue檔案採取定長設計,每一個條目共20個位元組,分別為8位元組的commitlog物理偏移量、4位元組的訊息長度、8位元組tag hashcode,單個檔案由30W個條目組成,可以像陣列一樣隨機訪問每一個條目,每個ConsumeQueue檔案大小約5.72M;

  其中fileName也是以偏移量作為命名依據,因為這樣才能根據offset快速查詢到資料所在的分片檔案。

  其儲存實現如下:

    // org.apache.rocketmq.store.ConsumeQueue#putMessagePositionInfo
    private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
        final long cqOffset) {

        if (offset + size <= this.maxPhysicOffset) {
            log.warn("Maybe try to build consume queue repeatedly maxPhysicOffset={} phyOffset={}", maxPhysicOffset, offset);
            return true;
        }
        // 依次寫入 offset + size + tagsCode
        this.byteBufferIndex.flip();
        this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
        this.byteBufferIndex.putLong(offset);
        this.byteBufferIndex.putInt(size);
        this.byteBufferIndex.putLong(tagsCode);

        final long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;

        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
        if (mappedFile != null) {

            if (mappedFile.isFirstCreateInQueue() && cqOffset != 0 && mappedFile.getWrotePosition() == 0) {
                this.minLogicOffset = expectLogicOffset;
                this.mappedFileQueue.setFlushedWhere(expectLogicOffset);
                this.mappedFileQueue.setCommittedWhere(expectLogicOffset);
                this.fillPreBlank(mappedFile, expectLogicOffset);
                log.info("fill pre blank space " + mappedFile.getFileName() + " " + expectLogicOffset + " "
                    + mappedFile.getWrotePosition());
            }

            if (cqOffset != 0) {
                long currentLogicOffset = mappedFile.getWrotePosition() + mappedFile.getFileFromOffset();

                if (expectLogicOffset < currentLogicOffset) {
                    log.warn("Build  consume queue repeatedly, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
                        expectLogicOffset, currentLogicOffset, this.topic, this.queueId, expectLogicOffset - currentLogicOffset);
                    return true;
                }

                if (expectLogicOffset != currentLogicOffset) {
                    LOG_ERROR.warn(
                        "[BUG]logic queue order maybe wrong, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
                        expectLogicOffset,
                        currentLogicOffset,
                        this.topic,
                        this.queueId,
                        expectLogicOffset - currentLogicOffset
                    );
                }
            }
            this.maxPhysicOffset = offset + size;
            // 將buffer寫入 consumequeue 的 mappedFile 中
            return mappedFile.appendMessage(this.byteBufferIndex.array());
        }
        return false;
    }
    當需要進行查詢進,也就會根據offset, 定位到某個 consumequeue 檔案,然後再根據偏移餘數資訊,再找到對應記錄,取出20位元組,即是 commitlog資訊。此處實現與 commitlog 的offset查詢實現如出一轍。
    // 查詢索引所在檔案的實現,如下:
    // org.apache.rocketmq.store.ConsumeQueue#getIndexBuffer
    public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
        int mappedFileSize = this.mappedFileSize;
        // 給到客戶端的偏移量是除以 20 之後的,也就是說 如果上一次的偏移量是 1, 那麼下一次的偏移量應該是2
        // 一次性消費多條記錄另算, 自行加減
        long offset = startIndex * CQ_STORE_UNIT_SIZE;
        if (offset >= this.getMinLogicOffset()) {
            // 委託給mappedFileQueue進行查詢到單個具體的consumequeue檔案
            // 根據 offset 和規範的命名,可以快速定位分片檔案,如上 commitlog 的查詢實現
            MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
            if (mappedFile != null) {
                // 再根據剩餘的偏移量,直接類似於陣列下標的形式,一次性定位到具體的資料記錄
                SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
                return result;
            }
        }
        return null;
    }

  如果想一次性消費多條訊息,則只需要依次從查詢到索引記錄開始,依次讀取多條,然後同理回查commitlog即可。即consumequeue的連續,成就了commitlog的不連續。如下訊息拉取實現:

    // org.apache.rocketmq.store.DefaultMessageStore#getMessage
    // 其中 bufferConsumeQueue 是剛剛查找出的consumequeue的起始消費位置
    // 基於此檔案迭代,完成多訊息記錄消費
    ...
    long nextPhyFileStartOffset = Long.MIN_VALUE;
    long maxPhyOffsetPulling = 0;

    int i = 0;
    final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);
    final boolean diskFallRecorded = this.messageStoreConfig.isDiskFallRecorded();
    ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
    for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
        // 依次取出commitlog的偏移量,資料大小,hashCode
        // 一次迴圈即是取走一條記錄,多次迴圈則依次往下讀取
        long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
        int sizePy = bufferConsumeQueue.getByteBuffer().getInt();
        long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();

        maxPhyOffsetPulling = offsetPy;

        if (nextPhyFileStartOffset != Long.MIN_VALUE) {
            if (offsetPy < nextPhyFileStartOffset)
                continue;
        }

        boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);

        if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(),
            isInDisk)) {
            break;
        }

        boolean extRet = false, isTagsCodeLegal = true;
        if (consumeQueue.isExtAddr(tagsCode)) {
            extRet = consumeQueue.getExt(tagsCode, cqExtUnit);
            if (extRet) {
                tagsCode = cqExtUnit.getTagsCode();
            } else {
                // can't find ext content.Client will filter messages by tag also.
                log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}, topic={}, group={}",
                    tagsCode, offsetPy, sizePy, topic, group);
                isTagsCodeLegal = false;
            }
        }

        if (messageFilter != null
            && !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) {
            if (getResult.getBufferTotalSize() == 0) {
                status = GetMessageStatus.NO_MATCHED_MESSAGE;
            }

            continue;
        }

        SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
        if (null == selectResult) {
            if (getResult.getBufferTotalSize() == 0) {
                status = GetMessageStatus.MESSAGE_WAS_REMOVING;
            }

            nextPhyFileStartOffset = this.commitLog.rollNextFile(offsetPy);
            continue;
        }

        if (messageFilter != null
            && !messageFilter.isMatchedByCommitLog(selectResult.getByteBuffer().slice(), null)) {
            if (getResult.getBufferTotalSize() == 0) {
                status = GetMessageStatus.NO_MATCHED_MESSAGE;
            }
            // release...
            selectResult.release();
            continue;
        }

        this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet();
        getResult.addMessage(selectResult);
        status = GetMessageStatus.FOUND;
        nextPhyFileStartOffset = Long.MIN_VALUE;
    }

    if (diskFallRecorded) {
        long fallBehind = maxOffsetPy - maxPhyOffsetPulling;
        brokerStatsManager.recordDiskFallBehindSize(group, topic, queueId, fallBehind);
    }
    // 分配下一次讀取的offset偏移資訊,同樣要除以單條索引大小
    nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);

    long diff = maxOffsetPy - maxPhyOffsetPulling;
    long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
        * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
    getResult.setSuggestPullingFromSlave(diff > memory);
    ...

  以上即理論的實現,無須多言。

 

5. index檔案的儲存結構

  index檔案是為搜尋場景而生的,如果沒有搜尋業務需求,則這個實現是意義不大的。一般這種搜尋,主要用於後臺查詢驗證類使用,或者有其他同的有妙用,不得而知。總之,一切為搜尋。它更多的需要藉助於時間限定,以key或者id進行查詢。

  官方描述如下:

IndexFile(索引檔案)提供了一種可以通過key或時間區間來查詢訊息的方法。Index檔案的儲存位置是:$HOME \store\index\${fileName},檔名fileName是以建立時的時間戳命名的,固定的單個IndexFile檔案大小約為400M,一個IndexFile可以儲存 2000W個索引,IndexFile的底層儲存設計為在檔案系統中實現HashMap結構,故rocketmq的索引檔案其底層實現為hash索引。
IndexFile索引檔案為使用者提供通過“按照Message Key查詢訊息”的訊息索引查詢服務,IndexFile檔案的儲存位置是:$HOME\store\index\${fileName},檔名fileName是以建立時的時間戳命名的,檔案大小是固定的,等於40+500W\*4+2000W\*20= 420000040個位元組大小。如果訊息的properties中設定了UNIQ_KEY這個屬性,就用 topic + “#” + UNIQ_KEY的value作為 key 來做寫入操作。如果訊息設定了KEYS屬性(多個KEY以空格分隔),也會用 topic + “#” + KEY 來做索引。
其中的索引資料包含了Key Hash/CommitLog Offset/Timestamp/NextIndex offset 這四個欄位,一共20 Byte。NextIndex offset 即前面讀出來的 slotValue,如果有 hash衝突,就可以用這個欄位將所有衝突的索引用連結串列的方式串起來了。Timestamp記錄的是訊息storeTimestamp之間的差,並不是一個絕對的時間。整個Index File的結構如圖,40 Byte 的Header用於儲存一些總的統計資訊,4\*500W的 Slot Table並不儲存真正的索引資料,而是儲存每個槽位對應的單向連結串列的頭。20\*2000W 是真正的索引資料,即一個 Index File 可以儲存 2000W個索引。

  具體結構圖如下:

  那麼,如果要查詢一個key, 應當如何查詢呢?rocketmq會根據時間段找到一個index索引分版,然後再根據key做hash得到一個值,然後定位到 slotValue . 然後再從slotValue去取出索引資料的地址,找到索引資料,然後再回查 commitlog 檔案。從而得到具體的訊息資料。也就是,相當於搜尋經歷了四級查詢: 索引分片檔案查詢 -> slotValue 查詢 -> 索引資料查詢 -> commitlog 查詢 。 

  具體查詢實現如下:

    // org.apache.rocketmq.broker.processor.QueryMessageProcessor#queryMessage
    public RemotingCommand queryMessage(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response =
            RemotingCommand.createResponseCommand(QueryMessageResponseHeader.class);
        final QueryMessageResponseHeader responseHeader =
            (QueryMessageResponseHeader) response.readCustomHeader();
        final QueryMessageRequestHeader requestHeader =
            (QueryMessageRequestHeader) request
                .decodeCommandCustomHeader(QueryMessageRequestHeader.class);

        response.setOpaque(request.getOpaque());

        String isUniqueKey = request.getExtFields().get(MixAll.UNIQUE_MSG_QUERY_FLAG);
        if (isUniqueKey != null && isUniqueKey.equals("true")) {
            requestHeader.setMaxNum(this.brokerController.getMessageStoreConfig().getDefaultQueryMaxNum());
        }
        // 從索引檔案中查詢訊息
        final QueryMessageResult queryMessageResult =
            this.brokerController.getMessageStore().queryMessage(requestHeader.getTopic(),
                requestHeader.getKey(), requestHeader.getMaxNum(), requestHeader.getBeginTimestamp(),
                requestHeader.getEndTimestamp());
        assert queryMessageResult != null;

        responseHeader.setIndexLastUpdatePhyoffset(queryMessageResult.getIndexLastUpdatePhyoffset());
        responseHeader.setIndexLastUpdateTimestamp(queryMessageResult.getIndexLastUpdateTimestamp());

        if (queryMessageResult.getBufferTotalSize() > 0) {
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);

            try {
                FileRegion fileRegion =
                    new QueryMessageTransfer(response.encodeHeader(queryMessageResult
                        .getBufferTotalSize()), queryMessageResult);
                ctx.channel().writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        queryMessageResult.release();
                        if (!future.isSuccess()) {
                            log.error("transfer query message by page cache failed, ", future.cause());
                        }
                    }
                });
            } catch (Throwable e) {
                log.error("", e);
                queryMessageResult.release();
            }

            return null;
        }

        response.setCode(ResponseCode.QUERY_NOT_FOUND);
        response.setRemark("can not find message, maybe time range not correct");
        return response;
    }
    // org.apache.rocketmq.store.DefaultMessageStore#queryMessage
    @Override
    public QueryMessageResult queryMessage(String topic, String key, int maxNum, long begin, long end) {
        QueryMessageResult queryMessageResult = new QueryMessageResult();

        long lastQueryMsgTime = end;

        for (int i = 0; i < 3; i++) {
            // 委託給 indexService 搜尋記錄, 時間是必備引數
            QueryOffsetResult queryOffsetResult = this.indexService.queryOffset(topic, key, maxNum, begin, lastQueryMsgTime);
            if (queryOffsetResult.getPhyOffsets().isEmpty()) {
                break;
            }

            Collections.sort(queryOffsetResult.getPhyOffsets());

            queryMessageResult.setIndexLastUpdatePhyoffset(queryOffsetResult.getIndexLastUpdatePhyoffset());
            queryMessageResult.setIndexLastUpdateTimestamp(queryOffsetResult.getIndexLastUpdateTimestamp());

            for (int m = 0; m < queryOffsetResult.getPhyOffsets().size(); m++) {
                long offset = queryOffsetResult.getPhyOffsets().get(m);

                try {

                    boolean match = true;
                    MessageExt msg = this.lookMessageByOffset(offset);
                    if (0 == m) {
                        lastQueryMsgTime = msg.getStoreTimestamp();
                    }

                    if (match) {
                        SelectMappedBufferResult result = this.commitLog.getData(offset, false);
                        if (result != null) {
                            int size = result.getByteBuffer().getInt(0);
                            result.getByteBuffer().limit(size);
                            result.setSize(size);
                            queryMessageResult.addMessage(result);
                        }
                    } else {
                        log.warn("queryMessage hash duplicate, {} {}", topic, key);
                    }
                } catch (Exception e) {
                    log.error("queryMessage exception", e);
                }
            }

            if (queryMessageResult.getBufferTotalSize() > 0) {
                break;
            }

            if (lastQueryMsgTime < begin) {
                break;
            }
        }

        return queryMessageResult;
    }

    public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
        List<Long> phyOffsets = new ArrayList<Long>(maxNum);

        long indexLastUpdateTimestamp = 0;
        long indexLastUpdatePhyoffset = 0;
        maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
        try {
            this.readWriteLock.readLock().lock();
            if (!this.indexFileList.isEmpty()) {
                //從最後一個索引檔案,依次搜尋
                for (int i = this.indexFileList.size(); i > 0; i--) {
                    IndexFile f = this.indexFileList.get(i - 1);
                    boolean lastFile = i == this.indexFileList.size();
                    if (lastFile) {
                        indexLastUpdateTimestamp = f.getEndTimestamp();
                        indexLastUpdatePhyoffset = f.getEndPhyOffset();
                    }
                    // 判定該時間段是否資料是否在該索引檔案中
                    if (f.isTimeMatched(begin, end)) {
                        // 構建出 key的hash, 然後查詢 slotValue, 然後得以索引資料, 然後將offset放入 phyOffsets 中
                        f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end, lastFile);
                    }

                    if (f.getBeginTimestamp() < begin) {
                        break;
                    }

                    if (phyOffsets.size() >= maxNum) {
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("queryMsg exception", e);
        } finally {
            this.readWriteLock.readLock().unlock();
        }

        return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
    }
    // org.apache.rocketmq.store.index.IndexFile#selectPhyOffset
    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()) {
            int keyHash = indexKeyHashMethod(key);
            int slotPos = keyHash % this.hashSlotNum;
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

            FileLock fileLock = null;
            try {
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);

                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
                    || this.indexHeader.getIndexCount() <= 1) {
                    // 超出搜尋範圍,不處理
                } else {
                    for (int nextIndexToRead = slotValue; ; ) {
                        if (phyOffsets.size() >= maxNum) {
                            break;
                        }

                        int absIndexPos =
                            IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                                + nextIndexToRead * indexSize;
                        // 依次讀出 keyHash+offset+timeDiff+nextOffset
                        int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
                        long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);

                        long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
                        int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);

                        if (timeDiff < 0) {
                            break;
                        }

                        timeDiff *= 1000L;
                        // 根據檔名可得到索引寫入時間
                        long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
                        boolean timeMatched = (timeRead >= begin) && (timeRead <= end);

                        if (keyHash == keyHashRead && timeMatched) {
                            phyOffsets.add(phyOffsetRead);
                        }

                        if (prevIndexRead <= invalidIndex
                            || prevIndexRead > this.indexHeader.getIndexCount()
                            || prevIndexRead == nextIndexToRead || timeRead < begin) {
                            break;
                        }

                        nextIndexToRead = prevIndexRead;
                    }
                }
            } catch (Exception e) {
                log.error("selectPhyOffset exception ", e);
            } finally {
                if (fileLock != null) {
                    try {
                        fileLock.release();
                    } catch (IOException e) {
                        log.error("Failed to release the lock", e);
                    }
                }

                this.mappedFile.release();
            }
        }
    }

  看起來挺費勁,但真正處理起來效能還好,雖然沒有consumequeue高效,但有mmap和pagecache的加持,效率還是扛扛的。而且,搜尋相對慢一些,使用者也是可以接受的嘛。畢竟這只是一個附加功能,並非核心所在。

  而索引檔案並沒有使用什麼高效的搜尋演算法,而是簡單從最後一個檔案遍歷完成,因為時間戳不一定總是有規律的,與其隨意查詢,還不如直接線性查詢。另外,實際上對於索引重建問題,搜尋可能不一定會有效。不過,我們可以通過擴大搜索時間範圍的方式,總是能夠找到存在的資料。而且因其使用hash索引實現,效能還是不錯的。

  另外,index索引檔案與commitlog和consumequeue有一個不一樣的地方,就是它不能進行順序寫,因為hash儲存,寫一定是任意的。且其slotValue以一些統計資訊可能隨時發生變化,這也給順序寫帶來了不可解決的問題。

  其具體寫索引過程如下:

    // org.apache.rocketmq.store.index.IndexFile#putKey
    public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        if (this.indexHeader.getIndexCount() < this.indexNum) {
            int keyHash = indexKeyHashMethod(key);
            int slotPos = keyHash % this.hashSlotNum;
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

            FileLock fileLock = null;

            try {
                // 先嚐試拉取slot對應的資料
                // 如果為0則說明是第一次寫入, 否則為當前的索引條數
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                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;
                }
                // 直接計算出本次儲存的索引記錄位置
                // 因索引條數只會依次增加,故索引資料將表現為順序寫樣子,主要是保證了資料不會寫衝突了
                int absIndexPos =
                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                        + this.indexHeader.getIndexCount() * indexSize;
                // 按協議寫入內容即可
                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);
                // 寫入slotValue為當前可知的索引記錄條數
                // 即每次寫入索引之後,如果存在hash衝突,那麼它會寫入自身的位置
                // 而此時 slotValue 必定存在一個值,那就是上一個發生衝突的索引,從而形成自然的連結串列
                // 查詢資料時,只需根據slotValue即可以找到上一個寫入的索引,這設計妙哉!
                // 做了2點關鍵性保證:1. 資料自增不衝突; 2. hash衝突自重新整理; 磁碟版的hash結構已然形成
                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());

                if (this.indexHeader.getIndexCount() <= 1) {
                    this.indexHeader.setBeginPhyOffset(phyOffset);
                    this.indexHeader.setBeginTimestamp(storeTimestamp);
                }

                if (invalidIndex == slotValue) {
                    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);
            } finally {
                if (fileLock != null) {
                    try {
                        fileLock.release();
                    } catch (IOException e) {
                        log.error("Failed to release the lock", e);
                    }
                }
            }
        } else {
            log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
                + "; index max num = " + this.indexNum);
        }

        return false;
    }

  rocketmq 巧妙地使用了自增結構和hash slot, 完美實現一個磁碟版的hash索引。相信這也會給我們平時的工作帶來一些提示。

  

6. 寫在最後

  以上就是本文對rocketmq的儲存模型設計的解析了,通過這些解析,相信大家對其工作原理也會有質的理解。儲存實際上是目前我們的許多的系統中的非常核心部分,因為大部分的業務幾乎都是在儲存之前做一些簡單的計算。

      很顯然業務很重要,但有了儲存的底子,還何愁業務實現難?

  

&n