一、概述

Producer对发送失败的消息进行重新发送的机制，称为消息发送重试机制，也称为消息重投机制。

对于消息重投，需要注意以下几点：

1. 生产者在发送消息时，若采用同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的;
2. 只有普通消息具有发送重试机制，顺序消息是没有的;
3. 消息重投机制可以保证消息尽可能发送成功、不丢失，但可能会造成消息重复。消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会成为大概率事件。

二、重试策略

消息发送重试有三种策略可以选择：同步发送失败策略、异步发送失败策略和消息刷盘失败策略。

2.1 同步发送失败策略

普通消息，消息发送默认采用round-robin策略来选择所发送到的队列。如果发送失败，默认重试2次。但在重试时是不会选择上次发送失败的Broker，而是选择其它Broker。

DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 设置同步发送失败时重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置发送超时时限为5s，默认3s
producer.setSendMsgTimeout(5000);

如果超过重试次数，则抛出异常，由Producer去保证消息不丢。

当Producer出现RemotingException、MQClientException、MQBrokerException时，Producer会自动重投消息。

2.2 异步发送失败策略

异步发送失败重试时，异步重试不会选择其他Broker，仅在当前Broker上做重试，所以该策略无法保证消息不丢失。

DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);

2.3 消息刷盘失败策略

消息刷盘超时(Master、Slave)，默认是不会将消息尝试发送到其他Broker。对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启。

三、生产者消息重试

有时因为网路等原因生产者也可能发送消息失败，也会进行消息重试，生产者消息重试比较简单，在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);

四、消费者消息重试

消费者消费某条消息失败后，会根据消息重试机制将该消息重新投递，若达到重试次数后消息还没有成功被消费，则消息将被投入死信队列。

一条消息无论重试多少次，这些重试消息的Message ID不会改变。

4.1 顺序消息的消费重试

顺序消息，当Consumer消费消息失败后，为了保证消息的顺序性，其会自动不断地进行消息重试，直到消费成功。消费重试默认间隔时间为1000ms。重试期间应用会出现消息消费被阻塞的情况。

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
// 顺序消息消费失败的消费重试时间间隔，单位毫秒，默认为1000，其取值范围为[10, 30000]
consumer.setSuspendCurrentQueueTimeMillis(100);

由于对顺序消息的重试是无休止的，不间断的，直至消费成功，所以，对于顺序消息的消费，务必要保证应用能够及时监控并处理消费失败的情况，避免消费被永久性阻塞。

注意：顺序消息没有发送失败重试机制，但具有消费失败重试机制

4.2 并发消息的消费重试

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

对于无序消息(普通消息、延时消息、事务消息)，当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的效果。

注意：无序消息的重试只针对集群消费模式生效；广播消费模式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。

对于无序消息集群消费下的重试消费，默认允许每条消息最多重试16次，如果消息重试16次后仍然失败，消息将被投递至死信队列。消息重试间隔时间如下：

某条消息在一直消费失败的前提下，将会在接下来的4小时46分钟之内进行16次重试，超过这个时间范围消息将不再重试投递，而被投递至死信队列。

修改消费重试次数：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
// 修改消费重试次数
consumer.setMaxReconsumeTimes(10);

4.3 并发消费和顺序消费区别

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

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

两者参数差别如下：

1. 并发消费状态

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

public enum ConsumeConcurrentlyStatus {

    /**
     * Success consumption
     */
    CONSUME_SUCCESS,

    /**
     * Failure consumption,later try to consume
     */
    RECONSUME_LATER;
}

2. 顺序消费状态

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

public enum ConsumeOrderlyStatus {
    /**
     * Success consumption
     */
    SUCCESS,
    
    /**
     * Rollback consumption(only for binlog consumption)
     */
    @Deprecated
    ROLLBACK,
    
    /**
     * Commit offset(only for binlog consumption)
     */
    @Deprecated
    COMMIT,
    
    /**
     * Suspend current queue a moment
     */
    SUSPEND_CURRENT_QUEUE_A_MOMENT;
}

五、配置

5.1 自定义消息最大重试次数

允许Consumer启动的时候设置最大重试次数，重试时间间隔将按照以下策略：

