当前位置：首页 > news >正文

源码解析：Apache RocketMQ重置消费位点

news 2025/7/10 14:13:54

引入

reset offset，即重置消费进度，一般在以下场景中使用：

需要重新消费已经消费过的消息，重置到最早位置或根据时间进行重置。
消息积压，不需要消费积压的消息，重置到最新位置，使其从最新位置开始消费。

重置到最早、或者根据时间进行重置与消息补发的区别？
● 消息补发是将原先的消息由生产者重发一次，与区别的那边消息本质上不是同一条消息（除了消息体一样以外）。
● 重置操作是操作消费位点（offset），本质上还是消费生产者之前发送的那条消息。
源码解析

重置offset起始调用位置：

org.apache.rocketmq.tools.admin.DefaultMQAdminExt#resetOffsetByTimestamporg.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl#resetOffsetNewConcurrent

区别：
org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl#resetOffsetNewConcurrent
● 这个看看用来并发的重置消费者的offset。可以多个consumer、多个queue可以同时进行处理。
org.apache.rocketmq.tools.admin.DefaultMQAdminExt#resetOffsetByTimestamp
● 用来根据给定的时间戳来重置消费者的偏移量。

这两个入口本质上都是resetOffset，没有本质上的区别，我们以resetOffsetNewConcurrent为例，具体流程如下图：
在这里插入图片描述

首先是examineTopicRouteInfo：主要是获取topic的路由信息，如果路由信息不存在，则无法进行后续操作。
再者是InvokeBrokerToResetOffset：根据上一步拿到的路由信息，遍历路由，一次向broker发起调用。
请求到达服务端（Broker端），判断是否是Broker端侧处理 ○ Broker端处理：
- 前置检查（look-ahead check）：判断当前BrokerRole是否正确、检验当前Topic、ConsumerGroup是否存在，不满足任意条件，直接返回。
- 将传递过来的offset或者根据timestamp查询到的offset统一放置到queueOffsetMap中
- assignResetOffset：将上一步的queueOffsetMap的offset放到对应的resetOffsetTable和offsetTable中。
- 最后prepare reset result并返回response。
Client端处理：
- 先执行queryOffset：查询当前topic下的group下offsetTable中是是否存储了offset信息，有就返回对应的值，没有返回-1；
- 前置检查（look-ahead check）：检查上一步返回结果consumerOffset是否为-1，为-1表示当前group不存在；检查timeStampOffset是否满足条件；满足上述所有条件将timeStampOffset/consumerOffset中较为小的值放到offsetTable中，如果是C的客户端，直接将timestampOffset放入offsetTable中。
- 请求到达客户端后，先将对应的consumer挂起（suspend），清除ProcessQueue中的消息，在sleep 10s。
- 再执行updateConsumeOffset：更新consumerOffset。
- 最后再resume，继续消费。

补充：
如果是服务端重置，重置之后的offset会写入resetOffsetTable中，在后续进行拉取操作的时候会删除resetOffsetTable中对应的offset；如果queryThenEraseResetOffset中有返回值，将resetOffset作为GetMessageResult的nextBeginOffset，拉取操作用的offset。

public Long queryThenEraseResetOffset(String topic, String group, Integer queueId) {String key = topic + TOPIC_GROUP_SEPARATOR + group;ConcurrentMap<Integer, Long> map = resetOffsetTable.get(key);if (null == map) {return null;} else {return map.remove(queueId);}
}