首页 > 资讯 > 后端开发 > Python >RocketMq消息重试机制及死信队列详解

683

分享到

RocketMq消息重试机制及死信队列详解

2024-04-02 19:04:59 683人浏览泡泡鱼

Python 官方文档：入门教程 => 点击学习

摘要

目录生产者消息重试消费者消息重试并发消费顺序消费并发消费和顺序消费区别死信队列实践出真知公共部分创建测试并发消费并发消费状态测试顺序消费顺序消费状态测试死信队列死信队列特性生产者消息

生产者消息重试

消息队列中的消息消费时并不能保证总是成功的，那失败的消息该怎么进行消息补偿呢？这就用到今天的主角消息重试和死信队列了。

有时因为网路等原因生产者也可能发送消息失败，也会进行消息重试，生产者消息重试比较简单，在SpringBoot中只要在配置文件中配置一下就可以了。

# 异步消息发送失败重试次数,默认为2
RocketMQ.producer.retry-times-when-send-async-failed=2
# 消息发送失败重试次数,默认为2
rocketMQ.producer.retry-times-when-send-failed=2

也可以通过下面这种方式配置

DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
defaultMQProducer.setRetryTimesWhenSendFailed(2);
defaultMQProducer.setRetryTimesWhenSendAsyncFailed(2);

消费者消息重试

Apache RocketMQ 有两种消费模式：集群消费模式和广播消费模式。消息重试只针对集群消费模式生效；广播消费模式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。

同时RocketMq Push消费提供了两种消费方式：并发消费和顺序消费。

并发消费

在并发消费中，可能会有多个线程同时消费一个队列的消息，因此即使发送端通过发送顺序消息保证消息在同一个队列中按照FIFO的顺序，也无法保证消息实际被顺序消费，所有并发消费也可以称之为无序消费。

顺序消费

顺序消费是消息生产者发送过来的消息会遵循FIFO队列的思想，先进先出有顺序的消费消息。对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。

并发消费和顺序消费区别

顺序消费和并发消费的重试机制并不相同，顺序消费消费失败后会先在客户端本地重试直到最大重试次数，这样可以避免消费失败的消息被跳过，消费下一条消息而打乱顺序消费的顺序，而并发消费消费失败后会将消费失败的消息重新投递回服务端，再等待服务端重新投递回来，在这期间会正常消费队列后面的消息。

并发消费失败后并不是投递回原Topic，而是投递到一个特殊Topic，其命名为%RETRY%ConsumerGroupName，集群模式下并发消费每一个ConsumerGroup会对应一个特殊Topic，并会订阅该Topic。

两者参数差别如下

消费类型	重试间隔	最大重试次数
顺序消费	间隔时间可通过自定义设置，SuspendCurrentQueueTimeMillis	最大重试次数可通过自定义参数MaxReconsumeTimes取值进行配置。该参数取值无最大限制。若未设置参数值，默认最大重试次数为Integer.MAX
并发消费	间隔时间根据重试次数阶梯变化，取值范围：1秒～2小时。不支持自定义配置	最大重试次数可通过自定义参数MaxReconsumeTimes取值进行配置。默认值为16次，该参数取值无最大限制，建议使用默认值

并发消费重试间隔如下：

第几次重试	与上次重试的间隔时间	第几次重试	与上次重试的间隔时间
1	10s	9	7min
2	30s	10	8min
3	1min	11	9min
4	2min	12	10min
5	3min	13	20min
6	4min	14	30min
7	5min	15	1h
8	6min	16	2h

死信队列

当一条消息初次消费失败，RocketMQ会自动进行消息重试，达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息。此时，该消息不会立刻被丢弃，而是将其发送到该消费者对应的特殊队列中，这类消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue），死信队列是死信Topic下分区数唯一的单独队列。如果产生了死信消息，那对应的ConsumerGroup的死信Topic名称为%DLQ%ConsumerGroupName，死信队列的消息将不会再被消费。可以利用RocketMQ Admin工具或者RocketMQ Dashboard上查询到对应死信消息的信息。

实践出真知

Talk is cheap,show you the code.

公共部分创建

配置文件

rocketmq.name-server=localhost:9876
# 消费者组
rocketmq.producer.group=producer_group
rocketmq.consumer.topic=consumer_topic
rocketmq.consumer.group=consumer_group

创建消费者RetryConsumerDemo

