消息队列 RocketMQ：（九）消息重试

2023-11-02

文章目录

消息队列 RocketMQ：（一）概述

消息队列 RocketMQ：（二）系统架构

消息队列 RocketMQ：（三）发送普通消息（三种方式）

消息队列 RocketMQ：（四）顺序消息

消息队列 RocketMQ：（五）延时消息

消息队列 RocketMQ：（七）批量消息

消息队列 RocketMQ：（八）消息过滤

七、消息发送重试

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

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

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

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

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、异步发送失败策略

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

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

3、消息刷盘失败策略

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

八、消息消费重试

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

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

1、顺序消息的消费重试

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

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

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

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

2、无序消息的消费重试

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

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

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

重试次数	与上次重试的间隔时间	重试次数	与上次重试的间隔时间
1	10秒	9	7分钟
2	30秒	10	8分钟
3	1分钟	11	9分钟
4	2分钟	12	10分钟
5	3分钟	13	20分钟
6	4分钟	14	30分钟
7	5分钟	15	1小时
8	6分钟	16	2小时

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

修改消费重试次数：

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

3、自定义消息最大重试次数

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

最大重试次数小于等于16次，则重试时间间隔同上表描述
最大重试次数大于16次，超过16次的重试时间间隔均为每次2小时

消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效

如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置，那么该配置对两个 Consumer 实例均生效

配置采用覆盖的方式生效，即最后启动的Consumer实例会覆盖之前启动的实例的配置

4、消费重试配置

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

返回 ConsumeConcurrentlyStatus.RECONSUME_LATER（推荐）
返回 Null
抛出异常

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、消费不重试配置

集群消费模式下，消息失败后期望消息不重试，需要捕获消费逻辑中可能抛出的异常，最终返回 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;
    }
});

6、获取消息重试次数

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

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

九、死信队列

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

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

1、死信消息特性

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

2、死信队列特性

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

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)