程序人生网站首页 程序人生

尚硅谷RabbitMQ视频-027-进阶篇-可靠性-消端确-测试

发布时间：2026-11-05 22:14:53编辑：雪饮阅读（）

对症下药

故障情况 1：消息没有发送到消息队列
- 解决思路 A：在生产者端进行确认，具体操作中我们会分别针对交换机和队列来确认，如果没有成功发送到消息队列服务器上，那就可以尝试重新发送
- 解决思路 B：为目标交换机指定备份交换机，当目标交换机投递失败时，把消息投递至备份交换机
故障情况 2：消息队列服务器宕机导致内存中消息丢失
- 解决思路：消息持久化到硬盘上，哪怕服务器重启也不会导致消息丢失
故障情况 3：消费端宕机或抛异常导致消息没有成功被消费
- 消费端消费消息成功，给服务器返回 ACK 信息，然后消息队列删除该消息
- 消费端消费消息失败，给服务器端返回 NACK 信息，同时把消息恢复为待消费的状态，这样就可以再次取回消息，重试一次（当然，这就需要消费端接口支持幂等性）

deliveryTag：交付标签机制

deliveryTag 是一个 64 位整数

消息往消费端投递时，会携带交付标签

消费端把消息处理结果 ACK、NACK、Reject 等返回给 Broker 之后，Broker 需要对对应的消息执行后续操作，例如删除消息、重新排队或标记为死信等等。那么 Broker 就必须知道它现在要操作的消息具体是哪一条。而 deliveryTag 作为消息的唯一标识就很好的满足了这个需求。

提问：如果交换机是 Fanout 模式，同一个消息广播到了不同队列，deliveryTag 会重复吗？

不会，deliveryTag 在 Broker 范围内唯一。

消息复制到各个队列但 deliveryTag 各不相同。

思考：更新购物车的微服务消费了消息返回 ACK 确认信息，然后 Broker 删除了消息，进而导致更新库存、更新积分的功能拿不到消息 —— 这种情况会发生吗？

接下来消费端代码

src\main\resources\application.yml:

spring:
rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: 123456
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual # 把消息确认模式改为手动确认,因为默认都是按成功处理，默认都是ack
    publisher-confirm-type: correlated #交换机的确认
    publisher-returns: true #队列的确认
logging:
level:
    org.example.springBoot.listener.MyMessageListener: info
    org.example.confirmProducer.listener.MyMessageListener: info
    org.example.alternate.listener.MyMessageListener: info

其中

listener:

  simple:

    acknowledge-mode: manual # 把消息确认模式改为手动确认,因为默认都是按成功处理，默认都是ack

这部分是新增

application.yml 里的 logging.level 是干嘛的

这只是日志级别配置，控制哪些包的 logger 输出什么级别的日志

为什么没配置它日志还是打印出来了

因为你没给配置级别时，它继承 root logger 的默认级别，Spring Boot 默认 root 级别就是`INFO`

然后是RabbitMQConsumerMainType.java：

package org.example.confirmConsumer;



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);

    }

}

然后是MyMessageListener.java：

package org.example.confirmConsumer.listener;

import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

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 QUEUE_NAME = "queue.order";



    @RabbitListener(queues = {QUEUE_NAME})

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

        long deliveryTag=message.getMessageProperties().getDeliveryTag();

        try {

            // 核心操作 处理消息

            log.info("org.example.confirmConsumer.listener 消费端 消息内容: " + dataString);





            /*

            核心操作成功: 返回 ACK 信息，

            multiple参数为：true为批量操作，false为非批量操作,

            basicAck作用：**成功确认**，Broker 收到 ack 之后，直接删除这条消息

            */

            channel.basicAck(deliveryTag, false);

        } catch (Exception e) {





            /*

            获取当前消息是否是重复投递的

            redelivered 为 true：说明当前消息已经重复投递过一次了

            redelivered 为 false：说明当前消息是第一次投递

            */

            Boolean redelivered = message.getMessageProperties().getRedelivered();



            /*

            核心操作失败: 返回 NACK 信息

            requeue 参数: 控制消息是否重新放回队列

            取值为 true: 重新放回队列, broker 会重新投递这个消息

            取值为 false: 不重新放回队列, broker 会丢弃这个消息

            basicNack作用：**失败拒绝**

            */



            //已是重投消息，则拒绝重投

            if(redelivered){

                channel.basicNack(deliveryTag, false, false);

            }

            //非重投消息，则重新投递

            if(!redelivered){

                channel.basicNack(deliveryTag, false, true);

            }

            throw new RuntimeException(e);

        }

    }

}

那么测试的时候就是直接运行

RabbitMQConsumerMainType.java的main方法

然后可以在rabbitmq管理后台中的exchange中找到和我们监听队列所绑定的交换机例如这里是exchange.direct.order里面通过routing key进行发消息到我们的监听队列中

之前我们建立的时候的routing key是order

在Publish message这里可以可以发送消息

当然你可以在deliveryTag变量前面开始打一个debug，然后一步步调试查看运行情况

如果是debug，那么需要RabbitMQConsumerMainType在调试模式下运行

那么由于debug的原因，queue.order中收到消息就卡在变量deliveryTag处了，所以queue.order队列中的unacked就可以看到有多一个的。

然后你也可以在org.example.confirmConsumer.listener. MyMessageListener的

try块中营造失败，比如10/0这种触发异常的操作，然后在catch

里仅仅让其重投，则在rabbitmq中的对应队列中例如queue.order中就能看到redelivered图示对应的在左侧折线图上面的展示就是重投的折线数据。

代表重复投递

那`multiple` 参数是干嘛的？

它是 RabbitMQ 协议层面（AMQP）的能力。`deliveryTag` 是一个通道内递增的序号（1, 2, 3...），`multiple=true` 表示“把 ≤ 当前 deliveryTag 的所有未确认消息一次性全部确认”。

## 什么场景会用 multiple=true？

虽然 Spring 监听器默认逐条回调，但以下场景确实会涉及“批量”：

- 手动批量确认：你在一个方法里循环处理了多条消息（或自己做了预取聚合），可以最后一次`basicAck(lastTag, true)` 一次性确认，减少网络往返，提升吞吐量。

- Prefetch 批量消费：配置了 prefetch 后，Broker 会预推送多条到客户端缓冲，批量 ack 可以一次性清掉多条。

- 批量监听容器：如果配置了`SimpleRabbitListenerContainerFactory` 的`consumerBatchEnabled=true` ，方法签名会变成`List<Message>` ，一次收到一批，这时逐条 ack 不如批量 ack 高效。

- 负向场景：`basicNack(tag, true, false)` 可以把当前及之前所有未确认的消息一起拒绝并丢弃。

关键字词：

上一篇：尚硅谷RabbitMQ视频-024-进阶篇-可靠性-持久化

下一篇：返回列表

无相关信息

您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-027-进阶篇-可靠性-消端确-测试

相关文章

您当前的位置：首页 > 慢生活 > 程序人生网站首页程序人生