[toc]

单词	词性	音标	含义
ack	n.	/æk/	确认（消息处理成功）
nack	n.	/næk/	未确认（消息处理失败）
reject	v.	/rɪˈdʒekt/	拒绝（处理失败且不放回队列）
deliveryTag	n.	/dɪˈlɪvəri tæɡ/	交付标签（broker 给每条消息的唯一标识）
requeue	v.	/ˌriːˈkjuː/	重新入队（失败后放回队列再投递）
quorum	n.	/ˈkwɔːrəm/	仲裁（Quorum 队列，支持原生延迟重试）
retry	v./n.	/ˈriːtraɪ/	重试（失败后再次投递处理）

Lecture05-RabbitMQ-消息可靠性-消费者端

1、ACK

ACK 是 acknowledge 的缩写，表示已确认

2、默认情况

默认情况下，消费端取回消息后，默认会自动返回 ACK 确认消息，所以在前面的测试中消息被消费端消费之后，RabbitMQ 得到 ACK 确认信息就会删除消息

但实际开发中，消费端根据消息队列投递的消息执行对应的业务，未必都能执行成功，如果希望能够多次重试，那么默认设定就不满足要求了

所以还是要修改成手动确认

希望的状态：

• 消费端处理消息成功之后：返回 ACK
• 消费端处理消息失败之后：返回 NACK 让 RabbitMQ 重新投递这个消息，消费端重新执行业务代码 根据具体失败的原因，重试并非全部都是无限次重试，也可以根据需要指定重试的次数

3、创建消费端module

1、配置POM

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.0.0</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

2、YAML

增加针对监听器的设置：

spring:
  rabbitmq:
    host: 192.168.100.100
    port: 5672
    username: admin
    password: admin123
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual # 把消息确认模式改为手动确认

3、主启动类

没有特殊设定：

package com.atguigu.mq;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class RabbitMQConsumerMainType {

    public static void main(String[] args) {
        SpringApplication.run(RabbitMQConsumerMainType.class, args);
    }

}

4、消费端监听器

1、创建监听器类

package com.atguigu.mq.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;

@Component
public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME  = "queue.order";

    public void processMessage(String dataString, Message message, Channel channel) {

    }

}

2、在接收消息的方法上应用注解

注意：如果 RabbitMQ 服务器上已经创建了交换机、队列等组件就只需要指定监听的队列本身

// 修饰监听方法
@RabbitListener(
        // 设置绑定关系
        bindings = @QueueBinding(

            // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
            value = @Queue(value = QUEUE_NAME, durable = "true", autoDelete = "false"),

            // 配置交换机信息：durable 设置为 true 表示交换机持久化；autoDelete 设置为 false 表示关闭自动删除
            exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),

            // 配置路由键信息
            key = {ROUTING_KEY}
))
public void processMessage(String dataString, Message message, Channel channel) {

}

3、接收消息方法内部逻辑

• 业务处理成功：手动返回ACK信息，表示消息成功消费
• 业务处理失败：手动返回NACK信息，表示消息消费失败。此时有两种后续操作供选择：
  • 把消息重新放回消息队列，RabbitMQ会重新投递这条消息，那么消费端将重新消费这条消息——从而让业务代码再执行一遍
  • 不把消息放回消息队列，返回reject信息表示拒绝，那么这条消息的处理就到此为止

4、相关API

先回到PPT理解“deliveryTag：交付标签机制”

下面我们探讨的三个方法都是来自于com.rabbitmq.client.Channel接口

①basicAck()方法

• 方法功能：给Broker返回ACK确认信息，表示消息已经在消费端成功消费，这样Broker就可以把消息删除了
• 参数列表：

参数名称	含义
long deliveryTag	Broker给每一条进入队列的消息都设定一个唯一标识
boolean multiple	取值为true：为小于、等于deliveryTag的消息批量返回ACK信息 取值为false：仅为指定的deliveryTag返回ACK信息

