RocketMQ消息读写

前言

之前写过一篇关于RockerMQ的消息存储&＃xff0c;本文接上文&＃xff0c;这里是详细介绍RocketMQ是怎么完成消息读写的&＃xff01;

消息写

一条消息进入到Broker后经历一下几个过程才最终被持久化

1. Broker更具QueueId&＃xff0c;获取到该消息对应索引条目在consumerqueue目录中的写入偏移量&＃xff0c;即QueueOffset
2. 将queueId、queueOffset等数据&＃xff0c;与消息一起封装为消息单元
3. 将消息写入到commitLog
4. 形成消息索引条目
5. 将消息索引条目拆分到相应的consumerqueue

消息读

当Consumer来拉取消息时会经历以下几个步骤

1. Consumer获取到其要消费消息所在的Queue的消息偏移量offset&＃xff0c;计算出其要消费的消息offset

消费offset即消息消费进度&＃xff0c;Consumer对某个queue消费的offset&＃xff0c;即消费到了该queue的第几条消息&＃xff0c;消息offset&＃61;消费offset&＃43;1&＃xff0c;这里实际上就是去读config目录下consumerOffset.json文件

举例topic为first-topic-str&＃xff0c;消费者为my-consumer-group&＃xff0c;订阅队列有queue0-3.对应消费为0号消费到2&＃xff0c;1号消费到2&＃xff0c;2号消费到1&＃xff0c;3号消费到1

2. Consumer向Broker发送拉取请求&＃xff0c;其中会包含拉取消息的Queue&＃xff0c;消息offset及消息Tag
3. Broker计算改ConsumerQueue中的queueOffset

queueOffset&＃61;消息offset*20字节

4. 该queueOffset处开始向后查找第一个指定Tag的索引条目
5. 解析该索引条目前8个字节&＃xff0c;即可定位到该消息在commitLong中的commit Offset
6. 从对应的commitLog offset中读取消息单元&＃xff0c;并发送给Consumer

补充消息点位重置

这里关于consumerOffset.json文件&＃xff0c;这个文件就是记录每个Topic下消费者消息的点位信息的

我们在rocket-console控制台中关于Topic中有一个重置消费位点的概念&＃xff0c;操作如下

用过rocketmq的管理控制平台的同学应该都会看到重置消费位点这个功能&＃xff0c;由于消费消息也只是对应的消费者的指针的偏移&＃xff0c;消息也并没有消失&＃xff0c;所以这个重置消费位点也就是指针的来回移动偏移而已。这里也很好理解&＃xff0c;首先选择消费组&＃xff0c;就是你订阅了这个topic的组&＃xff0c;下边的时间点就是你要讲指针移动到什么时候&＃xff0c;这个好像不能精确到其中的一条消息&＃xff0c;只能通过时间来定位。然后就是重置后你选的消费组就开始从你选择的时间点开始消费消息&＃xff0c;不过呢这个好像有点小bug还是怎么的 我测试的时候它的时间点好像并没有多么精确 大家可以自行测试。
还有一点 最重要的一点&＃xff0c;这个重置消费位点你点过重置后它不会立马重置&＃xff0c;你的消费者也不会立马接收到消息&＃xff0c;点过重置后&＃xff0c;好像会导致消息短暂的失效&＃xff08;这个时间好像是0-3分钟的样子&＃xff0c;所以得等一下&＃xff09;&＃xff0c;那么这里实际上重置的数据就是上面提到的consumerOffset.json文件