RocketMQ源码分析----刷盘的实现

注：可以先了解一下内存映射，然后再看RocketMq的刷盘，会更容易理解

Broker启动的时候，会调用CommitLog的start方法，然后再启动flushCommitLogService线程

在CommitLog的构造方法中，会判断刷盘的类型

public CommitLog(final DefaultMessageStore defaultMessageStore) {
this.mapedFileQueue = new MapedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),
defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog(), defaultMessageStore.getAllocateMapedFileService());
this.defaultMessageStore = defaultMessageStore;

if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
this.flushCommitLogService = new GroupCommitService();
} else {
this.flushCommitLogService = new FlushRealTimeService();
}

this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
}

同步刷盘使用GroupCommitService，异步刷盘使用FlushRealTimeService，默认是使用异步刷盘

同步和异步的区别在于，broker的处理Producer请求的时候，如果是同步刷盘，那么会进行刷盘后才返回给Producer发送成功，而异步刷盘则是唤醒刷盘线程后就返回

异步刷盘

异步刷盘是FlushRealTimeService，其run方法有个while循环，只要broker不关闭就一直循环下去

public void run() {
while (!this.isStoped()) {
boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();//是否定时刷盘，默认为实时刷盘
//刷盘间隔，默认1秒
int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();//刷CommitLog，至少刷几个PAGE
//刷CommitLog，彻底刷盘间隔时间
int flushPhysicQueueThoroughInterval =
CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();

boolean printFlushProgress = false;

// Print flush progress
long currentTimeMillis = System.currentTimeMillis();
if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {//每flushPhysicQueueThoroughInterva秒彻底刷盘一次

this.lastFlushTimestamp = currentTimeMillis;
flushPhysicQueueLeastPages = 0;//这个参数为0后面会说到
printFlushProgress = ((printTimes++ % 10) == 0);
}

try {
if (flushCommitLogTimed) {//定时刷
Thread.sleep(interval);
} else {//实时刷
this.waitForRunning(interval);
}

if (printFlushProgress) {
this.printFlushProgress();//空方法
}

CommitLog.this.mapedFileQueue.commit(flushPhysicQueueLeastPages);
long storeTimestamp = CommitLog.this.mapedFileQueue.getStoreTimestamp();
if (storeTimestamp > 0) {
CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
}
} catch (Exception 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 < RetryTimesOver && !result; i++) {
result = CommitLog.this.mapedFileQueue.commit(0);
CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
}

}

首先获取配置：刷盘间隔，刷盘时至少刷几个page，彻底刷盘间隔时间等，然后真正刷盘的地方是CommitLog.this.mapedFileQueue.commit(flushPhysicQueueLeastPages);

public boolean commit(final int flushLeastPages) {
boolean result = true;
MapedFile mapedFile = this.findMapedFileByOffset(this.committedWhere, true);
if (mapedFile != null) {
long tmpTimeStamp = mapedFile.getStoreTimestamp();
int offset = mapedFile.commit(flushLeastPages);
long where = mapedFile.getFileFromOffset() + offset;
result = (where == this.committedWhere);
this.committedWhere = where;
if (0 == flushLeastPages) {
this.storeTimestamp = tmpTimeStamp;
}
}

return result;
}

this.findMapedFileByOffset(this.committedWhere, true);是根据对应位移获取对应的文件

然后调用commit方法进行刷盘

public int commit(final int flushLeastPages) {
if (this.isAbleToFlush(flushLeastPages)) {
if (this.hold()) {
int value = this.wrotePostion.get();
this.mappedByteBuffer.force();
this.committedPosition.set(value);
this.release();
} else {
log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
this.committedPosition.set(this.wrotePostion.get());
}
}

return this.getCommittedPosition();
}

然后还要调用isAbleToFlush方法判断是否可以刷盘，真正刷盘是ByteBuffer的force方法

private boolean isAbleToFlush(final int flushLeastPages) {
int flush = this.committedPosition.get();//当前刷盘刷到的位置
int write = this.wrotePostion.get();//当前文件写到的位置

// 如果当前文件已经写满，应该立刻刷盘
if (this.isFull()) {
return true;
}

// 只有未刷盘数据满足指定page数目才刷盘
if (flushLeastPages > 0) {
return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
}

return write > flush;
}

第二个if的地方，如果当前写的位置和上一次刷盘的位置之间相差是flushLeastPages个页以上，才可以进行刷盘，所以即使是实时的刷盘也是要到达指定的页数后才会进行刷盘

当flushLeastPages等于0的情况，只要写的位置比上次刷盘的位置大就行了，而flushLeastPages是在FlushRealTimeService的run方法中设置的

if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {//每flushPhysicQueueThoroughInterva秒彻底刷盘一次

this.lastFlushTimestamp = currentTimeMillis;
flushPhysicQueueLeastPages = 0;
printFlushProgress = ((printTimes++ % 10) == 0);
}

当到达彻底刷盘的时间后，就讲flushLeastPages设置为0
当这个参数为0的时候，那么只要有数据就进行刷盘操作

当这个方法返回true的时候就将ByteBuffer里的数据刷到文件中

综上：broker启动的时候就会启动一个线程，去持续的进行刷盘操作

在处理broker发送请求的时候，有一个判断

// Synchronization flush
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
if (msg.isWaitStoreMsgOK()) {//是否等待服务器将消息存储完毕再返回
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: " + msg.getTopic() + " tags: " + msg.getTags()
+ " client address: " + msg.getBornHostString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
}
} else {
service.wakeup();
}
}
// Asynchronous flush
else {
this.flushCommitLogService.wakeup();
}