②basicNack()方法

• 方法功能：给Broker返回NACK信息，表示消息在消费端消费失败，此时Broker的后续操作取决于参数requeue的值
• 参数列表：

参数名称	含义
long deliveryTag	Broker给每一条进入队列的消息都设定一个唯一标识
boolean multiple	取值为true：为小于、等于deliveryTag的消息批量返回NACK信息 取值为false：仅为指定的deliveryTag返回NACK信息
boolean requeue	取值为true：Broker将消息重新放回队列，接下来会重新投递给消费端 取值为false：Broker将消息标记为已消费，不会放回队列

③basicReject()方法

• 方法功能：根据指定的deliveryTag，对该消息表示拒绝
• 参数列表：

参数名称	含义
long deliveryTag	Broker给每一条进入队列的消息都设定一个唯一标识
boolean requeue	取值为true：Broker将消息重新放回队列，接下来会重新投递给消费端 取值为false：Broker将消息标记为已消费，不会放回队列

• basicNack()和basicReject()有啥区别？
  • basicNack()有批量操作
  • basicReject()没有批量操作

5、完整代码示例

package com.atguigu.mq.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME  = "queue.order";

    // 修饰监听方法
    @RabbitListener(
            // 设置绑定关系
            bindings = @QueueBinding(

                // 配置队列信息：durable 设置为 true 表示队列持久化；autoDelete 设置为 false 表示关闭自动删除
                value = @Queue(value = QUEUE_NAME, durable = "true", autoDelete = "false"),

                // 配置交换机信息：durable 设置为 true 表示交换机持久化；autoDelete 设置为 false 表示关闭自动删除
                exchange = @Exchange(value = EXCHANGE_DIRECT, durable = "true", autoDelete = "false"),

                // 配置路由键信息
                key = {ROUTING_KEY}
    ))
    public void processMessage(String dataString, Message message, Channel channel) throws IOException {

        // 1、获取当前消息的 deliveryTag 值备用
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            // 2、正常业务操作
            log.info("消费端接收到消息内容：" + dataString);
            
            // System.out.println(10 / 0);

            // 3、给 RabbitMQ 服务器返回 ACK 确认信息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {

            // 4、获取信息，看当前消息是否曾经被投递过
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            if (!redelivered) {
                // 5、如果没有被投递过，那就重新放回队列，重新投递，再试一次
                channel.basicNack(deliveryTag, false, true);
            } else {
                // 6、如果已经被投递过，且这一次仍然进入了 catch 块，那么返回拒绝且不再放回队列
                channel.basicReject(deliveryTag, false);
            }

        }
    }

}

5、要点总结

• 要点1：把消息确认模式改为手动确认
• 要点2：调用Channel对象的方法返回信息
  • ACK：Acknowledgement，表示消息处理成功
  • NACK：Negative Acknowledgement，表示消息处理失败
  • Reject：拒绝，同样表示消息处理失败
• 要点3：后续操作
  • requeue为true：重新放回队列，重新投递，再次尝试
  • requeue为false：不放回队列，不重新投递
• 要点4：deliveryTag 消息的唯一标识，查找具体某一条消息的依据

6、流程梳理

7、多啰嗦一句

消费端如果设定消息重新放回队列，Broker重新投递消息，那么消费端就可以再次消费消息，这是一种“重试”机制，这需要消费端代码支持“幂等性”——这属于前置知识，不展开了。

8、消费失败后的延迟重试（Quorum 队列 4.3 原生）

前面我们讲的重试，是消息失败后 立即重新投递。但如果消费端一直处理失败，消息就会被反复重投，形成“快速重试循环”，不断压垮系统。

RabbitMQ 4.3 起，Quorum 队列提供了原生的延迟重试（delayed retry）：失败的消息先在队列里“缓一缓”，按退避间隔延迟后再投递——这是针对重试退避的能力，属于本节的可靠性话题。

