RocketMQ設計之異步刷盤

更新時間：2022年03月21日 10:29:20 作者：周杰倫本人

本文介紹RocketMQ設計之異步刷盤，RocketMQ消息存儲到磁盤上,這樣既保證斷電后恢復,也讓存儲消息量超出內(nèi)存限制,RocketMQ為了提高性能,會盡可能保證磁盤順序?qū)?消息通過Producer寫入RocketMQ的時候,有兩種方式,上篇介紹了同步刷盤,本文介紹異步刷盤,需要的朋友可以參考下

上一篇RocketMQ設計之同步刷盤

異步刷盤方式：在返回寫成功狀態(tài)時，消息可能只是被寫入了內(nèi)存的PAGECACHE，寫操作的返回快，吞吐量大；當內(nèi)存里的消息量積累到一定程度時，統(tǒng)一觸發(fā)寫磁盤操作，快速寫入

RocketMQ默認采用異步刷盤，異步刷盤兩種策略：開啟緩沖池，不開啟緩沖池

CommitLog的handleDiskFlush方法：

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
? ? // Synchronization flush
? ? if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
? ? ? ? final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
? ? ? ? if (messageExt.isWaitStoreMsgOK()) {
? ? ? ? ? ? GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
? ? ? ? ? ? service.putRequest(request);
? ? ? ? ? ? boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
? ? ? ? ? ? if (!flushOK) {
? ? ? ? ? ? ? ? log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()
? ? ? ? ? ? ? ? ? ? + " client address: " + messageExt.getBornHostString());
? ? ? ? ? ? ? ? putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
? ? ? ? ? ? }
? ? ? ? } else {
? ? ? ? ? ? service.wakeup();
? ? ? ? }
? ? }
? ? // Asynchronous flush
? ? else {
? ? ? ? if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
? ? ? ? ? ? flushCommitLogService.wakeup();
? ? ? ? } else {
? ? ? ? ? ? commitLogService.wakeup();
? ? ? ? }
? ? }
}

不開啟緩沖池：默認不開啟，刷盤線程FlushRealTimeService會每間隔500毫秒嘗試去刷盤。

class FlushRealTimeService extends FlushCommitLogService {
? ? private long lastFlushTimestamp = 0;
? ? private long printTimes = 0;

? ? public void run() {
? ? ? ? CommitLog.log.info(this.getServiceName() + " service started");

? ? ? ? while (!this.isStopped()) {
? ? ? ? ? ? boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();

? ? ? ? ? ? //每次Flush間隔500毫秒
? ? ? ? ? ? int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
? ? ? ? ? ? //每次Flush最少4頁內(nèi)存數(shù)據(jù)（16KB）
? ? ? ? ? ? int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();

? ? ? ? ? ??? ?//距離上次刷盤時間閾值為10秒
? ? ? ? ? ? int flushPhysicQueueThoroughInterval =
? ? ? ? ? ? ? ? CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();

? ? ? ? ? ? boolean printFlushProgress = false;

? ? ? ? ? ? // Print flush progress
? ? ? ? ? ? long currentTimeMillis = System.currentTimeMillis();
? ? ? ? ? ? if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
? ? ? ? ? ? ? ? this.lastFlushTimestamp = currentTimeMillis;
? ? ? ? ? ? ? ? flushPhysicQueueLeastPages = 0;
? ? ? ? ? ? ? ? printFlushProgress = (printTimes++ % 10) == 0;
? ? ? ? ? ? }

? ? ? ? ? ? try {
? ? ? ? ? ? ? ? if (flushCommitLogTimed) {
? ? ? ? ? ? ? ? ? ? Thread.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();
? ? ? ? ? ? }
? ? ? ? }

? ? ? ? // Normal shutdown, to ensure that all the flush before exit
? ? ? ? boolean result = false;
? ? ? ? for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
? ? ? ? ? ? result = CommitLog.this.mappedFileQueue.flush(0);
? ? ? ? ? ? CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
? ? ? ? }

? ? ? ? this.printFlushProgress();

? ? ? ? CommitLog.log.info(this.getServiceName() + " service end");
? ? }

? ? @Override
? ? public String getServiceName() {
? ? ? ? return FlushRealTimeService.class.getSimpleName();
? ? }

? ? private void printFlushProgress() {
? ? ? ? // CommitLog.log.info("how much disk fall behind memory, "
? ? ? ? // + CommitLog.this.mappedFileQueue.howMuchFallBehind());
? ? }

? ? @Override
? ? public long getJointime() {
? ? ? ? return 1000 * 60 * 5;
? ? }
}

判斷是否超過10秒沒刷盤了，如果超過強制刷盤
等待Flush間隔500ms
通過MappedFile刷盤
設置StoreCheckpoint刷盤時間點
超過500ms的刷盤記錄日志
Broker正常停止前，把內(nèi)存page中的數(shù)據(jù)刷盤

開啟緩沖池：

class CommitRealTimeService extends FlushCommitLogService {

? ? private long lastCommitTimestamp = 0;

? ? @Override
? ? public String getServiceName() {
? ? ? ? return CommitRealTimeService.class.getSimpleName();
? ? }

? ? @Override
? ? public void run() {
? ? ? ? CommitLog.log.info(this.getServiceName() + " service started");
? ? ? ? while (!this.isStopped()) {
? ? ? ? ? ? //每次提交間隔200毫秒
? ? ? ? ? ? int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();

? ? ? ? ? ? //每次提交最少4頁內(nèi)存數(shù)據(jù)（16KB）
? ? ? ? ? ? int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();

? ? ? ? ? ? //距離上次提交時間閾值為200毫秒
? ? ? ? ? ? int commitDataThoroughInterval =
? ? ? ? ? ? ? ? CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();

? ? ? ? ? ? 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.
? ? ? ? ? ? ? ? ? ? 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);
? ? ? ? ? ? }
? ? ? ? }

? ? ? ? boolean result = false;
? ? ? ? for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
? ? ? ? ? ? result = CommitLog.this.mappedFileQueue.commit(0);
? ? ? ? ? ? CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
? ? ? ? }
? ? ? ? CommitLog.log.info(this.getServiceName() + " service end");
? ? }
}

RocketMQ申請一塊和CommitLog文件相同大小的堆外內(nèi)存來做緩沖池，數(shù)據(jù)會先寫入緩沖池，提交線程CommitRealTimeService也每間隔500毫秒嘗試提交到文件通道等待刷盤，刷盤最終由FlushRealTimeService來完成，和不開啟緩沖池的處理一致。使用緩沖池的目的是多條數(shù)據(jù)合并寫入，從而提高io性能。