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

news2024/9/20 14:38:16

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

直接在 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 个消息。
结果也确实如此，将消息消费完，而且分配到了所有集群的所有队列。