RocketMQ 广播模式 消息堆积

Question

背景：项目使用RocketMQ的广播模式进行消费；
问题：每个结点消费完毕后，rocketmq消息监控上依旧显示消息没被消费掉，随着消息的增多，消息开始堆积；
每个节点消费完毕后，怎么清除这些消息？

Answer 1

引自免费微信小程序：皆我百晓生

在RocketMQ中，广播模式下每个消费者都会收到消息的拷贝，且broker层面不会去检查消息是否被所有消费者都消费了。因此，在监控中看到的消息堆积可能是由于监控显示机制导致的，它可能只是表明有消费者接收到了消息，但并不表示消息未被正确处理。

针对您提到的问题，确保每个节点消费完毕后消息状态能够正确更新，可从以下几个方面排查和解决：

1. 消费逻辑确认： 确保消费者消费消息后调用了MessageListener接口中的consumeConcurrently()或consumeOrderly()方法，并返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS（并发消费）或ConsumeOrderlyStatus.SUCCESS（顺序消费）。这样RocketMQ才能认为该消息已被成功消费。

public class MyMessageListener implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        // 消息消费逻辑...
        // 处理完消息后，返回消费成功状态
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

1. 消费幂等性： 在广播模式下，保证消费逻辑的幂等性非常重要，即使一条消息被多个消费者多次消费，也应该达到预期的一致结果。

2. 消费进度管理： RocketMQ会通过Offset来记录消费者的消费进度，若发现消息堆积，可以检查消费者实例是否有正确提交offset。对于Java客户端，这通常是自动进行的，除非你手动关闭了自动提交offset的功能。

3. 检查消费线程与消费能力： 如果消费速度跟不上生产速度，也会造成消息堆积。你需要检查消费者的消费能力和设置的消费线程数是否合理，适当提升消费能力或者增加消费实例以提高并行消费能力。

4. Broker配置检查： 查看Broker端的相关配置，例如 broker.conf 中关于过期策略、队列满后的处理策略等设置，确保它们没有阻止消息被正确清理。

5. 查看消费者状态： 验证消费者实例是否健康运行，查看日志查找潜在问题。

如果上述步骤无法解决问题，请进一步分析具体日志和RocketMQ控制台的详细信息，以便找到更精确的问题定位点。同时，也需注意，广播模式下的消息理论上是会被所有消费者消费一遍，所以“清除”消息的概念并不完全适用，而是应当关注如何确保每个消费者都能正确消费并完成其业务逻辑。