從源碼分析RocketMq消息的存儲原理
rocketmq在存儲消息的時候，最終是通過mmap映射成磁碟文件進行存儲的，本文就消息的存儲流程作一個整理。源碼版本是4.9.2
主要的存儲組件有如下4個：
CommitLog：存儲的業務層，接收「保存消息」的請求
MappedFile：存儲的最底層對象，一個MappedFile對象就對應了一個實際的文件
MappedFileQueue：管理MappedFile的容器
AllocateMappedFileService：非同步創建mappedFile的服務
對於rocketmq來說，存儲消息的主要文件被稱為CommitLog，因此就從該類入手。處理存儲請求的入口方法是asyncPutMessage，主要流程如下：
public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
...
//可能會有多個執行緒並發請求，雖然支援集群，但是對於每個單獨的broker都是本地存儲，所以記憶體鎖就足夠了
putMessageLock.lock();
try {
//獲取最新的文件
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
...
//如果文件為空，或者已經存滿，則創建一個新的commitlog文件
if (null == mappedFile || mappedFile.isFull()) {
mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
}
...
//調用底層的mappedFile進行出處，但是注意此時還沒有刷盤
result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
...
} finally {
putMessageLock.unlock();
}
PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
...
}
因此對於Commitlog.asyncPutMessage來說，主要的工作就是2步：
1.獲取或者創建一個MappedFile
2.調用appendMessage進行存儲
接下去我們先看MappedFile的創建，查看mappedFileQueue.getLastMappedFile方法，最終會調用到doCreateMappedFile方法，調用流如下：
getLastMappedFile-->tryCreateMappedFile-->doCreateMappedFile
protected MappedFile doCreateMappedFile(String nextFilePath, String nextNextFilePath) {
MappedFile mappedFile = null;
//如果非同步服務對象不為空，那麼就採用非同步創建文件的方式
if (this.allocateMappedFileService != null) {
mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
nextNextFilePath, this.mappedFileSize);
} else {
//否則就同步創建
try {
mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
} catch (IOException e) {
log.error("create mappedFile exception", e);
}
}
...
return mappedFile;
}
因此對於MappedFileQueue來說，主要工作就2步：
1.如果有非同步服務，那麼就非同步創建mappedFile
2.否則就同步創建
接下去主要看非同步創建的流程，查看allocateMappedFileService.putRequestAndReturnMappedFile
public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
...
//創建mappedFile的請求，
AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
//將其放入ConcurrentHashMap中，主要用於並發判斷，保證不會創建重複的mappedFile
boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
//如果map添加成功，就可以將request放入隊列中，實際創建mappedFile的執行緒也是從該queue中獲取request
if (nextPutOK) {
boolean offerOK = this.requestQueue.offer(nextReq);
}
AllocateRequest result = this.requestTable.get(nextFilePath);
try {
if (result != null) {
//因為是非同步創建，所以這裡需要await，等待mappedFile被非同步創建成功
boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
//返回創建好的mappedFile
return result.getMappedFile();
}
} catch (InterruptedException e) {
log.warn(this.getServiceName() + " service has exception. ", e);
}
return null;
}
因此對於AllocateMappedFileService.putRequestAndReturnMappedFile，主要工作也是2步：
1.將「創建mappedFile」的請求放入隊列中
2.等待非同步執行緒實際創建完mappedFile
接下去看非同步執行緒是如何具體創建mappedFile的。既然AllocateMappedFileService本身就是負責創建mappedFile的，並且其本身也實現了Runnable介面，我們查看其run方法，其中會調用mmapOperation，這就是最終執行創建mappedFile的方法
private boolean mmapOperation() {
boolean isSuccess = false;
AllocateRequest req = null;
try {
//從隊列中拿request
req = this.requestQueue.take();
...
if (req.getMappedFile() == null) {
MappedFile mappedFile;
//判斷是否採用堆外記憶體
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
try {
//如果開啟了堆外記憶體，rocketmq允許外部注入自定義的MappedFile實現
mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
} catch (RuntimeException e) {
//如果沒有自定義實現，那麼就採用默認的實現
log.warn("Use default implementation.");
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
}
} else {
//如果未採用堆外記憶體，那麼就直接採用默認實現
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
}
...
//這裡會預熱文件，這裡涉及到了系統的底層調用
mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
req.setMappedFile(mappedFile);
}
...
} finally {
if (req != null && isSuccess)
//無論是否創建成功，都要喚醒putRequestAndReturnMappedFile方法中的等待執行緒
req.getCountDownLatch().countDown();
}
return true;
}
因此對於mmapOperation創建mappedFile，主要工作為4步：
1.從隊列中獲取putRequestAndReturnMappedFile方法存放的request
2.根據是否啟用對外記憶體，分支創建mappedFile
3.預熱mappedFile
4.喚醒putRequestAndReturnMappedFile方法中的等待執行緒
接下去查看mappedFile內部的具體實現，我們可以發現在構造函數中，也會調用內部的init方法，這就是主要實現mmap的方法
private void init(final String fileName, final int fileSize) throws IOException {
...
//創建文件對象
this.file = new File(fileName);
try {
//獲取fileChannel
this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
//進行mmap操作，將磁碟空間映射到記憶體
this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
...
} finally {
...
}
}
因此對於init執行mmap，主要工作分為2步：
1.獲取文件的fileChannel
2.執行mmap映射
而如果採用了堆外記憶體，那麼除了上述的mmap操作，還會額外分配對外記憶體
this.writeBuffer = transientStorePool.borrowBuffer();
到這裡，CommitLog.asyncPutMessage方法中的獲取或創建mappedFile就完成了。
接下去需要查看消息具體是符合被寫入文件中的。查看mappedFile的appendMessage方法，最終會調用到appendMessagesInner方法：
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,
PutMessageContext putMessageContext) {
//如果是啟用了對外記憶體，那麼會優先寫入對外記憶體，否則直接寫入mmap記憶體
ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
byteBuffer.position(currentPos);
AppendMessageResult result;
...
//調用外部的callback執行實際的寫入操作
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,
(MessageExtBrokerInner) messageExt, putMessageContext);
...
return result;
}
因此對於appendMessage方法，主要工作分為2步：
1.判斷是否啟用對外記憶體，從而選擇對應的buffer對象
2.調用傳入的callback方法進行實際寫入
接下去查看外部傳入的callback方法，是由CommitLog.asyncPutMessage傳入
result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
而this.appendMessageCallback則是在CommitLog的構造函數中初始化的
this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
查看DefaultAppendMessageCallback.doAppend方法，因為本文不關心消息的具體結構，所以省略了大部分構造buffer的程式碼：
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {
...
//獲取消息編碼後的buffer
ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
...
//寫入buffer中，如果啟用了對外記憶體，那麼就會寫入外部傳入的writerBuffer，否則直接寫入mappedByteBuffer中
byteBuffer.put(preEncodeBuffer);
...
return result;
}
因此對於doAppend方法，主要工作分為2步：
1.將消息編碼
2.將編碼後的消息寫入buffer中，可以是writerBuffer或者mappedByteBuffer
此時雖然位元組流已經寫入了buffer中，但是對於堆外記憶體，此時數據還僅存在於記憶體中，而對於mappedByteBuffer，雖然會有系統執行緒定時刷數據落盤，但是這並非我們可以控，因此也只能假設還未落盤。為了保證數據能落盤，rocketmq還有一個非同步刷盤的執行緒，接下去再來看下非同步刷盤是如何處理的。
查看CommitLog的構造函數，其中有3個service，分別負責同步刷盤、非同步刷盤和堆外記憶體寫入fileChannel
public CommitLog(final DefaultMessageStore defaultMessageStore) {
...
//同步刷盤
if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
this.flushCommitLogService = new GroupCommitService();
} else {
//非同步刷盤
this.flushCommitLogService = new FlushRealTimeService();
}
//將對外記憶體的數據寫入fileChannel
this.commitLogService = new CommitRealTimeService();
...
}
先看CommitRealTimeService.run方法，其中最關鍵的程式碼如下：
boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
查看mappedFileQueue.commit方法，關鍵如下：
int offset = mappedFile.commit(commitLeastPages);
查看mappedFile.commit方法：
public int commit(final int commitLeastPages) {
//如果為空，說明不是堆外記憶體，就不需要任何操作，只需等待刷盤即可
if (writeBuffer == null) {
return this.wrotePosition.get();
}
if (this.isAbleToCommit(commitLeastPages)) {
if (this.hold()) {
//如果是堆外記憶體，那麼需要做commit
commit0();
this.release();
}
...
}
return this.committedPosition.get();
}
查看commit0方法：
protected void commit0() {
...
//獲取堆外記憶體
ByteBuffer byteBuffer = writeBuffer.slice();
//寫入fileChannel
this.fileChannel.write(byteBuffer);
...
}
因此對於CommitRealTimeService，工作主要分2步：
1.判斷是否是對外記憶體，如果不是那就不需要處理
2.如果是對外記憶體，則寫入fileChannel
最後查看同步刷盤的GroupCommitService和非同步刷盤FlushRealTimeService，查看其run方法，會發現其本質都是調用了如下方法：
CommitLog.this.mappedFileQueue.flush
當然在處理的邏輯上還有計算position等等邏輯，但這不是本文所關心的，所以就省略了。
同步和非同步的區別體現在了執行刷盤操作的時間間隔，對於同步刷盤，固定間隔10ms：
this.waitForRunning(10);
而對於非同步刷盤，時間間隔為配置值，默認500ms：
int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
...
if (flushCommitLogTimed) {
Thread.sleep(interval);
} else {
this.waitForRunning(interval);
}
最後查看mappedFileQueue.flush是如何刷盤的。最終會調用到mappedFile的flush方法：
public int flush(final int flushLeastPages) {
...
//如果是使用了堆外記憶體，那麼調用的是fileChannel的刷盤
if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
//如果非堆外記憶體，那麼調用的是mappedByteBuffer的刷盤
this.mappedByteBuffer.force();
}
...
return this.getFlushedPosition();
}
因此最終的刷盤，工作主要分2步，正和前面的CommitRealTimeService工作對應：
1.如果是使用了堆外記憶體，那麼調用fileChannel的刷盤
2.如果非堆外記憶體，那麼調用mappedByteBuffer的刷盤
至此，整個rocketmq消息落盤的流程就完成了，接下去重新整理下整個流程：
1.CommitLog：存儲的業務層，接收「保存消息」的請求，主要有2個功能：創建mappedFile、非同步寫入消息。
2.AllocateMappedFileService：非同步創建mappedFile的服務，通過構建AllocateRequest對象和隊列進行執行緒間的通訊。雖然MappedFile的實際創建是通過非同步執行緒執行的，但是當前執行緒會等待創建完成後再返回，所以實際上是非同步阻塞的。
3.MappedFile：存儲的最底層對象，一個MappedFile對象就對應了一個實際的文件。在init方法中創建了fileChannel，並完成了mmap操作。如果啟用了堆外記憶體，則會額外初始化writeBuffer，實現讀寫分離。
4.MappedFileQueue：管理MappedFile的容器。
5.寫入消息的時候，會根據是否啟用堆外記憶體，寫入writeBuffer或者mappedByteBuffer。
6.實際落盤是通過非同步的執行緒實現的，分為名義上的同步（GroupCommitService）和非同步（FlushRealTimeService），不過主要區別在於執行落盤方法的時間間隔不同，最終都是調用mappedFile的flush方法
7.落盤會根據是否啟用對外記憶體，分別調用fileChannel.force或者mappedByteBuffer.force
