RabbitMQ——消息确认

news2025/5/23 1:23:43

一、消息确认机制

生产者发送的消息,可能有以下两种情况:

1> 消息消费成功

2> 消息消费失败

为了保证消息可靠的到达消费者(!!!注意:消息确认机制和前面的工作模式中的publisher confirms模式有很大区别,消息确认保证的是消息可靠的到达消费者,而publisher confirms保证的是消息可靠的到达RabbitMQServer),RabbitMQ引入了消息确认机制:

消费者在消费消息时,可以指定autoAck参数,对应着两种确认方式

(1)自动确认:消息只要到达消费者就会自动确认,不会考虑消费者是否正确消费了这些消息,直接从 内存/磁盘 中删除消息;

(2)手动确认:消息到达消费者,不会自动确认,会等待消费者调用Basic.Ack命令,才会从内存/磁盘 移除这条消息。

DefaultConsumer consumer = new DefaultConsumer(channel) {
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, 
AMQP.BasicProperties properties, byte[] body) throws IOException {
 System.out.println("接收到消息: " + new String(body));
 }
};
channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);

如果将basicConsume中的true改为false,对于RabbitMQ服务器来说,消息分成两个部分:

1>  未发送给消费者的消息;

2>  已经发送给消费者,但还没有被确认的消息。

对应管理界面:


二、手动确认方法

RabbitMQ提供了两种手动确认,分别是肯定确认和否定确认,对应着3个方法:

(1)肯定确认:Channel.basicAck(long deliveryTag, boolean multiple)

    其中,deliveryTag为消息的唯一表示,在每一个channel中都是唯一的,相当于TCP协议中的序号的作用,用来确认哪条消息已经收到,multiple为true表示是否批量确认,同样与TCP中确认序号的作用类似,表示在这个deliveryTag前的消息都已收到,为false则表示一次只确认一条消息。

(2)否定确认: Channel.basicReject(long deliveryTag, boolean requeue)

   如果消息到达消费者后,消费者未正确处理这条消息(如发生异常),就可以通过这个方法进行否定确认,其中,参数requeue表示这条消息是否需要重新入队,这个方法一次只能确认一条消息。

(3)否定确认:Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)

   与basicAck一样,这个方法可以通过multiple来实现批量确认


三、使用Spring Boot演示消息确认机制

  演示之前,需要先了解Spring-AMQP的确认机制,它和RabbitMQ JDK Client库的确认机制有些许不同:

1. AcknowledgeMode.NONE

  和RabbitMQ JDK Client库的自动确认机制一样,只要消息到达消费者,这条消息就会从队列中移除。

2. AcknowledgeMode.AUTO(和JDK Client库的区别)

  消息到达消费者且处理成功,才会自动确认,如果消息在处理过程中发生了异常,则不会自动确认。

3. AcknowledgeMode.MANUAL

  和JDK Client库一样,属于手动确认机制,同样分为肯定确认和否定确认。

接下来,进行准备工作,创建一个Spring Boot项目,添加RabbitMQ依赖:

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

添加RabbitMQ相关配置:

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.17.130:5672/extension
    listener:
      simple:
        acknowledge-mode: none #表示当前确认机制未AcknowledgeMode.NONE

添加下图的包,创建对应的类:

 


3.1  AcknowledgeMode.NONE

(1)编写常量类(由于只是验证消息确认机制,可以随便使用一种工作模式)

public class Constants {
    public static final String ACK_QUEUE = "ack.queue";
    public static final String ACK_EXCHANGE = "ack.exchange";
}

(2)配置声明队列、交换机、交换机与队列绑定关系

@Configuration
public class RabbitMQConfig {
    //1.声明队列
    @Bean("ackQueue")
    public Queue ackQueue(){
        return QueueBuilder.durable(Constants.ACK_QUEUE).build();
    }
    //2.声明交换机
    @Bean("ackExchange")
    public FanoutExchange ackExchange(){
        return ExchangeBuilder.fanoutExchange(Constants.ACK_EXCHANGE).build();
    }
    //3.声明队列与交换机的绑定关系
    @Bean
    public Binding ackBinding(@Qualifier("ackExchange") FanoutExchange fanoutExchange,@Qualifier("ackQueue") Queue queue){
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }
}

