RocketMq消费失败重发处理

迈不过友情╰ 2022-10-05 01:54 232阅读 0赞

### 文章目录 ###

*  前言
 *   *   *  消息重试分为两种：Producer发送消息的重试 和 Consumer消息消费的重试。
 *  一、Producer端重试
 *  二、 Consumer端重试

--------------------

# 前言 #

提示：不知道消费者消费失败的重新消费原理的出门左拐：[https://blog.csdn.net/weixin\_38308374/article/details/111415738][https_blog.csdn.net_weixin_38308374_article_details_111415738]

本文示例的需求：消费者接收消息，如果异常则重新信消费，当重试多次之后不再重试  
  
官方文档：https://github.com/apache/rocketmq/blob/master/docs/cn/features.md

### 消息重试分为两种：Producer发送消息的重试 和 Consumer消息消费的重试。 ###

# 一、Producer端重试 #

Producer端重试是指: Producer往MQ上发消息没有发送成功，比如网络原因导致生产者发送消息到MQ失败。

/**
         * 说明 抽取部分代码
         */
        private SendResult sendDefaultImpl(Message msg, final CommunicationMode communicationMode, final SendCallback sendCallback, final long timeout) { 
            
            //1、获取当前时间
            long beginTimestampFirst = System.currentTimeMillis();
            long beginTimestampPrev ;
            //2、去服务器看下有没有主题消息
            TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
            if (topicPublishInfo != null && topicPublishInfo.ok()) { 
                boolean callTimeout = false;
                //3、通过这里可以很明显看出 如果不是同步发送消息 那么消息重试只有1次
                int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
                //4、根据设置的重试次数，循环再去获取服务器主题消息
                for (times = 0; times < timesTotal; times++) { 
                    MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
                    beginTimestampPrev = System.currentTimeMillis();
                    long costTime = beginTimestampPrev - beginTimestampFirst;
                    //5、前后时间对比 如果前后时间差 大于 设置的等待时间 那么直接跳出for循环了 这就说明连接超时是不进行多次连接重试的
                    if (timeout < costTime) { 
                        callTimeout = true;
                        break;
    
                    }
                    //6、如果超时直接报错
                    if (callTimeout) { 
                        throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
                    }
            }
        }

通过这段源码很明显可以看出以下几点

1.如果是异步发送 那么重试次数只有1次。
    2.对于同步而言，超时异常也是不会再去重试。
    3.如果发生重试是在一个for 循环里去重试，所以它是立即重试而不是隔一段时间去重试。

实践出真知！！！

# 二、 Consumer端重试 #

消息重试等级为：默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”

可以使用setDelayLevelWhenNextConsume来设置

@Slf4j
    @Component
    public class Consumer { 
        /**
         * 消费者实体对象
         */
        private DefaultMQPushConsumer consumer;
        /**
         * 消费者组
         */
        public static final String CONSUMER_GROUP = "test_consumer";
        /**
         * 通过构造函数 实例化对象
         */
        public Consumer() throws MQClientException { 
            consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
            consumer.setNamesrvAddr("47.99.203.55:9876;47.99.203.55:9877");
            //订阅topic和 tags（ * 代表所有标签)下信息
            consumer.subscribe("topic_family", "*");
            //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
                //1、获取消息
                Message msg = msgs.get(0);
                try { 
                    //2、消费者获取消息
                    String body = new String(msg.getBody(), "utf-8");
                    //3、获取重试次数
                    int count = ((MessageExt) msg).getReconsumeTimes();
                    log.info("当前消费重试次数为 = {}", count);
                    //4、这里设置重试大于3次 那么通过保存数据库 人工来兜底
                    if (count >= 2) { 
                        log.info("该消息已经重试3次,保存数据库。topic={},keys={},msg={}", msg.getTopic(), msg.getKeys(), body);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //直接抛出异常
                    throw new Exception("=======这里出错了============");
                    //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) { 
                    e.printStackTrace();
                    context.setDelayLevelWhenNextConsume(4);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            //启动监听
            consumer.start();
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70]

**参考文章：https://blog.csdn.net/belongtocode/article/details/104310781**

[https_blog.csdn.net_weixin_38308374_article_details_111415738]: https://blog.csdn.net/weixin_38308374/article/details/111415738
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70]: /images/20221005/7f4a7248cf4a4e53b08687421e8a45c7.png