背景

1. 用户向任务中心提交导入/导出任务，提交成功即返回

2. 任务插入时伴随发出任务生成的事务消息

3. 业务服务监听任务生成消息，处理自己的任务类型，完成导入/导出逻辑

4. 过程中实时汇报进度给任务中心、完成之后提交最终产物

本文基于RocketMQ 4.6.0版本进行分析

日志分析

日志里的问题

1. 有3次消费没有结束

2. reconsumeTimes=1的重试消息被消费了5次之多

消费流程分析

2024-03-06 20:10:00 INFO ClientManageThread_19 - unregister a consumer[comment-export-consumer_pre] from consumerGroupInfo ClientChannelInfo [channel=[id: 0x9d51a3dd, L:/172.27.0.2:10911 - R:/10.11.25.28:41834], clientId=10.11.25.28@comment-service-8f4jq, language=JAVA, version=335, lastUpdateTimestamp=1709726979482]
2024-03-06 20:10:54 INFO HeartbeatThread_1 - new consumer connected, group: comment-export-consumer_pre CONSUME_PASSIVELY CLUSTERING channel: ClientChannelInfo [channel=[id: 0xbc2393d9, L:/172.27.0.2:10911 - R:/10.11.25.28:52420], clientId=10.11.25.28@comment-service-8f4jq, language=JAVA, version=335, lastUpdateTimestamp=1709727054905]
2024-03-06 20:12:20 INFO ClientManageThread_21 - unregister a consumer[comment-export-consumer_pre] from consumerGroupInfo ClientChannelInfo [channel=[id: 0x74565fe6, L:/172.27.0.2:10911 - R:/10.11.19.167:49572], clientId=10.11.19.167@comment-service-lhsts, language=JAVA, version=335, lastUpdateTimestamp=1709727113198]
2024-03-06 20:13:19 INFO HeartbeatThread_1 - new consumer connected, group: comment-export-consumer_pre CONSUME_PASSIVELY CLUSTERING channel: ClientChannelInfo [channel=[id: 0x02aaa89c, L:/172.27.0.2:10911 - R:/10.11.19.167:53384], clientId=10.11.19.167@comment-service-lhsts, language=JAVA, version=335, lastUpdateTimestamp=1709727199889]
2024-03-06 20:14:26 INFO ClientManageThread_18 - unregister a consumer[comment-export-consumer_pre] from consumerGroupInfo ClientChannelInfo [channel=[id: 0x02aaa89c, L:/172.27.0.2:10911 - R:/10.11.19.167:53384], clientId=10.11.19.167@comment-service-lhsts, language=JAVA, version=335, lastUpdateTimestamp=1709727239049]
2024-03-06 20:15:25 INFO HeartbeatThread_1 - new consumer connected, group: comment-export-consumer_pre CONSUME_PASSIVELY CLUSTERING channel: ClientChannelInfo [channel=[id: 0x1333b08d, L:/172.27.0.2:10911 - R:/10.11.19.167:55010], clientId=10.11.19.167@comment-service-lhsts, language=JAVA, version=335, lastUpdateTimestamp=1709727325724]
2024-03-06 20:16:11 INFO ClientManageThread_8 - unregister a consumer[comment-export-consumer_pre] from consumerGroupInfo ClientChannelInfo [channel=[id: 0xbc2393d9, L:/172.27.0.2:10911 - R:/10.11.25.28:52420], clientId=10.11.25.28@comment-service-8f4jq, language=JAVA, version=335, lastUpdateTimestamp=1709727366356]
2024-03-06 20:17:05 INFO HeartbeatThread_3 - new consumer connected, group: comment-export-consumer_pre CONSUME_PASSIVELY CLUSTERING channel: ClientChannelInfo [channel=[id: 0xde6ee11f, L:/172.27.0.2:10911 - R:/10.11.25.28:58648], clientId=10.11.25.28@comment-service-8f4jq, language=JAVA, version=335, lastUpdateTimestamp=1709727425990]

场景回溯

文字版

1. 正常消费，耗时40分钟左右，由于产物过大上传oss失败，导致消费失败。此时client正常向broker投递重试消息，重试消息的topic为retry topic。

2. 第一次重试消费，此时retry队列由10.11.25.28消费。消费过程中由于发布第一次代码优化，中断了消费逻辑。此时reconsumeTimes=1的消息未被ack（第1次）

