RocketMQ源码：ACK机制及消费进度管理

我就是我 2024-03-24 11:29 41阅读 0赞

#### 1. 消息的ACK机制 ####

consumer的每个实例是靠AllocateMessageQueueStrategy队列分配来决定如何消费消息的。那么消费进度具体是如何管理的，又是如何保证消息成功消费的?（RocketMQ有保证消息肯定消费成功的特性,失败则重试）？由于以上工作所有的机制都实现在PushConsumer中，所以本文的原理均只适用于RocketMQ中的PushConsumer即Java客户端中的DefaultPushConsumer。 若使用了PullConsumer模式，类似的工作如何ack，如何保证消费等均需要使用方自己实现。

#### 2. 消费进度管理 ####

在创建消费者添加了一个消费回调监听器：

//并发消费监听器
    consumer.registerMessageListener(new MessageListenerConcurrently() {
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
            //用户自定义业务处理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    复制代码

//顺序消费监听器
    consumer.registerMessageListener(new MessageListenerOrderly() {
    
        AtomicLong consumeTimes = new AtomicLong(0);
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,ConsumeOrderlyContext context) {
            
    		//用户业务处理
      
            return ConsumeOrderlyStatus.SUCCESS;
    
        }
    });
    复制代码

在执行完成监听器的业务逻辑后根据返回的状态客户端做后续的处理，这里分为两种：

1.  ConsumeOrderlyStatus(SUCCESS,SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT 其他的被Deprecated标记)
2.  ConsumeConcurrentlyStatus(CONSUME\_SUCCESS,RECONSUME\_LATER)

#### 3.并发消费进度管理 ####

并发消费主要通过ConsumeMessageConcurrentlyService来处理。ConsumeMessageConcurrentlyService\#processConsumeResult处理消费的。第一部分根据消费状态统计消费成功和消费失败的TPS信息：

switch (status) {
        case CONSUME_SUCCESS:
            if (ackIndex >= consumeRequest.getMsgs().size()) {
                ackIndex = consumeRequest.getMsgs().size() - 1;