wakeup方法如下

public void wakeup() {
synchronized (this) {
if (!this.hasNotified) {
this.hasNotified = true;
this.notify();
}
}
}

protected void waitForRunning(long interval) {
synchronized (this) {
if (this.hasNotified) {
this.hasNotified = false;
this.onWaitEnd();
return;
}

try {
this.wait(interval);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
this.hasNotified = false;
this.onWaitEnd();
}
}
}

将标志位设置为true，然后run的循环中，有行代码this.waitForRunning(interval);就是等待一段时间，然后再去刷盘，而调用了wakeup方法，就不用等待，直接就返回刷盘

另外，run方法的循环外有一段代码

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

当出了循环，证明broker已经停止了，那么需要将所有的数据进行刷盘。而且commit传入的参数为0，那么在判断是否刷盘的时候只要写的位置比上次刷盘的位置大就行了

 

总结：异步刷盘情况下，broker会开启一个线程，等待一段时间，然后判断刷盘条件是否符合，若符合就进行刷盘。在循环中会重复这个操作。在发送消息的时候只会进行一个唤醒操作，然后等待的线程马上返回进行判断刷盘。

在broker停止只会还会将所有的数据进行刷盘

同步刷盘

异步刷盘使用的是FlushRealTimeService，而同步刷盘使用的是GroupCommitService

然后发送消息的时候，有个判断

// Synchronization flush
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
if (msg.isWaitStoreMsgOK()) {//是否等待服务器将消息存储完毕再返回
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: " + msg.getTopic() + " tags: " + msg.getTags()
+ " client address: " + msg.getBornHostString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
}
} else {
service.wakeup();
}
}

首先创建一个GroupCommitRequest对象，参数为刷盘后文件的位置，即当前写到的位置+要刷盘的字节数

接着调用putRequest方法

public void putRequest(final GroupCommitRequest request) {
synchronized (this) {
this.requestsWrite.add(request);
if (!this.hasNotified) {
this.hasNotified = true;
this.notify();
}
}
}

把request放到list中，然后唤醒线程(这里和异步刷盘一样)

request.waitForFlush方法使用CountDownLatch来阻塞当前线程，直到超时或者刷盘完成(GroupCommitService中调用countDown()方法)

和异步刷盘一样，一个while循环中执行刷盘操作，当broker正常停止会把为刷盘的数据进行刷盘，如下

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

while (!this.isStoped()) {
try {
this.waitForRunning(0);
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");
}

刷盘的真正逻辑在CommitLog类中的doCommit方法

private void doCommit() {
if (!this.requestsRead.isEmpty()) {
for (GroupCommitRequest req : this.requestsRead) {
// There may be a message in the next file, so a maximum of
// two times the flush
boolean flushOK = false;
for (int i = 0; (i < 2) && !flushOK; i++) {
flushOK = (CommitLog.this.mapedFileQueue.getCommittedWhere() >= req.getNextOffset());//上一次刷盘的位置是否小于这个刷盘后的位置

if (!flushOK) {
CommitLog.this.mapedFileQueue.commit(0);//刷盘
}
}
//CountDownLatch计数器减1，putMessage中的request.waitForFlush继续执行
req.wakeupCustomer(flushOK);
}

long storeTimestamp = CommitLog.this.mapedFileQueue.getStoreTimestamp();
if (storeTimestamp > 0) {
CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
}

this.requestsRead.clear();
} else {
// Because of individual messages is set to not sync flush, it
// will come to this process
CommitLog.this.mapedFileQueue.commit(0);
}
}

逻辑相对简单，先判断位置是否正确，即刷盘后的位置要大于上次刷盘后的位置，然后和异步刷盘一样，使用commit方法进行刷盘，看了异步刷盘，就知道了参数为0，则只判断位置，不管“至少要刷多少个页”的这个配置，即有多少刷多少

刷完之后，会唤醒刚刚阻塞的线程

总结：

发送消息的时候，组装GroupCommitRequest对象(储存了刷盘位置信息)，唤醒同步刷盘线程GroupCommitService，然后使用CountDownLatch阻塞当前线程直至超时或刷盘完成，然后正在的刷盘操作在CommitLog的doCommit方法中