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

springcloud rocketmq 新增的消费者组从哪里开始消费

news 文章来源：https://blog.csdn.net/GDUT_xin/article/details/140636176 2024/9/20 13:39:42

如果新建一个新的消费者组，是否会消费历史消息，导致重复消费？

直接在 console 界面新增消费者组，但是没有办法绑定订阅关系，没有找到入口，在控制台项目源码 rocketmq-externals 也没有找到可以确定订阅关系的接口，在阿里云的生产控制台也没有绑定的入口。
在这里插入图片描述

所以只能是消费者启动后再注册订阅关系。
消费者从哪里消费的计算：
RebalancePushImpl.java

默认走的是：CONSUME_FROM_LAST_OFFSET 规则，按照官方说法，是从最后的消费位点开始继续消费。
关键的获取消费位点的逻辑：readOffset方法：
RemoteBrokerOffsetStore.java
在这里插入图片描述
集群模式下，是从远程获取的偏移量，跟据 fetchConsumeOffsetFromBroker 方法：

报错，其实就是服务端没有该消费者组的offset，被catch住默认返回 -1.
又不是重试队列，所以拿最大的偏置，broker-a queue-8 的 brokerOffset 是 25
在这里插入图片描述
出来到了 RebalanceImpl.java 的 updateProcessQueueTableInRebalance 方法。

然后会被添加到 pullRequestList 通过 this.dispatchPullRequest(pullRequestList)

控制台topic消费进度中已经保存了新的消费者组的消费进度，但 consumeOffset都是 0，还有 759 个消息没有消费。
在这里插入图片描述

消费者消费了一些比较早前的消息：
在这里插入图片描述

消费进度也随之更新。
在这里插入图片描述
为什么和官方的说法不一致呢？CONSUME_FROM_LAST_OFFSET 为什么没有起到作用？
参考官方的修复：Fix CONSUME_FROM_LAST_OFFSET mode may pull data from 0L #4909

~~
至于这个console怎么看？参考以下：
在rocketmq的控制台中，选择 topic -> consumer manage，就可以查看一个主题下的消费者组、集群、队列的消费情况。
在这里插入图片描述
其中，10.122.24.41 是本人的内网ip，如果我本地线程卡住了（或者debug中），这个在线状态也会下线的。目前我是分配到了其中的8个集群队列，broker-a（8~15）。

offsetTable 的内容和我所描述的一致。
此外：
我还有对比组，是之前创建的废弃的消费者组，集群位点 brokerOffset 不变，消费位点 consumerOffset 落后了许多，落后的总量 diffTotal 代表此消费者组还有这么多未消费的消息。而且也没有在线的消费者客户端 consumerClient。
在这里插入图片描述
如果这时，我配置启动消费者去消费此消费者组。预计会消费 delay = 294 个消息。
结果也确实如此，将消息消费完，而且分配到了所有集群的所有队列。

测试结果：默认策略会从offset = 0 开始消费。
在这里插入图片描述
所以该参数没有用，还是从0开始消费了，这时候只能靠消费者组重置位点操作了。

~~
tags过滤，是服务端过滤，为什么会直接将不需要的消息也丢失掉呢？
这就要涉及到订阅关系一致性。
在这里插入图片描述

参考：https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-5-x-series/developer-reference/message-filtering

tags过滤会将不匹配的直接跳过（丢失）
我的理解是，现在没有为某个tags有单独记录消费进度的地方，所谓的服务端过滤，也只是说用hashcode快速匹配拉取而已，之后也是直接将offset拉到队列尾的。

参考：https://mp.weixin.qq.com/s/RnS675dt-wErnEuolK6Zeg?spm=a2c6h.12873639.article-detail.18.3ba035175CHVos

相关文章：