RocketMQ消息存儲源碼分析

存儲目錄機(jī)構(gòu)

  • commitLog:消息存儲目錄(消息主要都存儲在這里)
  • config:運(yùn)行期間一些配置信息
  • consumerqueue:消息消費(fèi)隊列存儲目錄
  • index:消息索引文件存儲目錄
  • abort:如果存在該文件壽命Broker非正常關(guān)閉(啟動時生成驹尼,正常關(guān)閉時刪除敢会,0KB文件)
  • checkpoint:文件檢查點(diǎn),存儲CommitLog文件最后一次刷盤時間戳、consumerquueue最后一次刷盤時間麸塞,index索引文件最后一次刷盤時間戳

消息存儲大體流程

producer發(fā)送消息過來后根據(jù)netty監(jiān)聽端口調(diào)用DefaultMessageStore#putMessage方法笨奠,rocketMQ默認(rèn)是異步刷盤,首先會把消息寫入到內(nèi)存中梳毙,然后經(jīng)過一段時間再刷到磁盤里哺窄。
當(dāng)消息生產(chǎn)者提交的消息存儲在CommitLog文件中,ConsumerQueue账锹、IndexFile需要及時更新萌业,否則消息無法及時被消費(fèi),根據(jù)消息屬性查找消息也會出現(xiàn)較大延遲奸柬。RocketMQ通過開啟一個線程ReputMessageService來準(zhǔn)實時轉(zhuǎn)發(fā)CommitLog文件更新事件生年,相應(yīng)的任務(wù)處理器根據(jù)轉(zhuǎn)發(fā)的消息及時更新ConsumerQueue、IndexFile文件廓奕。

消息存儲到緩存區(qū)

當(dāng)我們調(diào)用發(fā)送消息接口后抱婉,會調(diào)用DefaultMessageStore#asyncPutMessage

    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
//主要是判斷當(dāng)前的broker是否可以寫入,如果為slave則不寫入等
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
        if (checkStoreStatus != PutMessageStatus.PUT_OK) {
            return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus, null));
        }
//判斷消息是否合法懂从,超長等
        PutMessageStatus msgCheckStatus = this.checkMessage(msg);
        if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
            return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));
        }

        long beginTime = this.getSystemClock().now();
//(下圖有詳細(xì)代碼講解) 將消息寫入commitlog授段,主要流程
        CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
        putResultFuture.thenAccept((result) -> {
            long elapsedTime = this.getSystemClock().now() - beginTime;
        });
...
        return putResultFuture;
    }
  • this.commitLog.asyncPutMessage(msg)代碼詳解
    public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
        // 記錄存儲時間
        msg.setStoreTimestamp(System.currentTimeMillis());
        // 加密消息體
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        AppendMessageResult result = null;
        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

        String topic = msg.getTopic();
        int queueId = msg.getQueueId();
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
//判斷是否為事務(wù)消息,這邊省略代碼番甩,后面另講
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {...  }

        long elapsedTimeInLock = 0;
        MappedFile unlockMappedFile = null;
//后去最后一次寫入的映射文件
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            this.beginTimeInLock = beginLockTimestamp;
            msg.setStoreTimestamp(beginLockTimestamp);
//判斷如果mappedFile如果為空或者已滿,創(chuàng)建新的mappedFile文件
            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
//如果創(chuàng)建失敗,直接返回侵贵,省略代碼...
            if (null == mappedFile) {...            }
//(下面有代碼詳細(xì)解釋),寫入消息到mappedFile中
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {
                case PUT_OK:
                    break;
                case END_OF_FILE:
                    unlockMappedFile = mappedFile;
                    // Create a new file, re-write the message
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                    if (null == mappedFile) {
                        // XXX: warn and notify me
                        log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                        beginTimeInLock = 0;
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));
                    }
                    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                    break;
                case MESSAGE_SIZE_EXCEEDED:
                case PROPERTIES_SIZE_EXCEEDED:
                    beginTimeInLock = 0;
                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
                case UNKNOWN_ERROR:
                    beginTimeInLock = 0;
                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
                default:
                    beginTimeInLock = 0;
                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
            }

            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            beginTimeInLock = 0;
        } finally {
            putMessageLock.unlock();
        }

        if (elapsedTimeInLock > 500) {
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
        }

        if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
            this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
        }

        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

        // Statistics
        storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
        storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
