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

发布时间:2023年12月17日

引入

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

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

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

重置offset起始调用位置:

org.apache.rocketmq.tools.admin.DefaultMQAdminExt#resetOffsetByTimestamp

org.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);
    }
}

参考:
● https://rocketmq.apache.org/
● https://github.com/apache/rocketmq

文章来源:https://blog.csdn.net/weixin_42339552/article/details/135033624
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。