您的位置:首页 > 其它

RocketMQ中Broker的刷盘源码分析

七月安生 2019-08-07 00:35 701 查看 https://www.cnblogs.com/a52658

上一篇博客的最后简单提了下CommitLog的刷盘  【RocketMQ中Broker的消息存储源码分析】 (这篇博客和上一篇有很大的联系)


Broker的CommitLog刷盘会启动一个线程,不停地将缓冲区的内容写入磁盘(CommitLog文件)中,主要分为异步刷盘和同步刷盘

异步刷盘又可以分为两种方式:
①缓存到mappedByteBuffer -> 写入磁盘(包括同步刷盘)
②缓存到writeBuffer -> 缓存到fileChannel -> 写入磁盘 (前面说过的开启内存字节缓冲区情况下)

 

CommitLog的两种刷盘模式:

public enum FlushDiskType {
SYNC_FLUSH,
ASYNC_FLUSH
}

同步和异步,同步刷盘由GroupCommitService实现,异步刷盘由FlushRealTimeService实现,默认采用异步刷盘

在采用异步刷盘的模式下,若是开启内存字节缓冲区,那么会在FlushRealTimeService的基础上开启CommitRealTimeService

 

同步刷盘:

启动GroupCommitService线程:

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

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

通过循环中的doCommit不断地进行刷盘


doCommit方法:

private void doCommit() {
synchronized (this.requestsRead) {
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.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();

if (!flushOK) {
CommitLog.this.mappedFileQueue.flush(0);
}
}

req.wakeupCustomer(flushOK);
}

long storeTimestamp = CommitLog.this.mappedFileQueue.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.mappedFileQueue.flush(0);
}
}
}

其中在GroupCommitService中管理着两张List:

private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();

GroupCommitRequest中封装了一个Offset

private final long nextOffset;

 


这里就需要看到上一篇博客结尾提到的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();
}
}
}

这个方法的调用发生在Broker接收到来自Producer的消息,并且完成了向ByteBuffer的写入

可以看到,在同步刷盘SYNC_FLUSH模式下,会从AppendMessageResult 中取出WroteOffset以及WroteBytes从而计算出nextOffset,把这个nextOffset封装到GroupCommitRequest中,然后通过GroupCommitService 的putRequest方法,将GroupCommitRequest添加到requestsWrite这个List中
putRequest方法:

public synchronized void putRequest(final GroupCommitRequest request) {
synchronized (this.requestsWrite) {
this.requestsWrite.add(request);
}
if (hasNotified.compareAndSet(false, true)) {
waitPoint.countDown(); // notify
}
}

在完成List的add操作后,会通过CAS操作修改hasNotified这个原子化的Boolean值,同时通过waitPoint的countDown进行唤醒操作,在后面会有用


由于这里这里是同步刷盘,所以需要通过GroupCommitRequest的waitForFlush方法,在超时时间内等待该记录对应的刷盘完成
而异步刷盘会通过wakeup方法唤醒刷盘任务,并没有进行等待,这就是二者区别


回到doCommit方法中,这时会发现这里是对requestsRead这条List进行的操作,而刚才是将记录存放在requestsWrite这条List中的
这就和在run方法中的waitForRunning方法有关了:

protected void waitForRunning(long interval) {
if (hasNotified.compareAndSet(true, false)) {
this.onWaitEnd();
return;
}

//entry to wait
waitPoint.reset();

try {
waitPoint.await(interval, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("Interrupted", e);
} finally {
hasNotified.set(false);
this.onWaitEnd();
}
}

这里通过CAS操作修改hasNotified值,从而调用onWaitEnd方法;如果修改失败,则因为await进入阻塞,等待上面所说的putRequest方法将其唤醒,也就是说当Producer发送的消息被缓存成功后,调用handleDiskFlush方法后,唤醒刷盘线工作,当然刷盘线程在达到超时时间interval后也会唤醒


再来看看onWaitEnd方法:

protected void onWaitEnd() {
this.swapRequests();
}

private void swapRequests() {
List<GroupCommitRequest> tmp = this.requestsWrite;
this.requestsWrite = this.requestsRead;
this.requestsRead = tmp;
}

可以看到,这里是将两个List进行了交换

这是一个非常有趣的做法,如果熟悉JVM的话,有没有觉得这其实很像新生代的标记-清除算法!
当刷盘线程阻塞的时候,requestsWrite中会填充记录,当刷盘线程被唤醒工作的时候,首先会将requestsWrite和requestsRead进行交换,那么此时的记录就是从requestsRead中读取的了,而同时requestsWrite会变为空的List,消息记录就会往这个空的List中填充,如此往复

可以看到doCommit方法中,当requestsRead不为空的时候,在最后会调用requestsRead的clear方法,由此证明了我上面的说法

 

仔细来看看是如何进行刷盘的:

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.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();

if (!flushOK) {
CommitLog.this.mappedFileQueue.flush(0);
}
}

