文章目錄

• • 消息存儲機制
  • • 1.前言
    • 2.核心存儲類：DefaultMessageStore
    • 3.消息存儲流程
    • 4.消息存儲文件
    • 5.存儲文件內存映射
    • • 5.1.MapperFileQueue
      • 5.2.MappedFile
      • • • 5.2.1.commit
          • 5.2.2.flush
      • 5.3.TransientStorePool
    • 6.刷盤機制
    • • • 6.1.同步刷盤
        • 6.2.異步刷盤

消息存儲機制

1.前言

本文主要講解內容是Broker接收到消息生產者發送的消息之后，如何將消息持久化存儲在Broker中。

2.核心存儲類：DefaultMessageStore

private final MessageStoreConfig messageStoreConfig; //消息配置屬性 private final CommitLog commitLog; //CommitLog文件存儲的實現類->消息存儲在commitLog中 private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable; //消息隊列存儲緩存表,按照消息主題分組 private final FlushConsumeQueueService flushConsumeQueueService; //消息隊列文件刷盤服務線程 private final CleanCommitLogService cleanCommitLogService; //過期CommitLog文件刪除服務 private final CleanConsumeQueueService cleanConsumeQueueService; //過期ConsumerQueue隊列文件刪除服務 private final IndexService indexService; //索引服務 private final AllocateMappedFileService allocateMappedFileService; //MappedFile分配服務->內存映射處理commitLog、consumerQueue文件 private final ReputMessageService reputMessageService;//CommitLog消息分發,根據CommitLog文件構建ConsumerQueue、IndexFile文件 private final HAService haService; //消息主從同步實現服務 private final ScheduleMessageService scheduleMessageService; //消息服務調度服務 private final StoreStatsService storeStatsService; //消息存儲服務 private final MessageArrivingListener messageArrivingListener; //消息到達監聽器 private final TransientStorePool transientStorePool; //消息堆外內存緩存 private final BrokerStatsManager brokerStatsManager; //Broker狀態管理器 private final MessageArrivingListener messageArrivingListener; //消息拉取長輪詢模式消息達到監聽器 private final BrokerConfig brokerConfig; //Broker配置類 private StoreCheckpoint storeCheckpoint; //文件刷盤監測點 private final LinkedList<CommitLogDispatcher> dispatcherList; //CommitLog文件轉發請求

以上屬性是消息存儲的核心，需要重點關注每個屬性的具體作用。

3.消息存儲流程

消息存儲時序圖如下：

消息存儲入口：DefaultMessageStore#putMessage

//檢查Broker是否是Slave || 判斷當前寫入狀態如果是正在寫入,則不能繼續 PutMessageStatus checkStoreStatus = this.checkStoreStatus(); if (checkStoreStatus != PutMessageStatus.PUT_OK) {return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus, null)); }//檢查消息主題和消息體長度是否合法 PutMessageStatus msgCheckStatus = this.checkMessages(messageExtBatch); if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null)); } //記錄開始寫入時間 long beginTime = this.getSystemClock().now(); //寫入消息 CompletableFuture<PutMessageResult> resultFuture = this.commitLog.asyncPutMessages(messageExtBatch);resultFuture.thenAccept((result) -> {long elapsedTime = this.getSystemClock().now() - beginTime;if (elapsedTime > 500) {log.warn("not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, messageExtBatch.getBody().length);}//記錄相關統計信息this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);//存儲失敗if (null == result || !result.isOk()) {//存儲狀態服務->消息存儲失敗次數自增this.storeStatsService.getPutMessageFailedTimes().add(1);} });return resultFuture;

DefaultMessageStore#checkStoreStatus

