RocketMQ中死信队列问题排查与实战解决

# 登录到Broker节点，查看死信Topic
bin/mqadmin topics -n localhost:9876 | grep DLQ

# 统计死信消息堆积
bin/mqadmin statsMessage -n localhost:9876 -t ORDER_PAY_CALLBACK_DLQ

public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-dlq-replay");
    consumer.setNamesrvAddr("localhost:9876");
    consumer.subscribe("ORDER_PAY_CALLBACK_DLQ", "*");
    consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) -> {
        for (MessageExt msg : msgs) {
            System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
}

@RocketMQMessageListener(topic = "ORDER_PAY_CALLBACK", consumerGroup = "OrderPayGroup")
public class OrderPayConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        try {
            OrderDto dto = JSON.parseObject(message, OrderDto.class);
            // 处理逻辑
        } catch (JsonSyntaxException e) {
            log.error("消息格式错误，直接丢弃，message={} ", message, e);
            // 直接消费成功，避免进入DLQ
        } catch (Exception e) {
            log.error("处理失败，触发重试", e);
            throw new RuntimeException(e);
        }
    }
}

rocketmq:
  name-server: localhost:9876
  consumer:
    order-pay:
      group: OrderPayGroup
      topic: ORDER_PAY_CALLBACK
      dlqTopic: ${rocketmq.consumer.order-pay.group}-DLQ

consumer.subscribe(dlqTopic, "*");

# 延迟等级：1 5s,2 10s,3 30s...
consumer.setMaxReconsumeTimes(6);
Message newMsg = msg.clone();
newMsg.setDelayTimeLevel(3);

# 批量重放脚本示例
pip install rocketmq-client-python
python replay_dlq.py --group OrderPayGroup --topic ORDER_PAY_CALLBACK