@Component
public class RetryConsumerDemo {
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;
    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
    @PostConstruct
    public void start() {
        try {
            consumer.setNamesrvAddr(namesrvAddr);
            //设置集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //设置消费超时时间(分钟)
            consumer.setConsumeTimeout(1);
            //订阅主题
            consumer.subscribe(topic , "*");
            //注册消息监听器
            consumer.reGISterMessageListener(new MessageListenerConcurrentlyImpl());
            //最大重试次数
            consumer.setMaxReconsumeTimes(2);
            //启动消费端
            consumer.start();
            System.out.println("Retry Consumer Start...");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}

测试并发消费

创建并发消费监听类并发消费监听类要实现MessageListenerConcurrently类

public class MessageListenerConcurrentlyImpl implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List&lt;MessageExt&gt; msgs, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt message = msgs.get(0);
        try {
            final LocalDateTime now = LocalDateTime.now();
            //逐条消费
            String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("当前时间："+now+"， messageId: " + message.getMsgId() + ",topic: " +
                    message.getTopic()  + ",messageBody: " + messageBody);
            //模拟消费失败
            if ("Concurrently_test".equals(messageBody)) {
                int a = 1 / 0;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}

注册监听类在消费者类RetryConsumerDemo中注册监听类

//注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrentlyImpl());

测试

@RunWith(springRunner.class)
@SpringBootTest(classes = RocketmqApplication.class)
class RocketmqApplicationTests {
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Autowired
    private RocketMQtemplate rocketMQTemplate;
    @Test
    public void testProducer(){
        String msg = "Concurrently_test";
        rocketMQTemplate.convertAndSend(topic , msg);
    }
}

测试结果：

后面重试时间太长就不做测试了，可以看到并发消费的消息时间都是按照上面那张时间间隔表来。

然后通过RocketMq Dashboard Topic一栏可以看到有一个重试消费者组%RETRY%consumer_group，这个消费者组内存放的就是consumer_group消费者组消费失败重试的消息。

并发消费的重试次数是可以修改的，重试次数对应参数DefaultMQPushConsumer类的maxReconsumeTimes属性，maxReconsumeTimes默认是-1，也就是默认会重试16次；

0代表不重试，只要失败就会放入死信队列；

1-16重试次数对应着上面时间间隔表中对应次数。

配置的最大重试次数超过16就按16处理。

并发消费状态

并发消费有两个状态CONSUME_SUCCESS和RECONSUME_LATER。返回CONSUME_SUCCESS代表着消费成功，返回RECONSUME_LATER代表进行消息重试。

public enum ConsumeConcurrentlyStatus {
    
    CONSUME_SUCCESS,
    
    RECONSUME_LATER;
}

当MessageListenerConcurrently接口的consumeMessage方法返回ConsumeConcurrentlyStatus#RECONSUME_LATER、null或者方法抛异常了，都会进行消息重试。当然还是推荐返回ConsumeConcurrentlyStatus#RECONSUME_LATER。

测试顺序消费

顺序消费和并行消费其实都差不多的，只不过顺序消费实现的是MessageListenerOrderly 接口

创建顺序消费监听类

public class MessageListenerOrderlyImpl implements MessageListenerOrderly {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List&lt;MessageExt&gt; msgs, ConsumeOrderlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeOrderlyStatus.SUCCESS;
        }
        MessageExt message = msgs.get(0);
        try {
            final LocalDateTime now = LocalDateTime.now();
            //逐条消费
            String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("当前时间："+now+"， messageId: " + message.getMsgId() + ",topic: " +
                    message.getTopic()  + ",messageBody: " + messageBody);
            //模拟消费失败
            if ("Orderly_test".equals(messageBody)) {
                int a = 1 / 0;
            }
            return ConsumeOrderlyStatus.SUCCESS;
        } catch (Exception e) {
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }
}

注册监听类

//最大重试次数
consumer.setMaxReconsumeTimes(2);
//顺序消费 重试时间间隔
consumer.setSuspendCurrentQueueTimeMillis(2000);

SuspendCurrentQueueTimeMillis表示重试的时间间隔，默认是1s，这里修改成2s

测试

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RocketmqApplication.class)
class RocketmqApplicationTests {
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Test
    public void testProducer(){
        String msg = "Orderly_test";
        rocketMQTemplate.convertAndSend(topic , msg);
    }
}

测试结果：

可以看到三条结果，第一条是第一次消费的，其余两条是隔了2s重试的。重试2次之后这条数据就进入了死信队列。

顺序消费状态

顺序消费目前也是两个状态：SUCCESS和SUSPEND_CURRENT_QUEUE_A_MOMENT。SUSPEND_CURRENT_QUEUE_A_MOMENT意思是先暂停消费一下，过SuspendCurrentQueueTimeMillis时间间隔后再重试一下，而不是放到重试队列里。

public enum ConsumeOrderlyStatus {
    
    SUCCESS,
    
    @Deprecated
    ROLLBACK,
    
    @Deprecated
    COMMIT,
    
    SUSPEND_CURRENT_QUEUE_A_MOMENT;
}