//存儲服務已停止 if (this.shutdown) {log.warn("message store has shutdown, so putMessage is forbidden");return PutMessageStatus.SERVICE_NOT_AVAILABLE; } //Broker為Slave->不可寫入 if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {long value = this.printTimes.getAndIncrement();if ((value % 50000) == 0) {log.warn("broke role is slave, so putMessage is forbidden");}return PutMessageStatus.SERVICE_NOT_AVAILABLE; }//不可寫入->broker磁盤已滿/寫入邏輯隊列錯誤/寫入索引文件錯誤 if (!this.runningFlags.isWriteable()) {long value = this.printTimes.getAndIncrement();if ((value % 50000) == 0) {log.warn("the message store is not writable. It may be caused by one of the following reasons: " +"the broker's disk is full, write to logic queue error, write to index file error, etc");}return PutMessageStatus.SERVICE_NOT_AVAILABLE; } else {this.printTimes.set(0); } //操作系統頁寫入是否繁忙 if (this.isOSPageCacheBusy()) {return PutMessageStatus.OS_PAGECACHE_BUSY; } return PutMessageStatus.PUT_OK;

CommitLog#asyncPutMessages

//記錄消息存儲時間 messageExtBatch.setStoreTimestamp(System.currentTimeMillis()); AppendMessageResult result;StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();final int tranType = MessageSysFlag.getTransactionValue(messageExtBatch.getSysFlag());//消息類型是否合法 if (tranType != MessageSysFlag.TRANSACTION_NOT_TYPE) {return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null)); }//....//獲取上一個MapperFile對象->內存映射的具體實現 MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();//追加消息需要加鎖->串行化處理 putMessageLock.lock(); try {long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();this.beginTimeInLock = beginLockTimestamp;//記錄消息存儲時間->保證消息的有序性messageExtBatch.setStoreTimestamp(beginLockTimestamp);//判斷如果mappedFile如果為空或者已滿,創建新的mappedFile文件if (null == mappedFile || mappedFile.isFull()) {mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise}//如果創建失敗,直接返回if (null == mappedFile) {log.error("Create mapped file1 error, topic: {} clientAddr: {}", messageExtBatch.getTopic(), messageExtBatch.getBornHostString());beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));}//!!!寫入消息到mappedFile中!!!result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback, putMessageContext);//根據寫入結果做不同的處理switch (result.getStatus()) {case PUT_OK:break;case END_OF_FILE:unlockMappedFile = mappedFile;// Create a new file, re-write the messagemappedFile = this.mappedFileQueue.getLastMappedFile(0);if (null == mappedFile) {// XXX: warn and notify melog.error("Create mapped file2 error, topic: {} clientAddr: {}", messageExtBatch.getTopic(), messageExtBatch.getBornHostString());beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));}result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback, putMessageContext);break;case MESSAGE_SIZE_EXCEEDED:case PROPERTIES_SIZE_EXCEEDED:beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));case UNKNOWN_ERROR: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]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, messageExtBatch.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(messageExtBatch.getTopic()).add(result.getMsgNum()); storeStatsService.getSinglePutMessageTopicSizeTotal(messageExtBatch.getTopic()).add(result.getWroteBytes());//根據刷盤策略進行刷盤 CompletableFuture<PutMessageStatus> flushOKFuture = submitFlushRequest(result, messageExtBatch); //主從同步 CompletableFuture<PutMessageStatus> replicaOKFuture = submitReplicaRequest(result, messageExtBatch);

MappedFile#appendMessagesInner

assert messageExt != null; assert cb != null;//獲取寫指針/寫入位置 int currentPos = this.wrotePosition.get();//寫指針偏移量小于文件指定大小 if (currentPos < this.fileSize) {//寫入緩沖區ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();byteBuffer.position(currentPos);AppendMessageResult result;//根據消息類型->批量/單個->進行不同處理if (messageExt instanceof MessageExtBrokerInner) {//單個消息//調用回調方法寫入磁盤->CommitLog#doAppendresult = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,(MessageExtBrokerInner) messageExt, putMessageContext);} else if (messageExt instanceof MessageExtBatch) {//批量消息result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,(MessageExtBatch) messageExt, putMessageContext);} 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);

CommitLog#doAppend