req.wakeupCustomer(flushOK);
}

通过遍历requestsRead,可以到得到GroupCommitRequest封装的NextOffset

其中flushedWhere是用来记录上一次刷盘完成后的offset,若是上一次的刷盘位置大于等于NextOffset,就说明从NextOffset位置起始已经被刷新过了,不需要刷新,否则调用mappedFileQueue的flush方法进行刷盘

MappedFileQueue的flush方法:

public boolean flush(final int flushLeastPages) {
boolean result = true;
MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);
if (mappedFile != null) {
long tmpTimeStamp = mappedFile.getStoreTimestamp();
int offset = mappedFile.flush(flushLeastPages);
long where = mappedFile.getFileFromOffset() + offset;
result = where == this.flushedWhere;
this.flushedWhere = where;
if (0 == flushLeastPages) {
this.storeTimestamp = tmpTimeStamp;
}
}

return result;
}

这里首先根据flushedWhere上一次刷盘完成后的offset,通过findMappedFileByOffset方法,找到CommitLog文件的映射MappedFile
有关MappedFile及其相关操作在我之前的博客中介绍过很多次,就不再累赘


再找到MappedFile后,调用其flush方法:

MappedFile的flush方法:

public int flush(final int flushLeastPages) {
if (this.isAbleToFlush(flushLeastPages)) {
if (this.hold()) {
int value = getReadPosition();

try {
//We only append data to fileChannel or mappedByteBuffer, never both.
if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
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();
}


首先isAbleToFlush方法:

private boolean isAbleToFlush(final int flushLeastPages) {
int flush = this.flushedPosition.get();
int write = getReadPosition();

if (this.isFull()) {
return true;
}

if (flushLeastPages > 0) {
return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
}

return write > flush;
}

其中flush记录的是上一次完成刷新后的位置,write记录的是当前消息内容写入后的位置
当flushLeastPages 大于0的时候,通过:

return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;

可以计算出是否满足page的要求,其中OS_PAGE_SIZE是4K,也就是说1个page大小是4k

由于这里是同步刷盘,flushLeastPages是0,不对page要求,只要有缓存有内容就会刷盘;但是在异步刷盘中,flushLeastPages是4,也就是说,只有当缓存的消息至少是4(page个数)*4K(page大小)= 16K时,异步刷盘才会将缓存写入文件

 

回到MappedFile的flush方法,在通过isAbleToFlush检查完写入要求后

int value = getReadPosition();
try {
//We only append data to fileChannel or mappedByteBuffer, never both.
if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
this.mappedByteBuffer.force();
}
} catch (Throwable e) {
log.error("Error occurred when force data to disk.", e);
}

this.flushedPosition.set(value);

首先通过getReadPosition获取当前消息内容写入后的位置,由于是同步刷盘,所以这里调用mappedByteBuffer的force方法,通过JDK的NIO操作,将mappedByteBuffer缓存中的数据写入CommitLog文件中
最后更新flushedPosition的值


再回到MappedFileQueue的flush方法,在完成MappedFile的flush后,还需要更新flushedWhere的值

此时缓存中的数据完成了持久化,同步刷盘结束

 

异步刷盘:

①FlushCommitLogService:

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

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

int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();

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");
}

flushCommitLogTimed:是否使用定时刷盘
interval:刷盘时间间隔,默认500ms
flushPhysicQueueLeastPages:page大小,默认4个
flushPhysicQueueThoroughInterval:彻底刷盘时间间隔,默认10s


首先根据lastFlushTimestamp(上一次刷盘时间)+ flushPhysicQueueThoroughInterval和当前时间比较,判断是否需要进行一次彻底刷盘,若达到了需要则将flushPhysicQueueLeastPages置为0


接着根据flushCommitLogTimed判断
当flushCommitLogTimed为true,使用sleep等待500ms
当flushCommitLogTimed为false,调用waitForRunning在超时时间为500ms下阻塞,其唤醒条件也就是在handleDiskFlush中的wakeup唤醒

最后,和同步刷盘一样,调用mappedFileQueue的flush方法
只不过,这里的flushPhysicQueueLeastPages决定了其是进行彻底刷新,还是按4page(16K)的标准刷新


②CommitRealTimeService
这种刷盘方式需要和FlushCommitLogService配合


CommitRealTimeService的run方法:

public void run() {
CommitLog.log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();

int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();

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");
}

这里的逻辑和FlushCommitLogService中相似,之不过参数略有不同

interval:提交时间间隔,默认200ms
commitDataLeastPages:page大小,默认4个
commitDataThoroughInterval:提交完成时间间隔,默认200ms

基本和FlushCommitLogService相似,只不过调用了mappedFileQueue的commit方法

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;
}