1. 最大重试次数小于等于16次，则重试时间间隔同上表描述
2. 最大重试次数大于16次，超过16次的重试时间间隔均为每次2小时
   • 消息最大重试次数的设置对相同Group ID下的所有Consumer实例有效
   • 如果只对相同Group ID下两个Consumer实例中的其中一个设置，那么该配置对两个Consumer实例均生效
   • 配置采用覆盖的方式生效，即最后启动的Consumer实例会覆盖之前启动的实例的配置

5.2 消费重试配置

集群消费模式下，消息消费失败后期望消息重试，需要在消息监听器接口的实现中明确进行配置(三种方式任选一种)：

1. 返回ConsumeConcurrentlyStatus.RECONSUME_LATER(推荐)
2. 返回Null
3. 抛出异常

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
        ConsumeConcurrentlyContext context) {
        //消息处理逻辑抛出异常，消息将重试。
        doConsumeMessage(message);

        //方式1：返回Action.ReconsumeLater，消息将重试。
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;

        //方式2：返回null，消息将重试。
        return null;

        //方式3：直接抛出异常，消息将重试。
        throw new RuntimeException("Consumer Message exception");
    }
});

5.3 消费不重试配置

集群消费模式下，消息失败后期望消息不重试，需要捕获消费逻辑中可能抛出的异常，最终返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS，此后这条消息将不会再重试。

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
        ConsumeConcurrentlyContext context) {
        try {
            doConsumeMessage(message);
        } catch (Throwable e) {
            //捕获消费逻辑中的所有异常，并返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        //消息处理正常，直接返回消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

5.4 获取消息重试次数

消费者收到消息后，可按照以下方式获取消息的重试次数：

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    ConsumeConcurrentlyContext context) {
    
    for (MessageExt msg : msgs) {
        //获取消息的重试次数。
        System.out.println(msg.getReconsumeTimes());
    }

    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

六、死信队列

当一条消息初次消费失败，消息队列会自动进行消费重试；达到最大重试次数后(默认16次)，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

正常情况下无法被消费的消息称为死信消息(Dead-Letter Message)，存储死信消息的特殊队列称为死信队列(Dead-Letter Queue)。

6.1 死信消息特性

不会再被消费者正常消费
有效期与正常消息相同，均为3天，3天后会被自动删除

6.2 死信队列特性

• 一个死信队列对应一个Group ID，而不是对应单个消费者实例。名称为%DLQ%consumerGroup@consumerGroup
• 如果一个Group ID未产生死信消息，则不会为其创建相应的死信队列
• 一个死信队列包含了对应Group ID产生的所有死信消息，不论该消息属于哪个Topic

七、案例

公共部分创建

1. 配置文件

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

rocketmq.consumer.topic=consumer_topic
rocketmq.consumer.group=consumer_group

2. 创建消费者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();
        }
    }
}

7.1 测试并发消费

7.1.1 创建并发消费监听类

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

public class MessageListenerConcurrentlyImpl implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> 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;
        }
    }
}

7.1.2 注册监听类

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

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

7.1.3 测试

@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处理。

7.2 测试顺序消费

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

7.2.1 创建顺序消费监听类

public class MessageListenerOrderlyImpl implements MessageListenerOrderly {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> 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;
        }
    }
}

7.2.2 注册监听类

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

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

7.2.3 测试

@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次之后这条数据就进入了死信队列。

7.3 测试死信队列

并发消费和顺序消费达到了最大重试次数之后就会放到死信队列。死信队列在一开始是不会被创建的，只有需要的时候才会被创建。就拿上面测试结果来看，进入到的死信队列就是%DLQ%consumer_group，进入死信队列的消息要收到处理。

本栏目推荐文章
• 从C++CLI工程的依赖库引用问题看.Net加载程序集机制
• Go语言defer的延迟执行机制
• 基于k8s Deployment的弹性扩缩容及滚动发布机制详解
• 22-安全机制：Kubernete 如何保障集群安全？
• js 垃圾回收机制
• SpringBoot中使用SpringRetry实现重试机制(重试调用第三方API)
• 详解GaussDB(DWS)通信安全的小妙招：连接认证机制
• 什么是刷新缓存机制
• 消息缓存机制有什么好处
• 什么情况用到缓存机制

RocketMQ 机制rocketmq机制 顺序rocketmq机制 rocketmq架构amp机制 rocketmq机制 数据 rocketmq机制 消息 笔记 rocketmq rocketmq5 rocketmq消息 rocketmq docker rocketmq名字