文章目錄

• • 消息存儲
  • • 1.實時更新消息消費隊列和索引文件
    • • 1.1.轉發到ConsumerQueue
      • 1.2.轉發到Index
    • 2.消息隊列和索引文件恢復
    • • 2.1.存儲文件加載
      • • 2.1.1.加載commitLog文件
        • 2.1.2.加載消息消費隊列
        • 2.1.3.加載索引文件
      • 2.2.正常恢復
      • 2.3.異常恢復

消息存儲

1.實時更新消息消費隊列和索引文件

消息消費隊文件、消息屬性索引文件都是基于CommitLog文件構建的，當消息生產者提交的消息存儲在CommitLog文件中，ConsumerQueue、IndexFile需要及時更新，否則消息無法及時被消費，根據消息屬性查找消息也會出現較大延遲。RocketMQ通過開啟一個線程ReputMessageService來準實時轉發CommitLog文件更新事件，相應的任務處理器根據轉發的消息及時更新ConsumerQueue、IndexFile文件。

構建消息消費隊列和索引文件的時序圖：

DefaultMessageStore#start

log.info("[SetReputOffset] maxPhysicalPosInLogicQueue={} clMinOffset={} clMaxOffset={} clConfirmedOffset={}",maxPhysicalPosInLogicQueue, this.commitLog.getMinOffset(), this.commitLog.getMaxOffset(), this.commitLog.getConfirmOffset()); //設置CommitLog內存中最大偏移量 this.reputMessageService.setReputFromOffset(maxPhysicalPosInLogicQueue); //啟動消息分發服務線程 this.reputMessageService.start();

ReputMessageService#run

DefaultMessageStore.log.info(this.getServiceName() + " service started");//線程狀態為啟動狀態->每隔1毫秒就繼續嘗試推送消息到消息消費隊列和索引文件 while (!this.isStopped()) {try {Thread.sleep(1);//進行消息分發this.doReput();} catch (Exception e) {DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);} }DefaultMessageStore.log.info(this.getServiceName() + " service end");

ReputMessageService#doReput

