⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处民工哥技术之路「民工哥技术之路」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前面介绍了 RabbitMQ 基础配置、管理界面介绍、常用 API、消息收发与交换机、消息确认机制等相关的知识点，今天我将详细的为大家介绍 RabbitMQ 重试机制相关知识，希望大家能够从中收获多多！如有帮助，请点在看、转发支持一波！！！

RabbitMQ重试机制的简介

RabbitMQ 不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息连接是否已经断开，这个设置的原因是 RabbitMQ 允许消费者消费一条消息的时间可以很久很久。

RabbitMQ 的 Web 管理平台上可以看到当前队列中的 “Ready” 状态和 “Unacknowledged” 状态的消息数，分别对应等待投递给消费者的消息数和已经投递给消费者但是未收到确认信号的消息数。如下图：

注意事项

如果在处理消息的过程中，消费者的服务器在处理消息的时候出现异常，那么可能这条正在处理的消息就没有完成消息消费，数据就会丢失。为了确保数据不会丢失，RabbitMQ 支持消息确认-ACK。
如果忘记了消息确认，那么后果很严重。当 Consumer 退出时候，Message 会一直重新分发。然后 RabbitMQ 会占用越来越多的内容，由于 RabbitMQ 会长时间运行，因此这个”内存泄漏”是致命的。

RabbitMQ的自动确认

自动确认分四种情况（第一就是正常消费，其他三种为异常情况）

消息成功被消费，没有抛出异常，则自动确认，回复ack。不涉及requeue，毕竟已经成功了。requeue是对被拒绝的消息生效。
当抛出ImmediateAcknowledgeAmqpException异常的时候，则视为成功消费，确认该消息。
当抛出AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue = false（该异常会在重试超过限制后抛出）
抛出其他的异常，消息会被拒绝，且requeue = true

我遇到的是第四种情况，导致mq消息阻塞，并且消费者一直在消费同一条消息，然后抛异常，此时就进入了死循环。消息未被确认时如下图所示：

重试机制有2种情况

消息是自动确认时，如果抛出了异常导致多次重试都失败，消息被自动确认，消息就丢失了
消息是手动确认时，如果抛出了异常导致多次重试都失败，消息没被确认，也无法nack，就一直是unacked状态，导致消息积压。

核心配置

spring:
  # 项目名称
  application:
    name: rabbitmq-consumer
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # 重试机制
        retry:
          enabled: true #是否开启消费者重试
          max-attempts: 5 #最大重试次数
          initial-interval: 5000ms #重试间隔时间（单位毫秒）
          max-interval: 1200000ms #重试最大时间间隔（单位毫秒）
          multiplier: 2 #间隔时间乘子，间隔时间*乘子=下一次的间隔时间，最大不能超过设置的最大间隔时间

RabbitMQ重试机制的实现

下面将通过示例来讲解 RabbitMQ 重试机制的实现。首先需要创建两个 SpringBoot 项目并整合 RabbitMQ 客户端。

实现消息发送端

创建第一个 SpringBoot 项目（ rabbitmq-provider 消息发送项目）

在pom.xml配置信息文件中，添加相关依赖文件：

<!-- AMQP客户端 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.4.1</version>
</dependency>

在 application.yml 配置文件中配置 RabbitMQ 服务：

spring:
  # 项目名称
  application:
    name: rabbitmq-provider
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

配置队列

在 rabbitmq-provider（消息发送项目）中，配置队列名称，并将队列交由 IoC 管理，代码如下：

package com.pjb.config;
 
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * RabbitMQ配置类
 * @author pan_junbiao
 **/
@Configuration
public class RabbitMqConfig
{
    public static final String QUEUE_NAME = "queue_name"; //队列名称
    public static final String EXCHANGE_NAME = "exchange_name"; //交换器名称
    public static final String ROUTING_KEY = "routing_key"; //路由键
 
    /**
     * 队列
     */
    @Bean
    public Queue queue()
    {
        /**
         * 创建队列，参数说明：
         * String name：队列名称。
         * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
         * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。
         * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。
         * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
         * 当没有生产者或者消费者使用此队列，该队列会自动删除。
         * Map<String, Object> arguments：设置队列的其他一些参数。
         */
        return new Queue(QUEUE_NAME, true, false, false, null);
    }
 
    /**
     * Direct交换器
     */
    @Bean
    public DirectExchange exchange()
    {
        /**
         * 创建交换器，参数说明：
         * String name：交换器名称
         * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
         * 持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
         * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
         */
        return new DirectExchange(EXCHANGE_NAME, true, false);
    }
 
    /**
     * 绑定
     */
    @Bean
    Binding binding(DirectExchange exchange, Queue queue)
    {
        //将队列和交换机绑定, 并设置用于匹配键：routingKey
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

创建发送者

在 rabbitmq-provider（消息发送项目）中，创建发送者，利用 rabbitTemplate.convertAndSend() 方法发送消息，代码如下：

package com.pjb;
 
import com.pjb.config.RabbitMqConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
 
/**
 * RabbitMq测试类
 * @author pan_junbiao
 **/
@SpringBootTest
public class RabbitMqTest
{
    @Autowired
    RabbitTemplate rabbitTemplate;
 
    @Test
    public void sendMessage()
    {
        String message = "您好，欢迎访问 pan_junbiao的博客";
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_NAME, RabbitMqConfig.ROUTING_KEY, message);
        System.out.println("消息发送成功！");
    }
}

实现消息接收端

创建第二个 SpringBoot 项目（ rabbitmq-consumer 消息接收项目）

在pom.xml配置信息文件中，添加相关依赖文件：

<!-- AMQP客户端 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.4.1</version>
</dependency>

在 application.yml 配置文件中配置 RabbitMQ 服务，这里需要配置 RabbitMQ 重试机制：

spring:
  # 项目名称
  application:
    name: rabbitmq-consumer
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # 重试机制
        retry:
          enabled: true #是否开启消费者重试
          max-attempts: 5 #最大重试次数
          initial-interval: 5000ms #重试间隔时间（单位毫秒）
          max-interval: 1200000ms #重试最大时间间隔（单位毫秒）
          multiplier: 2 #间隔时间乘子，间隔时间*乘子=下一次的间隔时间，最大不能超过设置的最大间隔时间

创建接收者

在 rabbitmq-consumer（消息接收项目）中，创建创建接收者，注意，发送者和接收者的 Queue 名称必须一致，否则不能接收消息。

接收者接收到消息后，打印输出消息，然后程序抛出运行时异常，观察现象。代码如下：

package com.pjb.receiver;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.text.SimpleDateFormat;
import java.util.Date;
 
/**
 * 接收者
 * @author pan_junbiao
 **/
@Component
@RabbitListener(queues="queue_name")
public class Receiver
{
    @RabbitHandler
    public void process(String message)
    {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("接收消息: " + message + " 接收时间：" + sdf.format(new Date()));
        throw new RuntimeException();
    }
}

特别注意：

如果在消息接收端的 application.yml 配置文件中没有添加 RabbitMQ 重试机制的相关配置，当接收端收到消息后程序抛出异常，那么发送端将得不到消息确认（ACK），此时发送端将会循环的发送消息，最终导致内存溢出。

执行结果

从上述执行结果来看，当接收端重试5次后，将消息确认（ACK)。

芋道源码 —— 纯源码解析博客

硬卷消息中间件系列（八）：RabbitMQ 重试机制详解