前天发完版本好不容易休息一下,又遇到一个问题,项目组反馈说RocketMQ的一个消费组一条消息,消费了两次,但两者之间的间隔超过了10个小时,现象如下图所示:
这是为什么呢?两者之间相差了差不多10个多小时,是不是这条消息重复消费了16次,但从日志中并没有打印出16次消息题,只打印了两条消息,从日志角度来看应该不是重复消费了16次。
本着不轻易相信日志的原则,我觉得应该去RocketMQ服务器上看看这条消息存储的重试次数,从而推断出消息是否消费失败而进行消费重试,正好项目组也提供了消息的Key,根据消息key、发送主题名称去查找消息,只能查询到一条消息:
说明发送端确实只发送了一条消息。
那个时候服务集群并没有发现什么异常,消费者没有重启、队列也没发生重平衡,不符合RocketMQ会重复推送消息给客户端的场景,那基本就可以断定是消息消费失败重启引起的,但RocketMQ消息消费重试重试延迟是采取间隔的,往往第一次重试只需10秒就会发生重试,不应该是10个小时?
为此,我特意根据Key分别查找了主题SCHEDULE_TOPIC_XXXX与%RETRY%重试主题,发现这个key也只有一条消息,说明这还仅仅是第一次重试,如下图:
但从这里可以看到消息写入到SCHEDULE_TOPIC_XXXX的时间为2022-05-27 22:26:20,然后过了10s就通过调度机制进入到重试主题,并开始被消费,故进行了第一次重试。
扩展机制知识:RocketMQ并发消费的消费者,在客户端消费失败后会向服务端发送ACK,根据重试次数进入到SCHEDULE_TOPIC_XXXX主题不同的队列中,每一个队列代表的延迟时间不一样,经过一定延迟时间后再次调度到消费组的重试主题,被消费者再次消费,实现时间间隔的重试,提高重试成功率。
那为什么第一次消费是12点51分,为什么这么久才进入到Broker的SCHEDULE_TOPIC_XXXX呢?
这个就是问题的关键所在,这个细节我平时也没关注,故接下来得分析这段代码,代码具体定义在ConsumeMessageConcurrentlyService的processConsumeResult方法中。
原来,如果客户端消费失败后向Broker发生ACK失败后,会加入到msgBack失败队列中,并重新提交到消费者这边消费,并且这条消息的位点不会提交,因为有关键代码:
consumeRequest.getMsgs().removeAll(msgBackFailed); long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs()); if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) { this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true); }
RocketMQ消费位点采取最小位点提交,只要消息存在于本地处理队列,位点就不会提交,从而会触发消息积压。但尴尬的是这个主题平时消息量很少,并没有通过积压来发现该问题。
根据现象与代码结合,原因是客户端一直消费失败,但向Broker提交ACK一直失败,直到晚上22:26分才发送成功,从而才触发重新消费。
那现在的关键是为什么会发生ACK会失败呢?这次比较遗憾,因为项目组使用的是容器,这块代码并没有采集到日志集群,导致无法查看错误日志,而且所在的集群是正常的