this.reputFromOffset = result.getStartOffset();for (int readSize = 0; readSize < result.getSize() && doNext; ) {//從result中循環遍歷消息,一次讀一條,創建DispatchRequest對象。DispatchRequest dispatchRequest =DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();if (dispatchRequest.isSuccess()) {if (size > 0) {//---------------------------↓↓↓-------------------------//分發請求DefaultMessageStore.this.doDispatch(dispatchRequest);//當新消息達到 進行通知監聽器進行處理if (BrokerRole.SLAVE != DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole()&& DefaultMessageStore.this.brokerConfig.isLongPollingEnable()&& DefaultMessageStore.this.messageArrivingListener != null) {//messageArrivingListener對新發送的消息進行監聽DefaultMessageStore.this.messageArrivingListener.arriving(dispatchRequest.getTopic(),dispatchRequest.getQueueId(), dispatchRequest.getConsumeQueueOffset() + 1,dispatchRequest.getTagsCode(), dispatchRequest.getStoreTimestamp(),dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());}//更新消息分發偏移量 this.reputFromOffset += size;readSize += size;if (DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {DefaultMessageStore.this.storeStatsService.getSinglePutMessageTopicTimesTotal(dispatchRequest.getTopic()).add(1);DefaultMessageStore.this.storeStatsService.getSinglePutMessageTopicSizeTotal(dispatchRequest.getTopic()).add(dispatchRequest.getMsgSize());}} .... }

DispatchRequest

String topic; //消息主題名稱 int queueId; //消息隊列ID long commitLogOffset; //消息物理偏移量 int msgSize; //消息長度 long tagsCode; //消息過濾tag hashCode long storeTimestamp; //消息存儲時間戳 long consumeQueueOffset; //消息隊列偏移量 String keys; //消息索引key boolean success; //是否成功解析到完整的消息 String uniqKey; //消息唯一鍵 int sysFlag; //消息系統標記 long preparedTransactionOffset; //消息預處理事務偏移量 Map<String, String> propertiesMap; //消息屬性 byte[] bitMap; //位圖

DefaultMessageStore#doDispatch

for (CommitLogDispatcher dispatcher : this.dispatcherList) {dispatcher.dispatch(req); }

1.1.轉發到ConsumerQueue

CommitLogDispatcherBuildConsumeQueue#dispatch

final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag()); switch (tranType) {case MessageSysFlag.TRANSACTION_NOT_TYPE:case MessageSysFlag.TRANSACTION_COMMIT_TYPE://消息分發DefaultMessageStore.this.putMessagePositionInfo(request);break;case MessageSysFlag.TRANSACTION_PREPARED_TYPE:case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:break; }

DefaultMessageStore#putMessagePositionInfo

//根據請求中的主題和隊列ID獲得消費隊列 ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId()); //消息隊列分發消息 cq.putMessagePositionInfoWrapper(dispatchRequest);

ConsumeQueue#putMessagePositionInfoWrapper

boolean result = this.putMessagePositionInfo(request.getCommitLogOffset(),

ConsumeQueue#putMessagePositionInfo

private boolean putMessagePositionInfo(final long offset, //commitLog偏移量final int size, //消息體大小final long tagsCode, //消息tags的hashCodefinal long cqOffset ) { //寫入consumerqueue的偏移量//依次將消息偏移量、消息長度、tag寫入到ByteBuffer中this.byteBufferIndex.flip();this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);this.byteBufferIndex.putLong(offset);this.byteBufferIndex.putInt(size);this.byteBufferIndex.putLong(tagsCode);//計算期望插入ConsumerQueue的文件位置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());}//將消息追加到內存映射文件,異步輸盤->整個過程都是基于MappedFilereturn mappedFile.appendMessage(this.byteBufferIndex.array());} }

1.2.轉發到Index

核心實現類是IndexService，存儲Index文件的封裝類是IndexFile。

IndexFile

// 每個 hash 槽所占的字節數 private static int hashSlotSize = 4; //每個indexFile條目所占用字節數 private static int indexSize = 20; //用來驗證是否是一個有效索引 private static int invalidIndex = 0; //indexFile中hash槽的個數 private final int hashSlotNum; //indexFile中包含的條目數 private final int indexNum; //對應的MappedFile private final MappedFile mappedFile; //文件傳輸通道 private final FileChannel fileChannel; //pageCache private final MappedByteBuffer mappedByteBuffer; //每一個IndexFile的頭部信息 private final IndexHeader indexHeader;

消息分發到索引文件的時序圖：

CommitLogDispatcherBuildIndex#dispatch

//開啟文件索引 if (DefaultMessageStore.this.messageStoreConfig.isMessageIndexEnable()) {//構建索引DefaultMessageStore.this.indexService.buildIndex(request); }

IndexService#buildIndex

//創建或獲取索引文件 IndexFile indexFile = retryGetAndCreateIndexFile(); if (indexFile != null) {//獲得文件最大物理偏移量long endPhyOffset = indexFile.getEndPhyOffset();DispatchRequest msg = req;//消息主題String topic = msg.getTopic();//消息keyString keys = msg.getKeys();//如果該消息的物理偏移量小于索引文件中的最大物理偏移量,則說明是重復數據,忽略本次索引構建if (msg.getCommitLogOffset() < endPhyOffset) {return;}final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());switch (tranType) {case MessageSysFlag.TRANSACTION_NOT_TYPE:case MessageSysFlag.TRANSACTION_PREPARED_TYPE:case MessageSysFlag.TRANSACTION_COMMIT_TYPE:break;case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:return;}//如果消息ID不為空,則添加到Hash索引中if (req.getUniqKey() != null) {indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));if (indexFile == null) {log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());return;}}//構建索引key,RocketMQ支持為同一個消息建立多個索引,多個索引鍵空格隔開.if (keys != null && keys.length() > 0) {String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);for (int i = 0; i < keyset.length; i++) {String key = keyset[i];if (key.length() > 0) {//-----------↓-----------indexFile = putKey(indexFile, msg, buildKey(topic, key));if (indexFile == null) {log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());return;}}}} } else {log.error("build index error, stop building index"); }

IndexService#putKey

for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) {log.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one");//重試-獲取或者創建Index文件indexFile = retryGetAndCreateIndexFile();if (null == indexFile) {return null;}ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); }return indexFile;

IndexFile#putKey

public boolean putKey(final String key, final long phyOffset, //消息存儲在commitLog的偏移量final long storeTimestamp) { //消息存入commitLog的時間戳 //如果IndexFile存儲的條目數小于最大條目數限制-允許存儲//否則表示存儲失敗-返回falseif (this.indexHeader.getIndexCount() < this.indexNum) { int keyHash = indexKeyHashMethod(key);int slotPos = keyHash % this.hashSlotNum; //根據keyHash和hash槽數量取模->得到該key在槽中的下標int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;FileLock fileLock = null;try {// fileLock = this.fileChannel.lock(absSlotPos, hashSlotSize,// false);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);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; }

這里涉及到索引文件的存儲格式和查找，暫不在此做解析。

---

2.消息隊列和索引文件恢復

由于RocketMQ存儲首先將消息全量存儲在CommitLog文件中，然后異步生成轉發任務更新ConsumerQueue和Index文件。如果消息成功存儲到CommitLog文件中，轉發任務未成功執行，此時消息服務器Broker由于某種原因宕機，導致CommitLog、ConsumerQueue、IndexFile文件數據不一致。如果不加以人工修復的話，會有一部分消息即便在CommitLog中文件中存在，但由于沒有轉發到ConsumerQueue，這部分消息將永遠不被消費者消費。

2.1.存儲文件加載

判斷上一次是否是異常退出——實現機制是Broker在啟動時會創建abort文件，在退出時會通過JVM鉤子函數刪除abort文件。如果下次啟動時發現abort仍然存在，則說明Broker上次退出是異常的，CommitLog和ConsumerQueue數據可能存在不一致的情況，需要進行文件修復。

//檢查上次退出是否異常 boolean lastExitOK = !this.isTempFileExist();↓↓↓ private boolean isTempFileExist() {//嘗試獲取abort臨時文件 String fileName = StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir());File file = new File(fileName);//如果存在即說明上次退出異常 return file.exists(); }

DefaultMessageStore#load

//加載結果 boolean result = true;try {//判斷是否是正常退出boolean lastExitOK = !this.isTempFileExist();log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally");//1.加載commitLog文件result = result && this.commitLog.load();//2.加載consumeQueue文件result = result && this.loadConsumeQueue();if (result) {//加載store目錄下的所有存儲文件this.storeCheckpoint =new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));//3.加載index文件this.indexService.load(lastExitOK);//4.根據broker是否異常退出采取不同的恢復策略this.recover(lastExitOK);log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset());if (null != scheduleMessageService) {result = this.scheduleMessageService.load();}}} catch (Exception e) {log.error("load exception", e);result = false; } //如果加載失敗->關閉MappedFile分配服務線程 if (!result) {this.allocateMappedFileService.shutdown(); }return result;

2.1.1.加載commitLog文件

commitLog.load()→mappedFileQueue.load()

MappedFileQueue#load

//根據文件目錄獲取文件 File dir = new File(this.storePath); //獲得文件數組 File[] ls = dir.listFiles(); if (ls != null) {return doLoad(Arrays.asList(ls)); } return true;

MappedFileQueue#doLoad

//文件進行排序 files.sort(Comparator.comparing(File::getName)); //遍歷文件列表 for (File file : files) {//如果文件大小與配置文件不一致 退出if (file.length() != this.mappedFileSize) {log.warn(file + "\t" + file.length()+ " length not matched message store config value, ignore it");return true;}try {//創建映射文件MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);//設置映射文件的指針mappedFile.setWrotePosition(this.mappedFileSize);mappedFile.setFlushedPosition(this.mappedFileSize);mappedFile.setCommittedPosition(this.mappedFileSize);//將映射文件加到隊列中this.mappedFiles.add(mappedFile);log.info("load " + file.getPath() + " OK");} catch (IOException e) {log.error("load file " + file + " error", e);return false;} } return true;

2.1.2.加載消息消費隊列

DefaultMessageStore#loadConsumeQueue

//獲得消費隊列目錄 File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir())); File[] fileTopicList = dirLogic.listFiles(); if (fileTopicList != null) {//遍歷消費隊列目錄for (File fileTopic : fileTopicList) {//獲得子目錄名稱，即為topic名稱String topic = fileTopic.getName();//遍歷子目錄下的消費隊列文件File[] fileQueueIdList = fileTopic.listFiles();if (fileQueueIdList != null) {//遍歷文件for (File fileQueueId : fileQueueIdList) {int queueId;try {queueId = Integer.parseInt(fileQueueId.getName());} catch (NumberFormatException e) {continue;}//創建消費目錄并加載到內存中ConsumeQueue logic = new ConsumeQueue(topic,queueId,StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),this);//存入consumeQueueTable消息隊列表this.putConsumeQueue(topic, queueId, logic);if (!logic.load()) {return false;}}}} } //打印加載消費隊列成功日志 log.info("load logics queue all over, OK");return true;

2.1.3.加載索引文件

IndexService#load

//根據索引文件目錄加載文件 File dir = new File(this.storePath); //獲取文件數組 File[] files = dir.listFiles(); if (files != null) {// ascending order//文件排序Arrays.sort(files);//文件遍歷for (File file : files) {try {//創建索引文件IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);//加載f.load();//如果異常退出if (!lastExitOK) {//索引文件上次刷盤時間大于檢測點時間戳if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint().getIndexMsgTimestamp()) {//該文件立刻刪除f.destroy(0);continue;}}log.info("load index file OK, " + f.getFileName());//將符合條件索引文件加入列表this.indexFileList.add(f);} catch (IOException e) {log.error("load file {} error", file, e);return false;} catch (NumberFormatException e) {log.error("load file {} error", file, e);}} }return true;

所有文件均加載完畢，進行文件恢復，根據Broker是否正常退出執行不同的恢復策略。

DefaultMessageStore#recover

long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();if (lastExitOK) {//正常恢復this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue); } else {//異常恢復this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue); } //在CommitLog中保存每個消息消費隊列當前的存儲邏輯偏移量 this.recoverTopicQueueTable();

DefaultMessageStore#recoverConsumeQueue

long maxPhysicOffset = -1; for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {for (ConsumeQueue logic : maps.values()) {//覆蓋消息隊列logic.recover();if (logic.getMaxPhysicOffset() > maxPhysicOffset) {maxPhysicOffset = logic.getMaxPhysicOffset();}} }return maxPhysicOffset;

ConsumeQueue#recover

//獲取消息隊列的所有映射文件 final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles(); if (!mappedFiles.isEmpty()) {int index = mappedFiles.size() - 3; //從倒數第三個文件開始if (index < 0) {index = 0;}//consumerQueue邏輯大小int mappedFileSizeLogics = this.mappedFileSize;//consumerQueue對應的映射文件MappedFile mappedFile = mappedFiles.get(index);//映射文件對應的ByteBufferByteBuffer byteBuffer = mappedFile.sliceByteBuffer();//處理的offset-默認從consumerQueue中存放的第一個條目開始long processOffset = mappedFile.getFileFromOffset();long mappedFileOffset = 0;long maxExtAddr = 1;while (true) {//循環驗證consumerQueue包含條目的有效性for (int i = 0; i < mappedFileSizeLogics; i += CQ_STORE_UNIT_SIZE) {//讀取條目內容//commitLog物理偏移量long offset = byteBuffer.getLong();//消息總長度int size = byteBuffer.getInt();//tag哈希值long tagsCode = byteBuffer.getLong();//offset大于0并且size大于0說明該條目是一個有效的if (offset >= 0 && size > 0) {mappedFileOffset = i + CQ_STORE_UNIT_SIZE;this.maxPhysicOffset = offset + size;if (isExtAddr(tagsCode)) {maxExtAddr = tagsCode;}} else {log.info("recover current consume queue file over, " + mappedFile.getFileName() + " "+ offset + " " + size + " " + tagsCode);break;}}//如果該consumeQueue中包含的條目全部有效則繼續驗證下一個文件-index++if (mappedFileOffset == mappedFileSizeLogics) {index++;//驗證完畢-退出-開始恢復消息隊列文件if (index >= mappedFiles.size()) {log.info("recover last consume queue file over, last mapped file "+ mappedFile.getFileName());break;} else {//獲取下一個映射文件-繼續循環檢查mappedFile = mappedFiles.get(index);byteBuffer = mappedFile.sliceByteBuffer();processOffset = mappedFile.getFileFromOffset();mappedFileOffset = 0;log.info("recover next consume queue file, " + mappedFile.getFileName());}} else {log.info("recover current consume queue queue over " + mappedFile.getFileName() + " "+ (processOffset + mappedFileOffset));break;}}//當前consumerQueue有效偏移量processOffset += mappedFileOffset;//設置flushWhere、committedWhere為有效偏移量this.mappedFileQueue.setFlushedWhere(processOffset);this.mappedFileQueue.setCommittedWhere(processOffset);//刪除冗余無效的consumerQueue文件this.mappedFileQueue.truncateDirtyFiles(processOffset);if (isExtReadEnable()) {this.consumeQueueExt.recover();log.info("Truncate consume queue extend file by max {}", maxExtAddr);this.consumeQueueExt.truncateByMaxAddress(maxExtAddr);} }

DefaultMessageStore#recoverTopicQueueTable

HashMap<String/* topic-queueid */, Long/* offset */> table = new HashMap<String, Long>(1024); //CommitLog最小偏移量 long minPhyOffset = this.commitLog.getMinOffset(); //遍歷消費隊列,將消費隊列保存在CommitLog中 for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {for (ConsumeQueue logic : maps.values()) {String key = logic.getTopic() + "-" + logic.getQueueId();table.put(key, logic.getMaxOffsetInQueue());logic.correctMinOffset(minPhyOffset);} } //覆蓋主題消息隊列信息 this.commitLog.setTopicQueueTable(table);

2.2.正常恢復

CommitLog恢復過程與ConsumeQueue恢復過程極其相似。

CommitLog#recoverNormally

boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover(); //獲取映射文件列表 final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles(); if (!mappedFiles.isEmpty()) {//Broker正常停止再重啟時,從倒數第三個開始恢復,如果不足3個文件,則從第一個文件開始恢復。int index = mappedFiles.size() - 3;if (index < 0) {index = 0;}MappedFile mappedFile = mappedFiles.get(index);ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();long processOffset = mappedFile.getFileFromOffset();//代表當前已校驗通過的offsetlong mappedFileOffset = 0;while (true) {//檢查消息并返回分發請求DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);//消息長度int size = dispatchRequest.getMsgSize();//查找結果為true,并且消息長度大于0,表示消息正確.mappedFileOffset向前移動本消息長度if (dispatchRequest.isSuccess() && size > 0) {mappedFileOffset += size;}//如果查找結果為true且消息長度等于0,表示已到該文件末尾,如果還有下一個文件,則重置processOffset和MappedFileOffset重復查找下一個文件,否則跳出循環。else if (dispatchRequest.isSuccess() && size == 0) {index++;//完成檢查-退出循環if (index >= mappedFiles.size()) {// Current branch can not happenlog.info("recover last 3 physics file over, last mapped file " + mappedFile.getFileName());break;} else {//獲取并檢查下一個映射文件mappedFile = mappedFiles.get(index);byteBuffer = mappedFile.sliceByteBuffer();processOffset = mappedFile.getFileFromOffset();//重置檢查offsetmappedFileOffset = 0;log.info("recover next physics file, " + mappedFile.getFileName());}}// Intermediate file read errorelse if (!dispatchRequest.isSuccess()) {log.info("recover physics file end, " + mappedFile.getFileName());break;}}//更新MappedFileQueue的flushedWhere和committedWhere指針processOffset += mappedFileOffset;this.mappedFileQueue.setFlushedWhere(processOffset);this.mappedFileQueue.setCommittedWhere(processOffset);//刪除之后的冗余無效的文件this.mappedFileQueue.truncateDirtyFiles(processOffset);// Clear ConsumeQueue redundant dataif (maxPhyOffsetOfConsumeQueue >= processOffset) {log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);} } else {// Commitlog case files are deletedlog.warn("The commitlog files are deleted, and delete the consume queue files");this.mappedFileQueue.setFlushedWhere(0);this.mappedFileQueue.setCommittedWhere(0);this.defaultMessageStore.destroyLogics();

MappedFileQueue#truncateDirtyFiles

//即將移除文件 List<MappedFile> willRemoveFiles = new ArrayList<MappedFile>(); //遍歷映射文件 for (MappedFile file : this.mappedFiles) {long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;//文件尾部的偏移量大于offsetif (fileTailOffset > offset) {//offset大于文件的起始偏移量if (offset >= file.getFileFromOffset()) {//更新wrotePosition、committedPosition、flushedPosistionfile.setWrotePosition((int) (offset % this.mappedFileSize));file.setCommittedPosition((int) (offset % this.mappedFileSize));file.setFlushedPosition((int) (offset % this.mappedFileSize));} else {//offset小于文件的起始偏移量,說明該文件是有效文件后面創建的,釋放mappedFile占用內存,刪除文件file.destroy(1000);willRemoveFiles.add(file);}} }this.deleteExpiredFile(willRemoveFiles);

2.3.異常恢復

Broker異常停止文件恢復的實現為CommitLog#recoverAbnormally。異常文件恢復步驟與正常停止文件恢復流程基本相同，其主要差別有兩個。首先，正常停止默認從倒數第三個文件開始進行恢復，而異常停止則需要從最后一個文件往前走，找到第一個消息存儲正常的文件。其次，如果CommitLog目錄沒有消息文件，如果消息消費隊列目錄下存在文件，則需要銷毀。

CommitLog#recoverAbnormally

boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover(); final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles(); if (!mappedFiles.isEmpty()) {// Looking beginning to recover from which file//從最后一個文件開始檢查int index = mappedFiles.size() - 1;MappedFile mappedFile = null;for (; index >= 0; index--) {mappedFile = mappedFiles.get(index);//如果該文件是一個正確的文件if (this.isMappedFileMatchedRecover(mappedFile)) {log.info("recover from this mapped file " + mappedFile.getFileName());break;}}//恰好正確文件是第一個-<index-->之后index小于0if (index < 0) {//需要將索引校正為0index = 0;//取出第一個文件-從第一個文件開始恢復mappedFile = mappedFiles.get(index);}ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();long processOffset = mappedFile.getFileFromOffset();long mappedFileOffset = 0;while (true) {//驗證消息是否合法DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);int size = dispatchRequest.getMsgSize();//消息校驗成功if (dispatchRequest.isSuccess()) {// 正qif (size > 0) {mappedFileOffset += size;//將消息轉發到消息消費隊列和索引文件if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {this.defaultMessageStore.doDispatch(dispatchRequest);}} else {this.defaultMessageStore.doDispatch(dispatchRequest);}}//到達文件尾else if (size == 0) {index++;//文件索引向后移動//文件遍歷完畢if (index >= mappedFiles.size()) {// The current branch under normal circumstances should// not happenlog.info("recover physics file over, last mapped file " + mappedFile.getFileName());break;} else {//獲取下一個的映射文件mappedFile = mappedFiles.get(index);byteBuffer = mappedFile.sliceByteBuffer();processOffset = mappedFile.getFileFromOffset();mappedFileOffset = 0;log.info("recover next physics file, " + mappedFile.getFileName());}}} else {log.info("recover physics file end, " + mappedFile.getFileName() + " pos=" + byteBuffer.position());break;}}//當前有效偏移量processOffset += mappedFileOffset;//設置flushWhere、committedWhere為有效偏移量this.mappedFileQueue.setFlushedWhere(processOffset);this.mappedFileQueue.setCommittedWhere(processOffset);//清除冗余的數據this.mappedFileQueue.truncateDirtyFiles(processOffset);//清除冗余的數據if (maxPhyOffsetOfConsumeQueue >= processOffset) {log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);} } //文件被刪除 else {log.warn("The commitlog files are deleted, and delete the consume queue files");//flushWhere committedWhere重置this.mappedFileQueue.setFlushedWhere(0);this.mappedFileQueue.setCommittedWhere(0);//銷毀消息隊列文件this.defaultMessageStore.destroyLogics(); }

本文僅作為個人學習使用，如有不足或錯誤請指正！

總結

以上是生活随笔為你收集整理的RocketMQ：消息消费队列与索引文件的实时更新以及文件恢复源码解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。