RocketMQ存储中如何实现同步刷盘和异步刷盘

这篇文章将为大家详细讲解有关RocketMQ存储中如何实现同步刷盘和异步刷盘,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。

专注于为中小企业提供成都网站设计、网站建设服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业玄武免费做网站提供优质的服务。我们立足成都,凝聚了一批互联网行业人才,有力地推动了上千企业的稳健成长,帮助中小企业通过网站建设实现规模扩充和转变。

一、问题思考

1.同步刷盘是怎么工作的?
2.异步刷盘是怎么工作的?
3.上篇文章的疑问,写入堆外内存的消息如何落盘的?

二、Broker启动刷盘有关调用链
1.调用链

//初始化链条
@1 BrokerStartup#main
start(createBrokerController(args));
@2 BrokerStartup#createBrokerController
final BrokerController controller = new BrokerController(...)
boolean initResult = controller.initialize();
@3 BrokerController#initialize
this.messageStore = new DefaultMessageStore(...);
@4 DefaultMessageStore#DefaultMessageStore()
this.commitLog = new CommitLog(this);
@5 CommitLog#CommitLog()
if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig()
.getFlushDiskType()) {
this.flushCommitLogService = new GroupCommitService();
} else {
this.flushCommitLogService = new FlushRealTimeService();
}
this.commitLogService = new CommitRealTimeService();
//启动链条
@6 BrokerStartup#start
controller.start();
@7 BrokerController#start()
this.messageStore.start();
@8 DefaultMessageStore#start()
this.commitLog.start();
@9 CommitLog#start()
this.flushCommitLogService.start();
if (defaultMessageStore.getMessageStoreConfig()
.isTransientStorePoolEnable()) {
this.commitLogService.start();
}

小结:由调用链可以看出,初始化并启动了以下线程类

1.同步刷盘 GroupCommitService

2.异步刷盘 FlushRealTimeService

3.如果开启堆外内存并且为异步刷盘 CommitRealTimeService


2.线程类关系图

RocketMQ存储中如何实现同步刷盘和异步刷盘

三、线程类工作流程

既然线程类在Broker启动时就启动了,他们在做啥呢?

1.堆外内存线程类CommitRealTimeService工作流程

RocketMQ存储中如何实现同步刷盘和异步刷盘


小结:
1.CommitRealTimeService主要工作是将写入堆外内存(writeBuffer)的消息,写入到fileChannel中,fileChannel为commitLog文件通道

2.committedPosition用于记录将writeBuffer数据写入到fileChannel中的内存位点(相对偏移量offset)
3.committedWhere用于记录写入fileChannel中的物理偏移量(文件名称+相对偏移量offset)

2.同步刷盘线程类GroupCommitService工作流程

RocketMQ存储中如何实现同步刷盘和异步刷盘

注1:

1.执行onWaitEnd时交换读写容器,该线程类提供两个容器来装GroupCommitRequest

2.requestsWrite和requestsRead,每次执行提交(刷盘)前都会进行容器交换

3.好处:读写请求容器分离,避免潜在的锁竞争

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

注2:

1.flushedPosition 标记已经刷盘内存的位点。即刷盘相对偏移量,刷盘到什么位置了,下次从此处刷盘即可

2.flushedWhere 标记已经刷盘的物理偏移量,根据此位置可精确查找到文件中消息的存储位置。flushedWhere = 当前刷盘文件名称(该日志文件的起始物理偏移量) + flushedPosition

注3:流程图中标记红色部分,将刷盘结果通知给等待线程

小结:同步刷盘线程类GroupCommitService主要工作
将请求从读容器中取出并通过mappedByteBuffer.force()将数据落盘。

3.异步刷盘线程类FlushRealTimeService工作流程

RocketMQ存储中如何实现同步刷盘和异步刷盘

小结:FlushRealTimeService主要工作
1.不开启堆外外内存刷盘方式为mappedByteBuffer.force()
2.开启堆外内存刷盘方式为fileChannel.force


疑问:同步刷盘线程类GroupCommitService每执行一次都会交换读写容器,那刷盘请求什么时候放到写容器(requestsWrite)呢?


四、消息追加与线程类的交互

分析完线程类后,把镜头切换到消息追加,看看消息进来后是如何跟线程类交互的?


1.调用链

@1 CommitLog#putMessage
//同步刷盘或者异步刷盘
handleDiskFlush(result, putMessageResult, msg);
@2 CommitLog#handleDiskFlush

2.同步刷盘主要代码

同步刷盘时构造刷盘请求,将请求提交给线程类GroupCommitService,service.putRequest(request),并获取刷盘结果。

if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
//等待MappedFile刷盘成功状态通过countDownLatch来控制
boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
}
}

3.异步刷盘主要代码

未开启堆外内存唤醒FlushRealTimeServicee,开启堆外内存唤醒CommitRealTimeService。

if (!this.defaultMessageStore.getMessageStoreConfig()
.isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}

五、刷盘方式示意图
1.同步刷盘示意图

RocketMQ存储中如何实现同步刷盘和异步刷盘


2.异步刷盘未开启堆外缓存示意图

RocketMQ存储中如何实现同步刷盘和异步刷盘


3.异步刷盘开启堆外缓存示意图

RocketMQ存储中如何实现同步刷盘和异步刷盘

关于“RocketMQ存储中如何实现同步刷盘和异步刷盘”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,使各位可以学到更多知识,如果觉得文章不错,请把它分享出去让更多的人看到。


网站栏目:RocketMQ存储中如何实现同步刷盘和异步刷盘
标题链接:http://bzwzjz.com/article/gepcce.html

其他资讯

Copyright © 2007-2020 广东宝晨空调科技有限公司 All Rights Reserved 粤ICP备2022107769号
友情链接: 成都模版网站建设 做网站设计 成都网站制作公司 网站设计 成都网站设计公司 移动网站建设 网站制作公司 定制网站设计 网站制作 成都网站建设 商城网站建设 网站建设开发 泸州网站建设 高端网站设计推广 成都做网站建设公司 成都网站设计 成都网站制作 高端网站建设 成都响应式网站建设 外贸网站设计方案 成都网站设计 成都营销网站制作