这里和mappedFileQueue的flush方法很相似,通过committedWhere寻找MappedFile

然后调用MappedFile的commit方法:

public int commit(final int commitLeastPages) {
if (writeBuffer == null) {
//no need to commit data to file channel, so just regard wrotePosition as committedPosition.
return this.wrotePosition.get();
}
if (this.isAbleToCommit(commitLeastPages)) {
if (this.hold()) {
commit0(commitLeastPages);
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的flush方法很相似,在isAbleToCommit检查完page后调用commit0方法


MappedFile的commit0方法:

protected void commit0(final int commitLeastPages) {
int writePos = this.wrotePosition.get();
int lastCommittedPosition = this.committedPosition.get();

if (writePos - this.committedPosition.get() > 0) {
try {
ByteBuffer byteBuffer = writeBuffer.slice();
byteBuffer.position(lastCommittedPosition);
byteBuffer.limit(writePos);
this.fileChannel.position(lastCommittedPosition);
this.fileChannel.write(byteBuffer);
this.committedPosition.set(writePos);
} catch (Throwable e) {
log.error("Error occurred when commit data to FileChannel.", e);
}
}
}

 【RocketMQ中Broker的消息存储源码分析】 

中说过,当使用这种方式时,会先将消息缓存在writeBuffer中而不是之前的mappedByteBuffer
这里就可以清楚地看到将writeBuffer中从lastCommittedPosition(上次提交位置)开始到writePos(缓存消息结束位置)的内容缓存到了fileChannel中相同的位置,并没有写入磁盘
在缓存到fileChannel后,会更新committedPosition值


回到commit方法,在向fileCfihannel缓存完毕后,会检查committedPosition是否达到了fileSize,也就是判断writeBuffer中的内容是不是去全部提交完毕

若是全部提交,需要通过transientStorePool的returnBuffer方法来回收利用writeBuffer
transientStorePool其实是一个双向队列,由CommitLog来管理
TransientStorePool:

public class TransientStorePool {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

private final int poolSize;
private final int fileSize;
private final Deque<ByteBuffer> availableBuffers;
private final MessageStoreConfig storeConfig;

public TransientStorePool(final MessageStoreConfig storeConfig) {
this.storeConfig = storeConfig;
this.poolSize = storeConfig.getTransientStorePoolSize();
this.fileSize = storeConfig.getMapedFileSizeCommitLog();
this.availableBuffers = new ConcurrentLinkedDeque<>();
}
......
}


returnBuffer方法:

public void returnBuffer(ByteBuffer byteBuffer) {
byteBuffer.position(0);
byteBuffer.limit(fileSize);
this.availableBuffers.offerFirst(byteBuffer);
}

这里就可以清楚地看到byteBuffer确实被回收了

 

回到MappedFileQueue的commit方法:

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的commit后,通过where和committedWhere来判断是否真的向fileCfihannel缓存了 ,只有确实缓存了result才是false!
之后会更新committedWhere,并返回result

 

那么回到CommitRealTimeService的run方法,在完成commit之后,会判断result
只有真的向fileCfihannel缓存后,才会调用flushCommitLogService的wakeup方法,也就是唤醒了FlushCommitLogService的刷盘线程

唯一和之前分析的FlushCommitLogService不同的地方是在MappedFile的flush方法中:

if (writeBuffer != null || this.fileChannel.position() != 0) {
this.fileChannel.force(false);
} else {
this.mappedByteBuffer.force();
}

之前在没有开启内存字节缓冲区的情况下,是将mappedByteBuffer中的内容写入磁盘
而这时,终于轮到fileChannel了

可以看到这里的条件判断,当writeBuffer不等与null,或者fileChannel的position不等与0
writeBuffer等于null的情况会在TransientStorePool对其回收之后


到这里就可以明白开启内存字节缓冲区的情况下,其实是进行了两次缓存才写入磁盘

 

至此,Broker的消息持久化以及刷盘的整个过程完毕

标签: