devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RocketMQ详解(12)——RocketMQ的重试机制_rocketmq重试队列消息翻倍

作者：盐析白兔 | 2024-05-21 17:39:40

赞

踩

rocketmq重试队列消息翻倍

RocketMQ详解(12)——RocketMQ的重试机制

一. MQ的重试机制

由于MQ经常处于复杂的分布式系统中，考虑网络波动、服务宕机、程序异常因素，很有可能出现消息发送或者消费失败的问题。因此，消息的重试就是所有MQ中间件必须考虑到的一个关键点。如果没有消息重试，就可能产生消息丢失的问题，可能对系统产生很大的影响。所以，秉承宁可多发消息，也不可丢失消息的原则，大部分MQ都对消息重试提供了很好的支持。

RocketMQ为使用者封装了消息重试的处理流程，无需开发人员手动处理。RocketMQ支持了生产端和消费端两类重试机制。

二. 生产端重试

  1. 如果由于网络抖动等原因，Producer程序向Broker发送消息时没有成功，即发送端没有收到Broker的ACK，导致最终Consumer无法消费消息，此时RocketMQ会自动进行重试。

  2. 相关API

    DefaultMQProducer可以设置消息发送失败的最大重试次数，并可以结合发送的超时时间来进行重试的处理，具体API如下：

    //设置消息发送失败时的最大重试次数
public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
   this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
}

//同步发送消息，并指定超时时间
public SendResult send(Message msg,
                      long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
   return this.defaultMQProducerImpl.send(msg, timeout);
}
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    因此，实现生产端的重试十分简单，例如下面的代码可以设置Producer如果在5s内没有发送成功，则重试5次：

    //同步发送消息，如果5秒内没有发送成功，则重试5次
DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer");
producer.setRetryTimesWhenSendFailed(5);
producer.send(msg,5000L);
    • 1
    • 2
    • 3
    • 4

三. 消费端的重试

  1. 消费状态：

    消费者消费消息后，需要给Broker返回消费状态。以MessageListenerConcurrently监听器为例，Consumer消费完成后需要返回ConsumeConcurrentlyStatus并发消费状态。查看源码，ConsumeConcurrentlyStatus是一个枚举，共有两种状态：

    public enum ConsumeConcurrentlyStatus {
   //消费成功
   ConsumeConcurrentlyStatus,

   //消费失败，一段时间后重试
   RECONSUME_LATER;
}
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    Consumer端的重试包括两种情况

    1. 异常重试：由于Consumer端逻辑出现了异常，导致返回了RECONSUME_LATER状态，那么Broker就会在一段时间后尝试重试。
    2. 超时重试：如果Consumer端处理时间过长，或者由于某些原因线程挂起，导致迟迟没有返回消费状态，Broker就会认为Consumer消费超时，此时会发起超时重试。

    因此，如果Consumer端正常消费成功，一定要返回ConsumeConcurrentlyStatus.ConsumeConcurrentlyStatus状态。

    下面分别演示两种重试。

  2. 异常重试

    RocketMQ可在broker.conf文件中配置Consumer端的重试次数和重试时间间隔，如下：

    messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    • 1

    但是在大部分情况下，如果Consumer端逻辑出现异常，重试太多次也没有很大的意义，我们可以在代码中指定最大的重试次数。如下：

    package william.rmq.consumer.quickstart;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import william.rmq.common.constant.RocketMQConstant;

import javax.annotation.PostConstruct;
import java.util.List;

/**
* @Auther: ZhangShenao
* @Date: 2018/9/7 11:06
* @Description:RocketMQ消息消费者
*/
@Slf4j
@Service
public class MessageConsumer implements MessageListenerConcurrently {
   @Value("${spring.rocketmq.namesrvAddr}")
   private String namesrvAddr;

   private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultConsumer");


   @PostConstruct
   public void start() {
       try {
           consumer.setNamesrvAddr(namesrvAddr);

           //从消息队列头部开始消费
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

           //设置集群消费模式
           consumer.setMessageModel(MessageModel.CLUSTERING);

           //订阅主题
           consumer.subscribe("DefaultCluster", "*");

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

           //启动消费端
           consumer.start();

           log.info("Message Consumer Start...");
           System.err.println("Message Consumer Start...");
       } catch (MQClientException e) {
           log.error("Message Consumer Start Error!!",e);
       }

   }

   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       if (CollectionUtils.isEmpty(msgs)) {
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }

       MessageExt message = msgs.get(0);
       try {
           //逐条消费
           String messageBody = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
           System.err.println("Message Consumer: Handle New Message: messageId: " + message.getMsgId() + ",topic: " +
                   message.getTopic() + ",tags: " + message.getTags() + ",messageBody: " + messageBody);

           //模拟业务异常
           int i = 1 / 0;
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       } catch (Exception e) {
           log.error("Consume Message Error!!", e);
           //抛出异常时，返回ConsumeConcurrentlyStatus.RECONSUME_LATER，尝试重试。当重试指定次数后返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
           int reconsumeTimes = message.getReconsumeTimes();
           System.err.println("Now Retry Times: " + reconsumeTimes);
           if (reconsumeTimes >= RocketMQConstant.MAX_RETRY_TIMES) {
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           }
           return ConsumeConcurrentlyStatus.RECONSUME_LATER;
       }
   }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92

    可以看到控制台打印如下：

    Now Retry Times: 3
Message Consumer: Handle New Message: messageId: 0A0E096CA14618B4AAC2562C6D5B0000,topic: DefaultCluster,tags: Tags,messageBody: Message-1
Now Retry Times: 3
Message Consumer: Handle New Message: messageId: C0A81FFA7FF318B4AAC24A37C32C0007,topic: DefaultCluster,tags: Tags,messageBody: Order-2-完成
Now Retry Times: 3
Now Retry Times: 3
Message Consumer: Handle New Message: messageId: C0A81FFA7FF318B4AAC24A37C3290006,topic: DefaultCluster,tags: Tags,messageBody: Order-2-支付
Now Retry Times: 3
Now Retry Times: 3

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    消息重试指定的次数后，就返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS不再重试了。

    注：只有在消息模式为MessageModel.CLUSTERING集群模式时，Broker才会自动进行重试，广播消息是不会重试的。

  3. 超时重试

    当Consumer处理时间过长，在超时时间内没有返回给Broker消费状态，那么Broker也会自动重试

    package william.rmq.consumer.quickstart;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import william.rmq.common.constant.RocketMQConstant;

import javax.annotation.PostConstruct;
import java.util.List;

/**
* @Auther: ZhangShenao
* @Date: 2018/9/7 11:06
* @Description:RocketMQ消息消费者
*/
@Slf4j
@Service
public class MessageConsumer implements MessageListenerConcurrently {
   @Value("${spring.rocketmq.namesrvAddr}")
   private String namesrvAddr;

   private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultConsumer");


   @PostConstruct
   public void start() {
       try {
           consumer.setNamesrvAddr(namesrvAddr);

           //从消息队列头部开始消费
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

           //设置集群消费模式
           consumer.setMessageModel(MessageModel.CLUSTERING);

           //设置消费超时时间(分钟)
           consumer.setConsumeTimeout(RocketMQConstant.CONSUMER_TIMEOUT_MINUTES);

           //订阅主题
           consumer.subscribe("DefaultCluster", "*");

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

           //启动消费端
           consumer.start();

           log.info("Message Consumer Start...");
           System.err.println("Message Consumer Start...");
       } catch (MQClientException e) {
           log.error("Message Consumer Start Error!!",e);
       }

   }

   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       if (CollectionUtils.isEmpty(msgs)) {
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }

       MessageExt message = msgs.get(0);
       try {
           //逐条消费
           String messageBody = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
           System.err.println("Message Consumer: Handle New Message: messageId: " + message.getMsgId() + ",topic: " +
                   message.getTopic() + ",tags: " + message.getTags() + ",messageBody: " + messageBody);

           //模拟耗时操作2分钟，大于设置的消费超时时间
           Thread.sleep(1000L * 60 * 2);
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       } catch (Exception e) {
           log.error("Consume Message Error!!", e);
           return ConsumeConcurrentlyStatus.RECONSUME_LATER;
       }
   }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89

四. 消息的幂等去重

由于MQ的重试机制，难免会引起消息的重复消费问题。比如一个ConsumerGroup中有两个，Consumer1和Consumer2，以集群方式消费。假设一条消息发往ConsumerGroup，由Consumer1消费，但是由于Consumer1消费过慢导致超时，次数Broker会将消息发送给Consumer2去消费，这样就产生了重复消费问题。因此，使用MQ时应该对一些关键消息进行幂等去重的处理。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/603864
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号