//(下面有代碼詳解)開啟刷盤
        CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, putMessageResult, msg);
//(下面有代碼詳解)分發(fā)數(shù)據(jù)給indexFile缘薛,還是consumer
        CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, putMessageResult, msg);
        return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
            if (flushStatus != PutMessageStatus.PUT_OK) {
                putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
            }
            if (replicaStatus != PutMessageStatus.PUT_OK) {
                putMessageResult.setPutMessageStatus(replicaStatus);
            }
            return putMessageResult;
        });
    }
  • mappedFile.appendMessage(msg, this.appendMessageCallback)代碼詳解
    public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
        assert messageExt != null;
        assert cb != null;
//獲取文件的寫入指針位置
        int currentPos = this.wrotePosition.get();
//如果指針位置大于文件大小則直接返回
        if (currentPos < this.fileSize) {
//通過writeBuffer.slice()創(chuàng)建一個與MappedFile共享的內(nèi)存區(qū),并設(shè)置position為當(dāng)前指針
            ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
            byteBuffer.position(currentPos);
            AppendMessageResult result;
            if (messageExt instanceof MessageExtBrokerInner) {
//通過回調(diào)方法寫入
                result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
            } else if (messageExt instanceof MessageExtBatch) {
                result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
            } else {
                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
            }
            this.wrotePosition.addAndGet(result.getWroteBytes());
            this.storeTimestamp = result.getStoreTimestamp();
            return result;
        }
        log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
    }
具體代碼

        public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
            final MessageExtBrokerInner msgInner) {
//文件寫入位置
            long wroteOffset = fileFromOffset + byteBuffer.position();
            int sysflag = msgInner.getSysFlag();
            int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
            int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
            ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
            ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
            this.resetByteBuffer(storeHostHolder, storeHostLength);
//設(shè)置消息id
            String msgId;
            if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
                msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
            } else {
                msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
            }

            keyBuilder.setLength(0);
            keyBuilder.append(msgInner.getTopic());
            keyBuilder.append('-');
            keyBuilder.append(msgInner.getQueueId());
            String key = keyBuilder.toString();
//獲取該消息在隊列中的偏移量
            Long queueOffset = CommitLog.this.topicQueueTable.get(key);
            if (null == queueOffset) {
                queueOffset = 0L;
                CommitLog.this.topicQueueTable.put(key, queueOffset);
            }

            // Transaction messages that require special handling
            final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
            switch (tranType) {
                // Prepared and Rollback message is not consumed, will not enter the
                // consumer queuec
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    queueOffset = 0L;
                    break;
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                default:
                    break;
            }

            /**
             * Serialize message
             */
            final byte[] propertiesData =
                msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);

            final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;

            if (propertiesLength > Short.MAX_VALUE) {
                log.warn("putMessage message properties length too long. length={}", propertiesData.length);
                return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
            }
//獲取消息主題大小
            final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
            final int topicLength = topicData.length;
//獲取消息體大小
            final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
//計算消息的總大小
            final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
//消息不能超過限制長度
            if (msgLen > this.maxMessageSize) {
                CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
                    + ", maxMessageSize: " + this.maxMessageSize);
                return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
            }
//如果沒有足夠的存儲空間則新創(chuàng)建CommitLog文件
            if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
                this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
                // 1 TOTALSIZE
                this.msgStoreItemMemory.putInt(maxBlank);
                // 2 MAGICCODE
                this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
                // 3 The remaining space may be any value
                // Here the length of the specially set maxBlank
                final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
                byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
                return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
                    queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
            }

            // Initialization of storage space省略一大堆初始化空間的
            this.resetByteBuffer(msgStoreItemMemory, msgLen);
         ...

            final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
            // 把消息寫入byteBuffer緩沖區(qū)中
            byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);

            AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
                msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);

            switch (tranType) {
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    break;
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    // 更新消息隊列偏移量
                    CommitLog.this.topicQueueTable.put(key, ++queueOffset);
                    break;
                default:
                    break;
            }
            return result;
        }

分發(fā)通知consumeQueue跟indexFile更新

刷盤

RocketMQ的存儲是基于JDK NIO的內(nèi)存映射機(jī)制(MappedByteBuffer)的窍育,消息存儲首先將消息追加到內(nèi)存,再根據(jù)配置的刷盤策略在不同時間進(jìn)行刷寫磁盤宴胧。

啟動時校驗是否正確this.mappedFileQueue.load()

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末漱抓,一起剝皮案震驚了整個濱河市,隨后出現(xiàn)的幾起案子恕齐,更是在濱河造成了極大的恐慌乞娄,老刑警劉巖,帶你破解...
    沈念sama閱讀 211,817評論 6 492
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件,死亡現(xiàn)場離奇詭異仪或,居然都是意外死亡确镊,警方通過查閱死者的電腦和手機(jī),發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 90,329評論 3 385
  • 文/潘曉璐 我一進(jìn)店門范删,熙熙樓的掌柜王于貴愁眉苦臉地迎上來蕾域,“玉大人,你說我怎么就攤上這事到旦≈枷铮” “怎么了?”我有些...
    開封第一講書人閱讀 157,354評論 0 348
  • 文/不壞的土叔 我叫張陵添忘,是天一觀的道長采呐。 經(jīng)常有香客問我,道長昔汉,這世上最難降的妖魔是什么懈万? 我笑而不...
    開封第一講書人閱讀 56,498評論 1 284
  • 正文 為了忘掉前任,我火速辦了婚禮靶病,結(jié)果婚禮上,老公的妹妹穿的比我還像新娘口予。我一直安慰自己娄周,他們只是感情好,可當(dāng)我...
    茶點(diǎn)故事閱讀 65,600評論 6 386
  • 文/花漫 我一把揭開白布沪停。 她就那樣靜靜地躺著煤辨,像睡著了一般。 火紅的嫁衣襯著肌膚如雪木张。 梳的紋絲不亂的頭發(fā)上众辨,一...
    開封第一講書人閱讀 49,829評論 1 290
  • 那天,我揣著相機(jī)與錄音舷礼,去河邊找鬼鹃彻。 笑死,一個胖子當(dāng)著我的面吹牛妻献,可吹牛的內(nèi)容都是我干的蛛株。 我是一名探鬼主播,決...
    沈念sama閱讀 38,979評論 3 408
  • 文/蒼蘭香墨 我猛地睜開眼育拨,長吁一口氣:“原來是場噩夢啊……” “哼谨履!你這毒婦竟也來了?” 一聲冷哼從身側(cè)響起熬丧,我...
    開封第一講書人閱讀 37,722評論 0 266
  • 序言:老撾萬榮一對情侶失蹤笋粟,失蹤者是張志新(化名)和其女友劉穎,沒想到半個月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體害捕,經(jīng)...
    沈念sama閱讀 44,189評論 1 303
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡唆香,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 36,519評論 2 327
  • 正文 我和宋清朗相戀三年,在試婚紗的時候發(fā)現(xiàn)自己被綠了吨艇。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片躬它。...
    茶點(diǎn)故事閱讀 38,654評論 1 340
  • 序言:一個原本活蹦亂跳的男人離奇死亡,死狀恐怖东涡,靈堂內(nèi)的尸體忽然破棺而出冯吓,到底是詐尸還是另有隱情,我是刑警寧澤疮跑,帶...
    沈念sama閱讀 34,329評論 4 330
  • 正文 年R本政府宣布组贺,位于F島的核電站,受9級特大地震影響祖娘,放射性物質(zhì)發(fā)生泄漏失尖。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 39,940評論 3 313
  • 文/蒙蒙 一渐苏、第九天 我趴在偏房一處隱蔽的房頂上張望掀潮。 院中可真熱鬧,春花似錦琼富、人聲如沸仪吧。這莊子的主人今日做“春日...
    開封第一講書人閱讀 30,762評論 0 21
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽薯鼠。三九已至,卻和暖如春械蹋,著一層夾襖步出監(jiān)牢的瞬間出皇,已是汗流浹背。 一陣腳步聲響...
    開封第一講書人閱讀 31,993評論 1 266
  • 我被黑心中介騙來泰國打工哗戈, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留郊艘,地道東北人。 一個月前我還...
    沈念sama閱讀 46,382評論 2 360
  • 正文 我出身青樓谱醇,卻偏偏與公主長得像暇仲,于是被迫代替她去往敵國和親。 傳聞我的和親對象是個殘疾皇子副渴,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 43,543評論 2 349

推薦閱讀更多精彩內(nèi)容