3. 由于是10.11.25.28先发布，发布第一步先要停机，停机又触发了自身从Broker下线，导致retry队列被分配给10.11.19.167 。此时reconsumeTimes= 1的消息被10.11.19.167消费（第2次）

4. 紧接着，10.11.25.28 启动成功，此时又会去Broker上注册，导致retry队列被分配给了自己。此时reconsumeTimes=1的消息被10.11.25.28消费（第3次）

5. 轮到 10.11.19.167 发布，也是先停机，中断了消费逻辑，然后去Broker下线，但此时retry队列本身就不是自己在消费的，所以没有变化。此时reconsumeTimes= 1的消息同样未被ack。

6. 第二次发布，这次先发的是10.11.19.167，此时retry队列本身就不是自己在消费的，所以没有变化。

7. 轮到10.11.25.28 发布，先停机，导致retry队列被分配给10.11.19.167。此时消息又一次被10.11.19.167消费（第4次）

8. 紧接着，10.11.25.28 停机，导致retry队列被分配给自己。此时消息又一次被10.11.25.28消费（第5次）

• 当retry队列重新分配之后，之前正在消费中的消息并不会受到影响。比如，最后一次虽然队列被分配给了10.11.25.28，但是10.11.19.167 在第7步中消费到的消息还在继续执行整个消费流程。只不过，在消费结束的时候，消费组会识别出此队列并不该由自己处理，所以无论消费结果是成功和失败，client都不会再进行处理（比如失败后需要发送重试消息到broker）

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {

				// 省略部分代码
        @Override
        public void run() {
            if (this.processQueue.isDropped()) {
                log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
                return;
            }
            
            status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);

            // 这里会判断是否这个队列还是归属我这台实例消费，如果不归我，那就不处理消费结果
            if (!processQueue.isDropped()) {
                ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
            } else {
                log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
            }
        }
    }
}

2024-03-06 20:56:51,051 WARN RocketmqClient - processQueue is dropped without process consume result. messageQueue=MessageQueue [topic=%RETRY%comment-export-consumer_pre, brokerName=broker-b, queueId=0], msgs=[Messa
geExt [queueId=0, storeSize=1313, queueOffset=32, sysFlag=0, bornTimestamp=1709723220108, bornHost=/10.11.20.52:41884, storeTimestamp=1709725695735, storeHost=/10.10.1.36:10911, msgId=0A0A012400002A9F00000071E6B743C
A, commitLogOffset=489202074570, bodyCRC=1074741402, reconsumeTimes=1, preparedTransactionOffset=0, toString()=Message{topic='haoduo-service', flag=0, properties={TRACE_ID=65e84e52603377430cff3912, CONSUME_START_TIM
E=1709727371370, GRAY_TAG=pre, MIN_OFFSET=16, REAL_TOPIC=%RETRY%comment-export-consumer_pre, ORIGIN_MESSAGE_ID=0A0A012400002A9F00000071E4456D50, RETRY_TOPIC=haoduo-service, MAX_OFFSET=33, KEYS=218487139854566, id=31
11b110-0b90-003e-55dc-f635e4d64781, UNIQ_KEY=0A0B14340000685F4C2E1DD9E8880000, CLUSTER=DefaultCluster, WAIT=false, contentType=application/json, DELAY=3, TAGS=lesson-feedback-img-export, timestamp=1709723220100, REA
L_QID=0}, body

• RocketMQ消费组内消费队列的分配是客户端行为，broker并不参与分配的过程。broker只是在实例上下线的时候会对整个消费组实例做广播通知。由客户端采用统一的算法去给自己分配队列。

图示版

one more thing

// org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService#cleanExpireMsg
private void cleanExpireMsg() {
    Iterator<Map.Entry<MessageQueue, ProcessQueue>> it =
        this.defaultMQPushConsumerImpl.getRebalanceImpl().getProcessQueueTable().entrySet().iterator();
    while (it.hasNext()) {
        Map.Entry<MessageQueue, ProcessQueue> next = it.next();
        ProcessQueue pq = next.getValue();
        pq.cleanExpiredMsg(this.defaultMQPushConsumer);
    }
}

public @interface RocketMQMessageListener {

/**
     * Max consumer timeout, default 30s.
     */
long consumeTimeout() default 30000L;
}

public void start() {
    this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            cleanExpireMsg();
        }

    }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
}