(3)编写生产者代码

@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/ack")
    public String ack(){
        rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,"","consumer ack mode test...");
        return "发送消息成功";
    }
}

(4)编写消费者代码

@Component
public class AckListener {
    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

            System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());
            //业务逻辑处理
            System.out.println("业务逻辑处理");
//            int num = 10/0;
            System.out.println("业务处理完成");
    }
}

(5)测试正常消费消息的情况


(6)测试消息消费异常情况(将消费者代码中得 int num  = 10/0  注解打开)


(7)总结

  经过确认,可以发现AcknowledgeMode.NONE机制,无论消费者是否正确消费消息,都会自动确认,不会保留异常消费的消息


3.2 AcknowledgeMode.AUTO

(1) 修改配置文件中的 acknowledge-mode: none 为 acknowledge-mode: auto

(2)演示消息正常消费的情况(将 int num = 10/0 注释)

  @RabbitListener(queues = Constants.ACK_QUEUE)
    public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

            System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());
            //业务逻辑处理
            System.out.println("业务逻辑处理");
//            int num = 10/0;
            System.out.println("业务处理完成");
    }

运行程序,访问 producer/ack 接口发送消息:


(3)演示消息处理异常情况(取消 int num = 10/0 的注释)

 @RabbitListener(queues = Constants.ACK_QUEUE)
    public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

            System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());
            //业务逻辑处理
            System.out.println("业务逻辑处理");
            int num = 10/0;
            System.out.println("业务处理完成");
    }

运行程序,访问生产者端口:

消息没有成功消费,如果此时我们修改代码(将 int num = 10/0 注释,使程序不再发生异常),再次运行程序,队列中保存的消息就会被正常消费:


3.3 AcknowledgeMode.MANUAL

  (1) 修改配置文件中 acknowledge-mode: auto 为 acknowledge-mode: manul

(2)修改消费者代码(由于是手动确认,需要在代码中添加正常消费时的 “肯定确认” 和 消费异常时的 “否定确认”)

    @RabbitListener(queues = Constants.ACK_QUEUE)
    public void ackListener(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try{
                System.out.printf("接收到消息:%s,deliveryTag: %d \n",new String(message.getBody(),"UTF-8"),message.getMessageProperties().getDeliveryTag());
                //业务逻辑处理
                System.out.println("业务逻辑处理");
//                int num = 10/0;
                System.out.println("业务处理完成");
                
                //消息正常消费,肯定确认
                channel.basicAck(deliveryTag,false);//单条确认
                
            }catch (Exception e){
                //消息消费异常,否定确认
                channel.basicNack(deliveryTag,false,true);//单条否定,异常后重新入队
            }
    }

(3)消息正常消费时的情况(注释 int num = 10/0)

接下来注释掉手动肯定确认的这行代码,看看会发生什么情况:

 //消息正常消费,肯定确认
 //channel.basicAck(deliveryTag,false);//单条确认

再次运行程序并通过接口 producer/ack 发送消息:

可以看到,在AcknowledgeMode.MANUAL机制下,如果不手动确认,队列不会移除已经被正常消费的消息:


(3)消息消费异常的情况下(取消 int num = 10/0 的注释)

接下来注释掉channel.basicNack,运行程序,访问接口发送消息:

如果将参数requeue置为false,会怎么样?

channel.basicNack(deliveryTag,false,false);//单条否定,重新发送消息

运行程序:


(4)总结

1> 无论是否正常处理消息都要进行手动确认;

2> 正常处理消息但未手动确认,管理界面中的队列会有 一条/多条 Unacked 的消息(重新启动程序后会重新消费);

3> 异常处理消息且未手动确认,也会有 一条/多条 Unacked 的消息(重启程序同样重新消费);

4> 如果异常处理,将requeue置为false,队列不会保存 这条/多条 异常消费的消息

                                               置为true,队列会不断重新发送 这条/多条 消息

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2383515.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

测试W5500的第2步_使用ioLibrary库创建TCP客户端

