Broker数据存储
2021-07-20 21:28:08 2 举报
登录查看完整内容
rocketmq之broker存储
作者其他创作
大纲/内容
宕机或扩容
消息
CommitLogDispatcherBuildConsumeQueue
预读取
Broker接收到消息后,存入commitlog磁盘文件中,每个commitlog文件大小1GB,顺序写入
ReputMessageService线程间隔1毫秒
同步刷盘
DefalutMessageStore.cleanFilesPeriodically方法fileReservedTime指定删除文件的时间,默认72小时
获取要转发的消息
消息数据
进入rebalance环节,重新进行messagequeue的分配;将所有的messagequeue负载重平衡到所有的消费者机器。
寻找消息
OS Cache
写入文件
commitlog顺序存储broker接受到的真实消息,而每个topic都messagequeue都对应一个consumequeue文件,此文件记录了messagequeue的消息地址,也就是偏移量。可以通过此文件找到某个topic的某个messagequeue在commitlog中的具体的消息。
FlushCommitLogService--线程CommitRealTimeService.run()CommitLog.this.mappedFileQueue.commitmappedFile.commit(xx)this.fileChannel.write(byteBuffer)
GroupCommitServiceCommitLog.this.mappedFileQueue.flush(0)mappedFile.flush(xx)this.mappedByteBuffer.force();
异步刷盘
messagequeue
储存地址
写入内存
对应
本质上是消费者组,找到topic对应的messagequeue所在的broker,然后通过messagequeue找到对应的consumequeue,拿到消息的偏移量,再从commitlog中读取消息
MappedByteBuffer内存映射缓冲区
定期清理
磁盘
doReput方法
消息消费
消费消息
引用
1.先写入System Page Cache2.OS线程异刷入到磁盘commitlog文件3.commitlog文件采用顺序写入4.存在消息丢失的风险,broker宕机,消息会丢失
1.先写入System Page Cache中2.同时将消息刷入到磁盘commitlog文件中3.如果broker宕机,则对于producer来说是消息发送失败4.消息不会丢失,但是吞吐性能会下降
this.cleanConsumeQueueService.run();
broker
topic
先写入系统缓存
CommitLogDispatcherBuildIndex
优先写入
转发消息
commitlog存储位置:rocketmq安装目录,store/commitlog/目录
删除条件1.当磁盘使用率达到85%时触发2.磁盘没满,则每天定时清理超过72小时的文件(包括未消费的消息);broker.conf中deleteWhen指定任务开始执行时间,默认04,也就是凌晨4点
MappedFile映射内存
this.cleanCommitLogService.run();
预读取一部分数据,方便高效查询
对应consumequeue文件,rocketmq安装目录,store/consumequeue目录,目录下以topic名称命名的目录store/consumequeue/{topic}/{queueId}/{fileName}
消息持久化
IndexFile
consumequeue文件储存messagequeue消息在commitlog中的地址,也就是偏移量
异步
消息者机器消费消息完毕后,会回传一个消费进度到broker上,也就是offset,然后broker去存储消费进度。
0 条评论
回复 删除
下一页