public AppendMessageResult doAppend(final long fileFromOffset, //文件序列偏移量final ByteBuffer byteBuffer, //NIO字節容器final int maxBlank, //最大可寫入字節數 final MessageExtBrokerInner msgInner, //消息封裝實體PutMessageContext putMessageContext) {//文件寫入偏移量long wroteOffset = fileFromOffset + byteBuffer.position();//構建msgIdSupplier<String> msgIdSupplier = () -> {//系統標識int sysflag = msgInner.getSysFlag();//msgId底層存儲由16個字節組成int msgIdLen = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;//分配16個字節的存儲空間ByteBuffer msgIdBuffer = ByteBuffer.allocate(msgIdLen);//8個字節->ip、host各占用4個字節MessageExt.socketAddress2ByteBuffer(msgInner.getStoreHost(), msgIdBuffer);//清除緩沖區->因為接下來需要翻轉緩沖區msgIdBuffer.clear();//剩下的8個字節用來存儲commitLog偏移量-wroteOffsetmsgIdBuffer.putLong(msgIdLen - 8, wroteOffset);return UtilAll.bytes2string(msgIdBuffer.array());};//獲取當前主題消息隊列唯一keyString key = putMessageContext.getTopicQueueTableKey();//根據key獲取消息存儲偏移量Long queueOffset = CommitLog.this.topicQueueTable.get(key);if (null == queueOffset) {queueOffset = 0L;CommitLog.this.topicQueueTable.put(key, queueOffset);}// Transaction messages that require special handlingfinal int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());switch (tranType) {// Prepared and Rollback message is not consumed, will not enter the// consumer queueccase MessageSysFlag.TRANSACTION_PREPARED_TYPE:case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:queueOffset = 0L;break;case MessageSysFlag.TRANSACTION_NOT_TYPE:case MessageSysFlag.TRANSACTION_COMMIT_TYPE:default:break;}ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();//計算消息存儲長度final int msgLen = preEncodeBuffer.getInt(0);// Determines whether there is sufficient free space//消息是如果沒有足夠的存儲空間則新創建CommitLog文件if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {this.msgStoreItemMemory.clear();// 1 TOTALSIZEthis.msgStoreItemMemory.putInt(maxBlank);// 2 MAGICCODEthis.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);// 3 The remaining space may be any value// Here the length of the specially set maxBlankfinal long beginTimeMills = CommitLog.this.defaultMessageStore.now();byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset,maxBlank, /* only wrote 8 bytes, but declare wrote maxBlank for compute write position */msgIdSupplier, msgInner.getStoreTimestamp(),queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);}int pos = 4 + 4 + 4 + 4 + 4;// 6 QUEUEOFFSETpreEncodeBuffer.putLong(pos, queueOffset);pos += 8;// 7 PHYSICALOFFSETpreEncodeBuffer.putLong(pos, fileFromOffset + byteBuffer.position());int ipLen = (msgInner.getSysFlag() & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;// 8 SYSFLAG, 9 BORNTIMESTAMP, 10 BORNHOST, 11 STORETIMESTAMPpos += 8 + 4 + 8 + ipLen;// refresh store time stamp in lockpreEncodeBuffer.putLong(pos, msgInner.getStoreTimestamp());final long beginTimeMills = CommitLog.this.defaultMessageStore.now();// Write messages to the queue buffer//將消息存儲到byteBuffer中byteBuffer.put(preEncodeBuffer);msgInner.setEncodedBuff(null);//返回AppendMessageResultAppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgIdSupplier,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:// The next update ConsumeQueue informationCommitLog.this.topicQueueTable.put(key, ++queueOffset);break;default:break;}return result; }

AppendMessageResult

public class AppendMessageResult {private AppendMessageStatus status; //消息追加結果private long wroteOffset; //消息寫入偏移量 private int wroteBytes; //消息待寫入字節private String msgId; //消息ID private Supplier<String> msgIdSupplier; //消息IDprivate long storeTimestamp; //消息寫入時間戳private long logicsOffset; //消息隊列偏移量private long pagecacheRT = 0; //消息開始寫入時間戳 }

返回消息寫入結果，回到CommitLog#asyncPutMessages

result = mappedFile.appendMessages(messageExtBatch, this.appendMessageCallback, putMessageContext); switch (result.getStatus()) {case PUT_OK:break; } //釋放鎖 putMessageLock.unlock(); //存儲數據統計 storeStatsService.getSinglePutMessageTopicTimesTotal(messageExtBatch.getTopic()).add(result.getMsgNum()); storeStatsService.getSinglePutMessageTopicSizeTotal(messageExtBatch.getTopic()).add(result.getWroteBytes());//根據刷盤策略進行刷盤 CompletableFuture<PutMessageStatus> flushOKFuture = submitFlushRequest(result, messageExtBatch); //消息主從同步 CompletableFuture<PutMessageStatus> replicaOKFuture = submitReplicaRequest(result, messageExtBatch);

4.消息存儲文件

• commitLog：消息存儲目錄
• config：配置信息
• consumerqueue：消息隊列存儲目錄
• index：消息索引文件存儲目錄
• abort：Broker異常關閉時信息記錄
• checkpoint：文件監測點，存儲commitlog、consumerqueue、index文件最后一次刷盤時間戳。

5.存儲文件內存映射

RocketMQ通過使用內存映射文件提高IO訪問性能，無論是CommitLog、ConsumerQueue還是IndexFile，單個文件都被設計為固定長度。

如果一個文件寫滿以后再創建一個新文件，文件名就為該文件第一條消息對應的全局物理偏移量，如下圖所示。

5.1.MapperFileQueue

//存儲目錄 private final String storePath;//單個文件大小 protected final int mappedFileSize;//MappedFile文件集合 protected final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();//映射文件MapperFile分配服務線程 private final AllocateMappedFileService allocateMappedFileService;//刷盤指針 protected long flushedWhere = 0;//當前數據提交指針 private long committedWhere = 0;

根據存儲時間獲取對應的MappedFile

public MappedFile getMappedFileByTime(final long timestamp) {//拷貝映射文件Object[] mfs = this.copyMappedFiles(0);if (null == mfs) {return null;}//遍歷映射文件數組for (int i = 0; i < mfs.length; i++) {MappedFile mappedFile = (MappedFile) mfs[i];//MappedFile的最后修改時間大于指定時間戳->返回該文件if (mappedFile.getLastModifiedTimestamp() >= timestamp) {return mappedFile;}}return (MappedFile) mfs[mfs.length - 1]; }

根據消息存儲偏移量查找MappedFile

public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {try {//分別獲取第一個和最后一個映射文件MappedFile firstMappedFile = this.getFirstMappedFile();MappedFile lastMappedFile = this.getLastMappedFile();//第一個文件和最后一個文件均不為空,則進行處理if (firstMappedFile != null && lastMappedFile != null) {if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",offset,firstMappedFile.getFileFromOffset(),lastMappedFile.getFileFromOffset() + this.mappedFileSize,this.mappedFileSize,this.mappedFiles.size());} else {//獲得文件索引int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));//目標映射文件MappedFile targetFile = null;try {//根據文件索引查找目標文件targetFile = this.mappedFiles.get(index);} catch (Exception ignored) {}//對獲取到的映射文件進行檢查-判空-偏移量是否合法if (targetFile != null && offset >= targetFile.getFileFromOffset()&& offset < targetFile.getFileFromOffset() + this.mappedFileSize) {return targetFile;}//繼續選擇映射文件for (MappedFile tmpMappedFile : this.mappedFiles) {if (offset >= tmpMappedFile.getFileFromOffset()&& offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {return tmpMappedFile;}}}//返回第一個映射文件if (returnFirstOnNotFound) {return firstMappedFile;}}} catch (Exception e) {log.error("findMappedFileByOffset Exception", e);}return null; }

獲取存儲文件最小偏移量

public long getMinOffset() {if (!this.mappedFiles.isEmpty()) {try {return this.mappedFiles.get(0).getFileFromOffset();} catch (IndexOutOfBoundsException e) {//continue;} catch (Exception e) {log.error("getMinOffset has exception.", e);}}return -1; }

獲取存儲文件最大偏移量

public long getMaxOffset() {//最后一個映射文件MappedFile mappedFile = getLastMappedFile();if (mappedFile != null) {return mappedFile.getFileFromOffset() + mappedFile.getReadPosition();}return 0; }

獲取存儲文件當前寫指針

public long getMaxWrotePosition() {MappedFile mappedFile = getLastMappedFile();if (mappedFile != null) {return mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();}return 0; }

5.2.MappedFile

//操作系統每頁刷寫大小,默認4K public static final int OS_PAGE_SIZE = 1024 * 4; //當前JVM實例中MappedFile虛擬內存 private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0); //當前JVM實例中MappedFile對象個數 private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0); //當前文件的寫指針 protected final AtomicInteger wrotePosition = new AtomicInteger(0); //當前文件的提交指針 protected final AtomicInteger committedPosition = new AtomicInteger(0); //刷盤指針 private final AtomicInteger flushedPosition = new AtomicInteger(0); //文件大小 protected int fileSize; //文件通道 protected FileChannel fileChannel;/*** 堆外內存ByteBuffer* Message will put to here first, and then reput to FileChannel if writeBuffer is not null.*/ protected ByteBuffer writeBuffer = null; //堆外內存池 protected TransientStorePool transientStorePool = null; //文件名稱 private String fileName; //該文件的處理偏移量 private long fileFromOffset; //物理文件 private File file; //文件映射緩沖區 private MappedByteBuffer mappedByteBuffer; //存儲時間戳 private volatile long storeTimestamp = 0; //是否是初次創建 private boolean firstCreateInQueue = false;

MappedFile初始化

private void init(final String fileName, final int fileSize) throws IOException {this.fileName = fileName;this.fileSize = fileSize;this.file = new File(fileName);this.fileFromOffset = Long.parseLong(this.file.getName());boolean ok = false;//確保文件目錄正確ensureDirOK(this.file.getParent());try {this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);TOTAL_MAPPED_FILES.incrementAndGet();ok = true;} catch (FileNotFoundException e) {log.error("Failed to create file " + this.fileName, e);throw e;} catch (IOException e) {log.error("Failed to map file " + this.fileName, e);throw e;} finally {if (!ok && this.fileChannel != null) {this.fileChannel.close();}} }

值得注意的是MappedFile還有一個屬性值transientStorePoolEnable，當這個屬性值為true時，數據會先存儲到對外內存，如何通過commit線程將數據提交到內存映射buffer中，最后通過flush線程將內存映射刷寫到磁盤中。

開啟transientStorePoolEnable

public void init(final String fileName, final int fileSize,final TransientStorePool transientStorePool) throws IOException {init(fileName, fileSize);//初始化對外內存緩沖區this.writeBuffer = transientStorePool.borrowBuffer();this.transientStorePool = transientStorePool; }

5.2.1.commit

刷盤文件提交流程大致如下：

DefaultMessageStore#flush→CommitLog→MappedFileQueue→MappedFile

//DefaultMessageStore public long flush() {return this.commitLog.flush(); } //CommitLog public long flush() {//----------↓-----------this.mappedFileQueue.commit(0);this.mappedFileQueue.flush(0);return this.mappedFileQueue.getFlushedWhere(); } //MappedFileQueue public boolean commit(final int commitLeastPages) {boolean result = true;MappedFile mappedFile = this.findMappedFileByOffset(this.committedWhere, this.committedWhere == 0);if (mappedFile != null) {//----------↓-----------int offset = mappedFile.commit(commitLeastPages);long where = mappedFile.getFileFromOffset() + offset;result = where == this.committedWhere;this.committedWhere = where;}return result; }

最后進入MappedFile進行數據刷寫提交：

MappedFile#commit

public int commit(final int commitLeastPages) {//如果為空->說明沒有開啟transientStorePoolEnable->無需向文件通道fileChannel提交數據 //將wrotePosition視為committedPosition并返回->然后直接進行flush操作if (writeBuffer == null) {return this.wrotePosition.get();}//提交數據頁數大于commitLeastPagesif (this.isAbleToCommit(commitLeastPages)) {//MappedFile是否被銷毀//hold()->isAvailable()->MappedFile.available<屬性繼承于ReferenceResource>//文件如何被摧毀可見下文中的shutdown()if (this.hold()) {//--↓--commit0();this.release();} else {log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());}}// All dirty data has been committed to FileChannel.// 所有數據提交后,清空緩沖區if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {this.transientStorePool.returnBuffer(writeBuffer);this.writeBuffer = null;}return this.committedPosition.get(); }

MappedFile#isAbleToCommit

//已提交刷盤的指針 int flush = this.committedPosition.get(); //文件寫指針 int write = this.wrotePosition.get();//刷盤已寫滿 if (this.isFull()) {return true; }if (commitLeastPages > 0) {//文件內容達到commitLeastPages->進行刷盤return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= commitLeastPages; } return write > flush;

MappedFile#commit0

//寫指針 int writePos = this.wrotePosition.get(); //上次提交指針 int lastCommittedPosition = this.committedPosition.get(); //寫指針一定要大于上次提交指針 if (writePos - lastCommittedPosition > 0) {try {//復制共享內存區域ByteBuffer byteBuffer = writeBuffer.slice();//設置提交位置是上次提交位置byteBuffer.position(lastCommittedPosition);//最大提交數量byteBuffer.limit(writePos);//設置fileChannel位置是上次提交位置this.fileChannel.position(lastCommittedPosition);//將lastCommittedPosition到writePos的數據復制到FileChannel中this.fileChannel.write(byteBuffer);//重置提交位置為writePos->以此反復避免提交重復數據this.committedPosition.set(writePos);} catch (Throwable e) {log.error("Error occurred when commit data to FileChannel.", e);} }

5.2.2.flush

刷寫磁盤，直接調用MappedByteBuffer或fileChannel的force方法將內存中的數據持久化到磁盤，那么flushedPosition應該等于MappedByteBuffer中的寫指針；

• 如果writeBuffer不為空，則flushPosition應該等于上一次的commit指針；因為上一次提交的數據就是進入到MappedByteBuffer中的數據；
• 如果writeBuffer為空，數據時直接進入到MappedByteBuffer，wrotePosition代表的是MappedByteBuffer中的指針，故設置flushPosition為wrotePosition。

提交數據到fileChannel后開始刷盤，步驟如下：

CommitLog#flush→MappedFileQueue#flush→MappedFile#flush

MappedFile#flush

//達到刷盤條件 if (this.isAbleToFlush(flushLeastPages)) {//加鎖，同步刷盤if (this.hold()) {//讀指針int value = getReadPosition();try {//開啟TransientStorePool->fileChannel//關閉TransientStorePool->mappedByteBuffer//We only append data to fileChannel or mappedByteBuffer, never both.//數據從writeBuffer提交數據到fileChannel->forceif (writeBuffer != null || this.fileChannel.position() != 0) {this.fileChannel.force(false);}//數據直接傳到mappedByteBuffer->forceelse {this.mappedByteBuffer.force();}} catch (Throwable e) {log.error("Error occurred when force data to disk.", e);}//更新刷盤位置this.flushedPosition.set(value);this.release();} else {log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());this.flushedPosition.set(getReadPosition());} } return this.getFlushedPosition();

MappedFile#getReadPosition

/*** 獲取當前文件最大可讀指針* @return The max position which have valid data*/ public int getReadPosition() {//如果writeBuffer為空直接返回當前的寫指針，否則返回上次提交的指針//在MappedFile中,只有提交了的數據（寫入到MappedByteBuffer或FileChannel中的數據）才是安全的數據return this.writeBuffer == null ? this.wrotePosition.get() : this.committedPosition.get(); }

MappedFile#shutdown

MappedFile文件銷毀的實現方法為ReferenceResource中的public boolean destory(long intervalForcibly)，intervalForcibly表示拒絕被銷毀的最大存活時間。

if (this.available) {//關閉MappedFilethis.available = false;//設置關閉時間戳this.firstShutdownTimestamp = System.currentTimeMillis();//釋放資源this.release(); } else if (this.getRefCount() > 0) {if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {this.refCount.set(-1000 - this.getRefCount());this.release();} }

5.3.TransientStorePool

用于短時間存儲數據的存儲池。RocketMQ單獨創建ByteBuffer內存緩沖區，用來臨時存儲數據，數據先寫入該內存映射，然后由commit線程將數據復制到目標物理文件所對應的內存映射中。RocketMQ引入該機制主要的原因是提供一種內存鎖定，將當前堆外內存一直鎖定在內存中，避免被進程將內存交換到磁盤。

private final int poolSize; //availableBuffers個數 private final int fileSize; //每個ByteBuffer大小 private final Deque<ByteBuffer> availableBuffers; //雙端隊列-存儲可用緩沖區的容器 private final MessageStoreConfig storeConfig; //消息存儲配置

初始化：

public void init() {//創建poolSize個堆外內存區for (int i = 0; i < poolSize; i++) {//分配內存ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);//內存地址final long address = ((DirectBuffer) byteBuffer).address();Pointer pointer = new Pointer(address);//使用com.sun.jna.Library類庫將該批內存鎖定,避免被置換到交換區,提高存儲性能LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));availableBuffers.offer(byteBuffer);} }

---

6.刷盤機制

6.1.同步刷盤

CommitLog#submitFlushRequest

//同步刷盤 if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {//刷寫CommitLog服務線程final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;//需要等待消息存儲結果if (messageExt.isWaitStoreMsgOK()) {//封裝刷盤請求GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes(),this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());//將request放入刷寫磁盤服務線程中//--------↓--------service.putRequest(request);//等待寫入結果返回return request.future();} else {//喚醒同步刷盤線程service.wakeup();return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);} } else {//異步刷盤.... }

GroupCommitRequest

public static class GroupCommitRequest {private final long nextOffset;private CompletableFuture<PutMessageStatus> flushOKFuture = new CompletableFuture<>();private final long startTimestamp = System.currentTimeMillis();private long timeoutMillis = Long.MAX_VALUE; }

GroupCommitService

class GroupCommitService extends FlushCommitLogService {//分別存儲寫請求和讀請求的容器private volatile LinkedList<GroupCommitRequest> requestsWrite = new LinkedList<GroupCommitRequest>();private volatile LinkedList<GroupCommitRequest> requestsRead = new LinkedList<GroupCommitRequest>();//消息存儲自旋鎖-保護以上容器線程安全private final PutMessageSpinLock lock = new PutMessageSpinLock(); }

GroupCommitService#putRequest

//加上自旋鎖 lock.lock(); try {//將寫請求放入容器this.requestsWrite.add(request); } finally {lock.unlock(); } //喚醒線程 this.wakeup();

GroupCommitService#run

CommitLog.log.info(this.getServiceName() + " service started"); while (!this.isStopped()) {try {//等待線程10sthis.waitForRunning(10);//執行提交任務this.doCommit();} catch (Exception e) {CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);} } // Under normal circumstances shutdown, wait for the arrival of the // request, and then flush try {Thread.sleep(10); } catch (InterruptedException e) {CommitLog.log.warn("GroupCommitService Exception, ", e); } synchronized (this) {this.swapRequests(); } this.doCommit(); CommitLog.log.info(this.getServiceName() + " service end");

GroupCommitService#doCommit

if (!this.requestsRead.isEmpty()) {//遍歷requestsReadfor (GroupCommitRequest req : this.requestsRead) {//刷盤后指針位置大于請求指針偏移量則代表已經刷盤成功//下一個文件中可能有消息，所以最多兩次flushboolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();for (int i = 0; i < 2 && !flushOK; i++) {CommitLog.this.mappedFileQueue.flush(0);flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();}//喚醒發送消息客戶端 req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);}//更新刷盤監測點long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();if (storeTimestamp > 0) {CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);}//清空任務容器this.requestsRead = new LinkedList<>(); } else {//因為個別消息設置為異步flush，所以會走到這個過程CommitLog.this.mappedFileQueue.flush(0); }

6.2.異步刷盤

在消息追加到內存后，立即返回給消息發送端。如果開啟transientStorePoolEnable，RocketMQ會單獨申請一個與目標物理文件（commitLog）同樣大小的堆外內存，該堆外內存將使用內存鎖定，確保不會被置換到虛擬內存中去，消息首先追加到堆外內存，然后提交到物理文件的內存映射中，然后刷寫到磁盤。如果未開啟transientStorePoolEnable，消息直接追加到物理文件直接映射文件中，然后刷寫到磁盤中。

開啟transientStorePoolEnable后異步刷盤步驟：

• 將消息直接追加到ByteBuffer堆外內存
• CommitRealTimeService線程每隔200ms將ByteBuffer中的消息提交到fileChannel
• commit操作成功，將commitedPosition向后移動
• FlushRealTimeService線程每隔500ms將fileChannel的數據刷寫到磁盤

// Synchronization flush if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {... } // Asynchronous flush else {//開啟TransientStorePoolEnableif (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {//喚醒flushCommitLogService服務線程flushCommitLogService.wakeup();} else {commitLogService.wakeup();}return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK); }

CommitRealTimeService#run

提交線程工作機制：

//間隔時間：200msint interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();//一次提交的最少頁數：4int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();//兩次提交的最大間隔：200msint commitDataThoroughInterval =CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();//上次提交間隔超過commitDataThoroughInterval,則忽略提交commitDataLeastPages參數,直接提交long begin = System.currentTimeMillis();if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {this.lastCommitTimestamp = begin;//忽略提交頁數要求commitDataLeastPages = 0;}try {//執行提交操作,將待提交數據提交到物理文件的內存映射區并返回提交結果boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);long end = System.currentTimeMillis();//提交成功if (!result) {this.lastCommitTimestamp = end; // result = false means some data committed.//now wake up flush thread.//喚醒刷盤線程FlushRealTimeService(FlushCommitLogService的子類)flushCommitLogService.wakeup();}if (end - begin > 500) {log.info("Commit data to file costs {} ms", end - begin);}this.waitForRunning(interval);} catch (Throwable e) {CommitLog.log.error(this.getServiceName() + " service has exception. ", e);} }

FlushCommitLogService#run

刷盤線程工作機制：

//線程不停止 while (!this.isStopped()) {//線程執行間隔：500msint interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();//一次刷盤任務最少包含頁數：4int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();//兩次刷盤任務最大間隔：10sint flushPhysicQueueThoroughInterval =CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();boolean printFlushProgress = false;// Print flush progresslong currentTimeMillis = System.currentTimeMillis();//如果當前時間戳大于上次刷盤時間+最大刷盤任務間隔 則本次刷盤任務忽略flushPhysicQueueLeastPages(設置為0) 直接提交刷盤任務if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {this.lastFlushTimestamp = currentTimeMillis;flushPhysicQueueLeastPages = 0;printFlushProgress = (printTimes++ % 10) == 0;}try {if (flushCommitLogTimed) {//線程執行間隔-500mThread.sleep(interval);} else {this.waitForRunning(interval);}if (printFlushProgress) {this.printFlushProgress();}long begin = System.currentTimeMillis();//刷寫磁盤CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);//更新存儲監測點文件的時間戳long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();if (storeTimestamp > 0) {CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);}long past = System.currentTimeMillis() - begin;if (past > 500) {log.info("Flush data to disk costs {} ms", past);}} catch (Throwable e) {CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);this.printFlushProgress();} }

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

總結

以上是生活随笔為你收集整理的RocketMQ：消息存储机制详解与源码解析的全部內容，希望文章能夠幫你解決所遇到的問題。

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