ioLibrary库下载地址&#xff1a;文件下载地址:https://gitee.com/wiznet-hk/STM32F10x_W5500_Examples 源文件下载地址:https://gitee.com/wiznet-hk 没有注册的&#xff0c;只能复制粘贴了。 本文介绍了如何初始化STM32的硬件资源&#xff0c;配置W5500的网络参数&#xff…

深度学习之用CelebA_Spoof数据集搭建一个活体检测-训练好的模型用MNN来推理

一、模型转换准备 首先确保已完成PyTorch到ONNX的转换&#xff1a;深度学习之用CelebA_Spoof数据集搭建活体检测系统&#xff1a;模型验证与测试。这里有将PyTorch到ONNX格式的模型转换。 二、ONNX转MNN 使用MNN转换工具进行格式转换&#xff1a;具体的编译过程可以参考MNN的…

开源安全大模型Foundation-Sec-8B实操

一、兴奋时刻 此时此刻,晚上22点55分,从今天早上6点左右开始折腾,花费了接近10刀的环境使用费,1天的休息时间,总算是把Foundation-Sec-8B模型跑起来了,中间有两次胜利就在眼前,但却总在远程端口转发环节出问题,让人难受。直到晚上远程Jupyter访问成功那一刻,眉开眼笑,…

【JavaWeb】MySQL

1 引言 1.1 为什么学&#xff1f; 在学习SpringBootWeb基础知识(IOC、DI等)时&#xff0c;在web开发中&#xff0c;为了应用程序职责单一&#xff0c;方便维护&#xff0c;一般将web应用程序分为三层&#xff0c;即&#xff1a;Controller、Service、Dao 。 之前的案例中&am…

微信小游戏流量主广告自动化浏览功能案例5

功能需求&#xff1a; 支持APP单行文本框输入1个小程序链接&#xff0c;在“文件传输助手”界面发送小程序链接并进入。 主要有“文章列表首页”和“文章内容”页面。每个页面支持点击弹窗广告、槽位广告、视频广告入口、视频广告内第三方广告。 弹窗广告、槽位广告、视频广…

软件的技术架构、应用架构、业务架构、数据架构、部署架构

一、各架构定义 1. 技术架构&#xff08;Technical Architecture&#xff09; 定义&#xff1a;技术架构关注的是支撑系统运行的底层技术基础设施和软件平台&#xff0c;包括硬件、操作系统、中间件、编程语言、框架、数据库管理系统等技术组件的选择和组合方式。它描述了系统…

CSS 文字样式全解析:从基础排版到视觉层次设计

CSS 文字样式目录 一、字体家族&#xff08;font-family&#xff09; 二、字体大小&#xff08;font-size&#xff09; 三、字体粗细&#xff08;font-weight&#xff09; 四、字体样式&#xff08;font-style&#xff09; 五、文本转换&#xff08;text-transform&#xf…

【高德开放平台-注册安全分析报告】

前言 由于网站注册入口容易被黑客攻击&#xff0c;存在如下安全问题&#xff1a; 暴力破解密码&#xff0c;造成用户信息泄露短信盗刷的安全问题&#xff0c;影响业务及导致用户投诉带来经济损失&#xff0c;尤其是后付费客户&#xff0c;风险巨大&#xff0c;造成亏损无底洞…

RabbitMQ的简介

三个概念 生产者&#xff1a;生产消息的服务消息代理&#xff1a;消息中间件&#xff0c;如RabbitMQ消费者&#xff1a;获取使用消息的服务 消息队列到达消费者的两种形式 队列&#xff08;queue&#xff09;:点对点消息通信&#xff08;point-to-point&#xff09; 消息进入队…

使用Gemini, LangChain, Gradio打造一个书籍推荐系统 (第一部分)

第一部分&#xff1a;数据处理 import kagglehub# Download latest version path kagglehub.dataset_download("dylanjcastillo/7k-books-with-metadata")print("Path to dataset files:", path)自动下载该数据集的 最新版本 并返回本地保存的路径 impo…

大语言模型 16 - Manus 超强智能体 Prompt分析 原理分析 包含工具列表分析