1、前提：队列必须是 Quorum 队列

• 延迟重试是 Quorum 队列独有的特性（4.3 新增），经典队列没有。
• 如果你用的是经典队列（前面 04 的 @Queue 默认就是经典队列），失败后只能走 requeue=true 立即重投，没有延迟退避。
• 要用延迟重试，队列必须声明为 Quorum（UI 里 Type 选 Quorum，或用 Spring 的 QueueBuilder.quorum()）。

2、延迟时间怎么算（线性退避）

delay = min( min_delay × delivery-count, max_delay )

示例（min=1000ms、max=30000ms）：

第几次失败返回	delay	说明
第 1 次	1000ms	min_delay
第 2 次	2000ms	×2
第 3 次	3000ms	×3
第 4 次及以上	30000ms	封顶到 max_delay

注意：delivery-count 只在“真正失败”时递增（reject、channel 终止时未确认的消息）；仅用 nack 不递增，这类消息会一直按最小延迟重试。

3、如何开启（配置项）

在声明 Quorum 队列时设置三个参数（也可用策略批量配置）：

参数	含义
x-delayed-retry-type	延迟哪些消息：disabled（默认）/ all / failed / returned
x-delayed-retry-min	最小延迟（毫秒，启用时必填）
x-delayed-retry-max	最大延迟（毫秒，可选）

delayed-retry-type 取值：

值	延迟哪些消息
disabled	不延迟（默认）
all	所有被返回的消息都延迟
failed	只延迟 delivery-count 递增的（真正失败）
returned	只延迟 delivery-count 不递增的（如 nack）

4、Spring Boot 声明带延迟参数的 Quorum 队列

@Configuration
public class RabbitConfig {
    public static final String QUEUE = "delay.retry.queue";

    @Bean
    public Queue retryQueue() {
        return QueueBuilder.durable(QUEUE)
                .quorum()                                    // 必须是 Quorum 队列
                .withArgument("x-delayed-retry-type", "all") // 所有返回都延迟
                .withArgument("x-delayed-retry-min", 1000)   // 最小延迟 1s
                .withArgument("x-delayed-retry-max", 30000)  // 最大延迟 30s
                .build();
    }
}

5、消费者：失败时 Nack（不重入队）触发延迟重试

沿用前面的手动确认（acknowledge-mode: manual）。处理失败时 basicNack(requeue=false)，让消息真正返回队列，进入延迟重试。

@Component
public class RetryConsumer {
    @RabbitListener(queues = RabbitConfig.QUEUE)
    public void on(Message message, Channel channel) throws Exception {
        long tag = message.getMessageProperties().getDeliveryTag();
        try {
            // 处理业务
            channel.basicAck(tag, false);
        } catch (Exception e) {
            // 失败 → 消息返回队列 → 延迟重试（先 1s，再 2s……封顶 30s）
            channel.basicNack(tag, false, false);
        }
    }
}

6、验证

发一条消息，让消费者第一次处理抛异常 → 观察约 1s 后再次投递；再失败 → 约 2s 后再次投递……延迟递增。Management UI 队列详情能看到 delayed 状态的消息计数。跑通标志：失败消息按退避间隔逐步延迟重投，而不是立即无限重投。

7、和“延迟队列（定时调度）”的区别（重要）

本节的延迟重试是消费失败后的重试退避；如果你要的是**“推迟 N 秒后执行某个操作”**（定时调度，比如下单 30 分钟后检查是否支付），那不是重试退避，而是延迟队列——用 TTL + 死信队列 实现，见 Lecture06-延迟队列（TTL+死信）.md。

9、本节验证

• 会配置手动确认（acknowledge-mode: manual）✅
• 会用 @RabbitListener + @QueueBinding 声明绑定关系 ✅
• 会用 basicAck / basicNack / basicReject 及 deliveryTag 处理消息 ✅
• 理解 Quorum 队列的原生延迟重试并配置参数 ✅