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

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

引入

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

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

重置到最早、或者根据时间进行重置与消息补发的区别?
● 消息补发是将原先的消息由生产者重发一次,与区别的那边消息本质上不是同一条消息(除了消息体一样以外)。
● 重置操作是操作消费位点(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);}
}

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

http://www.lryc.cn/news/261116.html

相关文章:

  • Python 自动化之处理docx文件(一)
  • Vue mixins详解
  • ssl证书问题导致本地启动前端服务报500
  • Rust 学习
  • 1.1 【应用开发】应用开发简介
  • 在windows系统搭建LVGL模拟器(codeblock工程)
  • 2023第十四届蓝桥杯国赛 C/C++ 大学 B 组
  • 如何在页面中加入百度地图
  • Windows VC++提升当前进程权限到管理员权限
  • 算法leetcode|92. 反转链表 II(rust重拳出击)
  • Chapter 7 - 3. Congestion Management in Ethernet Storage Networks以太网存储网络的拥塞管理
  • 优雅玩转实验室服务器(二)传输文件
  • 动态面板简介以及ERP原型图案列
  • 漏刻有时百度地图API实战开发(12)(切片工具的使用、添加自定义图层TileLayer)
  • python 爬虫 m3u8 视频文件 加密解密 整合mp4
  • mybatis中xml文件容易搞混的属性
  • android Retrofit2.0请求 延长超时操作
  • Axure之动态面板轮播图
  • 一文读懂算法中的时间复杂度和空间复杂度,O(1)、O(logn)、O(n)、O(n^2)、O(2^n) 附举例说明,常见的时间复杂度,空间复杂度
  • LWIP热插拔功能实现
  • android下的app性能测试应主要针对那些方面,如何开展?
  • 【深度学习】注意力机制(二)
  • 学习黑马vue
  • gdb本地调试版本移植至ARM-Linux系统
  • 《Linux C编程实战》笔记:实现自己的ls命令
  • Python个人代码随笔(观看无益,请跳过)
  • Unity中实现ShaderToy卡通火(总结篇)
  • 等保2.0的变化
  • 漏洞复现-网神SecGate3600防火墙敏感信息泄露漏洞(附漏洞检测脚本)
  • ArkTS入门