写在前面 Manus 是由中国初创公司 Monica.im 于 2025 年 3 月推出的全球首款通用型 AI 智能体&#xff08;AI Agent&#xff09;&#xff0c;旨在实现“知行合一”&#xff0c;即不仅具备强大的语言理解和推理能力&#xff0c;还能自主执行复杂任务&#xff0c;直接交付完整成…

以太联Intellinet带您深度解析PoE交换机的上行链路端口(Uplink Ports)

在当今网络技术日新月异的时代&#xff0c;以太网供电(PoE)交换机已然成为现代网络连接解决方案中不可或缺的“利器”。它不仅能够出色地完成数据传输任务&#xff0c;还能为所连接的设备提供电力支持&#xff0c;彻底摆脱了单独电源适配器的束缚&#xff0c;让网络部署更加简洁…

分类算法 Kmeans、KNN、Meanshift 实战

任务 1、采用 Kmeans 算法实现 2D 数据自动聚类&#xff0c;预测 V180,V260 数据类别&#xff1b; 2、计算预测准确率&#xff0c;完成结果矫正 3、采用 KNN、Meanshift 算法&#xff0c;重复步骤 1-2 代码工具&#xff1a;jupyter notebook 视频资料 无监督学习&#xff…

网络安全之身份验证绕过漏洞

漏洞简介 CrushFTP 是一款由 CrushFTP LLC 开发的强大文件传输服务器软件&#xff0c;支持FTP、SFTP、HTTP、WebDAV等多种协议&#xff0c;为企业和个人用户提供安全文件传输服务。近期&#xff0c;一个被编号为CVE-2025-2825的严重安全漏洞被发现&#xff0c;该漏洞影响版本1…

MySQL 主从复制搭建全流程:基于 Docker 与 Harbor 仓库

一、引言 在数据库管理中&#xff0c;MySQL 主从复制是一种非常重要的技术&#xff0c;它可以实现数据的备份、读写分离&#xff0c;减轻主数据库的压力。本文将详细介绍如何使用 Docker 和 Harbor 仓库来搭建 MySQL 主从复制环境&#xff0c;适合刚接触数据库和 Docker 的新手…

Django框架的前端部分使用Ajax请求一

Ajax请求 目录 1.ajax请求使用 2.增加任务列表功能(只有查看和新增) 3.代码展示集合 这篇文章, 要开始讲关于ajax请求的内容了。这个和以前文章中写道的Vue框架里面的axios请求, 很相似。后端代码, 会有一些细节点, 跟前几节文章写的有些区别。 一、ajax请求使用 我们先…

cmd如何从C盘默认路径切换到D盘某指定目录

以从C盘cmd打开后的默认目录切换到目录"D:\Program Files\MySQL\MySQL Server 8.0\bin\mysqld"为例 打开cmd 首先点击开始键&#xff0c;搜索cms&#xff0c;右键以管理员身份运行打开管理员端的命令行提示符 1、首先要先切换到D盘 直接输入D:然后回车就可以&…

每日Prompt:实物与手绘涂鸦创意广告

提示词 一则简约且富有创意的广告&#xff0c;设置在纯白背景上。 一个真实的 [真实物体] 与手绘黑色墨水涂鸦相结合&#xff0c;线条松散而俏皮。涂鸦描绘了&#xff1a;[涂鸦概念及交互&#xff1a;以巧妙、富有想象力的方式与物体互动]。在顶部或中部加入粗体黑色 [广告文案…

学习笔记:黑马程序员JavaWeb开发教程(2025.4.8)

12.11 登录校验-Filter-详解&#xff08;过滤器链&#xff09; 过滤器链及其执行顺序&#xff0c;一个Filter一个过滤器链&#xff0c;类名排名越靠前&#xff08;按照ABC这样的顺序&#xff09;&#xff0c;就先执行谁 12.12 登录校验-Filter-登录校验过滤器 获取请求参数&…

Ubuntu部署私有Gitlab

这个东西安装其实挺简单的&#xff0c;但是因为我这边迁移了数据目录和使用自己安装的 nginx 代理还是踩了几个坑&#xff0c;所以大家可以注意下 先看下安装 # 先安装必要组件 sudo apt update sudo apt install -y curl openssh-server ca-certificates tzdata perl# 添加gi…