当前位置：首页 > news >正文

【RabbitMQ实战】Springboot 整合RabbitMQ组件，多种编码示例，带你实践看完这一篇就够了

news 来源：原创 2024/7/5 4:16:36

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

前言
一、对RabbitMQ管理界面深入了解
- 1、在这个界面里面我们可以做些什么？
二、编码练习
- （1）使用direct exchange(直连型交换机)
- （2）使用Topic Exchange 主题交换机。
- （3）使用Fanout Exchang 扇型交换机。
三、消息确认种类
- A：消息发送确认
- B：消费接收确认
- - 方式一：通过配置类的方式实现
  - 方式二：通过yml配置来完成消费者确认

前言

该篇文章内容较多，包括有RabbitMQ一些理论介绍，provider消息推送实例，consumer消息消费实例，Direct、Topic、Fanout多种交换机的使用，同时简单介绍对消息回调、手动确认等。

这里面的每一种使用都包含实际编码示例，供大家理解，共同进步，如有不足。还请指教。

一、对RabbitMQ管理界面深入了解

装完rabbitMq，启动MQ后，本地浏览器输入http://ip:15672/ ，看到一个简单后台管理界面；
在这里插入图片描述
对于其中的一些具体指标的解释：

Ready：待消费的消息总数。
Unacked：待应答的消息总数。
Total：总数 Ready+Unacked。
Publish： producter pub消息的速率。
Publisher confirm： broker确认pub消息的速率。
Deliver(manual ack)： customer手动确认的速率。
Deliver( auto ack)： customer自动确认的速率。
Consumer ack： customer正在确认的速率。
Redelivered：正在传递’redelivered’标志集的消息的速率。
Get (manual ack)：响应basic.get而要求确认的消息的传输速率。
Get (auto ack)：响应于basic.get而发送不需要确认的消息的速率。
Return：将basic.return发送给producter的速率。
Disk read： queue从磁盘读取消息的速率。
Disk write： queue从磁盘写入消息的速率。

Connections：client的tcp连接的总数。
Channels：通道的总数。
Exchange：交换器的总数。
Queues：队列的总数。
Consumers：消费者的总数。

更详细的可见：
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/qq_19343089/article/details/135724659

1、在这个界面里面我们可以做些什么？

可以手动创建虚拟host，创建用户，分配权限，创建交换机，创建队列等等，还有查看队列消息，消费效率，推送效率等等。

以上这些管理界面的操作在这篇暂时不做扩展描述，我想着重介绍后面实例里会使用到的。

首先先介绍一个简单的一个消息推送到接收的流程，提供一个简单的图：
在这里插入图片描述
黄色的圈圈就是我们的消息推送服务，将消息推送到中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系（后面会详细讲）将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。

常用的交换机有以下三种，因为消费者是从队列获取信息的，队列是绑定交换机的（一般），所以对应的消息推送/接收模式也会有以下几种：

- Direct Exchange

直连型交换机，根据消息携带的路由键将消息投递给对应队列。

大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。

- Fanout Exchange

扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

- Topic Exchange

主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
简单地介绍下规则：

(星号) * 用来表示一个单词 (必须出现的)
(井号) # 用来表示任意数量（零个或多个）单词
通配的绑定键是跟队列进行绑定的，举个小例子
队列Q1 绑定键为 .TT. 队列Q2绑定键为 TT.#
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；

主题交换机是非常强大的，为啥这么膨胀？
当一个队列的绑定键为 “#”（井号）的时候，这个队列将会无视消息的路由键，接收所有的消息。
当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。

另外还有 Header Exchange 头交换机，Default Exchange 默认交换机，Dead Letter Exchange 死信交换机，这几个该篇暂不做讲述。

好了，一些简单的介绍到这里为止，接下来我们来一起编码。

二、编码练习

本次实例教程需要创建2个springboot项目，一个 rabbitmq-provider （生产者），一个rabbitmq-consumer（消费者）。【补充说明：我这里模块名称创建错了，其中生产者我创建成了rabbitmq-consumer，消费者我这里叫做 rabbitmq-consumer-true】

首先创建 rabbitmq-provider，

pom.xml里用到的jar依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.atguigu.gulimall</groupId><artifactId>rabbitmq-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>rabbitmq-consumer</name><description>RabbitMQ生产者模块</description><url/><licenses><license/></licenses><developers><developer/></developers><scm><connection/><developerConnection/><tag/><url/></scm><properties><java.version>1.8</java.version><!--        <spring-cloud.version>2021.0.4</spring-cloud.version>--><spring-cloud.version>2021.0.1</spring-cloud.version></properties><dependencies><dependency><groupId>com.atguigu.gulimall</groupId><artifactId>gulimall-common</artifactId><version>0.0.1-SNAPSHOT</version><exclusions><exclusion><artifactId>servlet-api</artifactId><groupId>javax.servlet</groupId></exclusion></exclusions></dependency><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.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

然后application.yml：

server:port: 8021#数据源配置
spring:datasource:username: rootpassword: rooturl: jdbc:mysql://192.168.56.10:3306/gulimall_umsdriver-class-name: com.mysql.cj.jdbc.Driver#注册到注册中心cloud:nacos:discovery:server-addr: 127.0.0.1:8848application:name: rabbitmq-consumer#配置rabbitMq 服务器rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guest#虚拟host 可以不设置,使用server默认hostvirtual-host: /
#    publisher-returns: true  #确认消息已发送到队列(Queue)  这个在生产者模块配置 这个后期再配置，这会还用不到
#    publisher-confirm-type: correlated   #确认消息已发送到交换机(Exchange) 这个在生产者模块配置 这个后期再配置，这会还用不到logging:level:com.atguigu.gulimall: debug   #调整product模块日志的输出模式是debug级别，这样就能在控制台看到dao包下的输出日志了。

一定要注意要注意要注意！！！！！
里面的virtual-host 是指RabbitMQ控制台中的下面的位置（我理解是指你的队列和交换机在哪个分组下面，可以为每一个项目创建单独的分组，但是在此我没有单独创建，直接放到了 / 下面）
在这里插入图片描述
那么怎么建一个单独的host呢？假如我就是想给某个项目接入，使用一个单独host，顺便使用一个单独的账号，就好像我文中配置的 root 这样。

其实也很简便：

virtual-host的创建：
在这里插入图片描述

账号user的创建：
在这里插入图片描述

然后记得给账号分配权限，指定使用某个virtual host：
指定给自己刚刚为某个项目单独创建的virtual host。
在这里插入图片描述

其实还可以特定指定交换机使用权等等：
在这里插入图片描述

（1）使用direct exchange(直连型交换机)

创建DirectRabbitConfig.java（对于队列和交换机持久化以及连接使用设置，在注释里有说明，后面的不同交换机的配置就不做同样说明了）：

package com.atguigu.gulimall.rabbitmqconsumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.core.Queue;/*** 这里使用的是direct exchange(直连型交换机),  也就是交换机和队列是一对一关系* 模拟 rabbitmq-provider （生产者）,这里模块名字写错了。这个是消息生产者** @author: jd* @create: 2024-06-24*/
@Configuration
public class DirectRabbitConfig {// 声明需要使用的交换机/路由Key/队列的名称public static final String DEFAULT_EXCHANGE = "TestDirectExchange";public static final String DEFAULT_ROUTE = "TestDirectRouting";public static final String DEFAULT_QUEUE = "TestDirectQueue";// 声明交换机，需要几个声明几个，这里就一个@Beanpublic DirectExchange directExchange(){return new DirectExchange(DEFAULT_EXCHANGE);}//创建队列//队列 起名：TestDirectQueue@Beanpublic Queue TestDirectQueue(){// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。//   return new Queue("TestDirectQueue",true,true,false);//一般设置一下队列的持久化就好,其余两个就是默认falsereturn new Queue(DEFAULT_QUEUE,true);}//绑定交换机和队列，并指定路由键//绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRoutingBinding bindingDirect(){return BindingBuilder.bind(TestDirectQueue()).to(directExchange()).with(DEFAULT_ROUTE);}/*** 这个是做什么用的 ,为了后面 生产者确认那，找到交换机，找不到队列用的，* @return*/@BeanDirectExchange lonelyDirectExchange() {return new DirectExchange("lonelyDirectExchange");}}

然后写个简单的接口进行消息推送（根据需求也可以改为定时任务等等，具体看需求），SendMessageController.java：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;/***  模拟 rabbitmq-provider （生产者） 这里模块名字写错了。这个是消息生产者，一般消息的生产者会直接在业务层调用，*  不会单独的搞一个消息生产者，这里因为没有业务调用，去调用这个MQ的生产者，所以这里直接创建一个模块模拟消息生产者** 发送消息控制器（MQ入消息的入口）* //原文链接：https://blog.csdn.net/qq_35387940/article/details/100514134* @author: jd* @create: 2024-06-24*/
@RestController
public class SendMessageController {@AutowiredRabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法/*** 通过postman发送消息给消息队列-直流交换机* @return*/@GetMapping("/sendDirectMessage")String sendDirectMessage(){String messageId = String.valueOf(UUID.randomUUID());String messageData = "test message, hello!";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String,Object> map=new HashMap<>();map.put("messageId",messageId);map.put("messageData",messageData);
//        map.put("messageData","666666");map.put("createTime",createTime);//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchangerabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);//        //生产者发送字符串类型消息，则后面的消息消费者，也需要接受字符串类型的入参进行消费
//        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", "77777");System.out.println("调用完毕");return "ok";}}

把rabbitmq-provider项目运行，调用下接口:

在这里插入图片描述

在这里插入图片描述
因为我们目前还没弄消费者 rabbitmq-consumer，消息没有被消费的，我们去rabbitMq管理页面看看，是否推送成功：（我这里发送了三次，所以有三个消息积压了）

再看看队列（界面上的各个英文项代表什么意思，可以自己查查哈，对理解还是有帮助的）：
在这里插入图片描述

很好，消息已经推送到rabbitMq服务器上面了。

接下来，创建rabbitmq-consumer项目：

pom.xml里的jar依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.atguigu.gulimall</groupId><artifactId>rabbitmq-consumer-true</artifactId><version>0.0.1-SNAPSHOT</version><name>rabbitmq-consumer-true</name><description>RabbitMQ消费者模块</description><url/><licenses><license/></licenses><developers><developer/></developers><scm><connection/><developerConnection/><tag/><url/></scm><properties><java.version>1.8</java.version><!--        <spring-cloud.version>2021.0.4</spring-cloud.version>--><spring-cloud.version>2021.0.1</spring-cloud.version></properties><dependencies><dependency><groupId>com.atguigu.gulimall</groupId><artifactId>gulimall-common</artifactId><version>0.0.1-SNAPSHOT</version><exclusions><exclusion><artifactId>servlet-api</artifactId><groupId>javax.servlet</groupId></exclusion></exclusions></dependency><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.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

然后是 application.yml：

server:port: 8022#数据源配置
spring:datasource:url: jdbc:mysql://192.168.56.10:3306/gulimall_umsusername: rootpassword: rootdriver-class-name:  com.mysql.cj.jdbc.Driver#配置nacoscloud:nacos:discovery:server-addr: 127.0.0.1#配置服务名称application:name: rabbitmq-consumer-true# 配置rabbitMq 服务器#spring.application.name=rabbitmq-consumer-truerabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guest#虚拟host 可以不设置,使用server默认hostvirtual-host: /
#    listener:  #这个在测试消费多个消息的时候，不能有下面这些配置，否则只能消费一个消息后就不继续消费了
#      simple:
#        acknowledge-mode: manual  #指定MQ消费者的确认模式是手动确认模式  这个在消费者者模块配置
#        prefetch: 1 #一次只能消费一条消息   这个在消费者者模块配置#配置日志输出级别
logging:level:com.atguigu.gulimall: debug#配置日志级别

然后一样，创建DirectRabbitConfig.java（消费者单纯的使用，其实可以不用添加这个配置，直接建后面的监听就好，使用注解来让监听器监听对应的队列即可。配置上了的话，其实消费者也是生成者的身份，也能推送该消息。）：

package com.atguigu.gulimall.consumertrue.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/***   消费者配置类**  原文链接：https://blog.csdn.net/qq_35387940/article/details/100514134* 创建DirectRabbitConfig.java  关于队列的配置只是消息的生产者中配置即可。这个消费者不用配置，配置了的话，就也可以当成生产者了* （消费者单纯的使用，其实可以不用添加这个配置，直接建后面的监听就好，* 使用注解来让监听器监听对应的队列即可。配置上了的话，其实消费者也是生成者的身份，也能推送该消息。）：** @author: jd* @create: 2024-06-25*/
@Configuration
public class DirectRabbitConfig {// 声明需要使用的交换机/路由Key/队列的名称public static final String DEFAULT_EXCHANGE = "TestDirectExchange";public static final String DEFAULT_ROUTE = "TestDirectRouting";public static final String DEFAULT_QUEUE = "TestDirectQueue";//队列 起名：TestDirectQueue@Beanpublic Queue TestDirectQueue() {return new Queue(DEFAULT_QUEUE,true);}//Direct交换机 起名：TestDirectExchange@BeanDirectExchange TestDirectExchange() {return new DirectExchange(DEFAULT_EXCHANGE);}//绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting@BeanBinding bindingDirect() {return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(DEFAULT_ROUTE);}}

然后是创建消息接收监听类，RabbitMQListener.java：

package com.atguigu.gulimall.consumertrue.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/*** 消息消费监听类* @author: jd* @create: 2024-06-25*/
@Component
@Slf4j
@RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
public class RabbitMQListener {/*** 当消息发送者发送的是Map的时候，通过这个消息处理器进行处理* @param testMessage*/@RabbitHandler(isDefault = true)public void process(Map testMessage) {System.out.println("RabbitMQListener消费者收到消息  : "+testMessage.toString());}/*** 当消息发送者发送的是String类型的时候，用这个监听处理器去接受消息并处理* @param testMessage*//* @RabbitHandler(isDefault = true)public void process(String testMessage) {System.out.println("DirectReceiver消费者收到消息  : "+testMessage);//正常开发中，会在消费到消息之后，开始做一些业务处理//模拟业务处理//业务开始String str = testMessage + "--消费成功";System.out.println("业务处理完毕"+str);//业务结束}*/}

然后将rabbitmq-consumer-true项目运行起来，可以看到把之前推送的那条消息消费下来了：
在这里插入图片描述
然后可以再继续调用rabbitmq-consumer项目的推送消息接口，可以看到消费者即时消费消息：

消费下来了

那么直连交换机既然是一对一，那如果咱们配置多台监听绑定到同一个直连交互的同一个队列，会怎么样？
在这里插入图片描述
消费的结果如下：

可以看到是实现了轮询的方式对消息进行消费，而且不存在重复消费。

（2）使用Topic Exchange 主题交换机。

在rabbitmq-consume项目里面创建TopicRabbitConfig.java：

package com.atguigu.gulimall.rabbitmqconsumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 使用Topic Exchange 主题交换机。** @author: jd* @create: 2024-06-25*/
@Configuration
public class TopicRabbitConfig {//设置绑定键public static final String man = "topic.man";public static final String woman = "topic.woman";public static final String TOPIC_EXCHANGE = "topicExchange";//创建队列/*** 第一个主题队列** @return*/@Beanpublic Queue firstQueue() {return new Queue(man);}/*** 第二个主题队列** @return*/@Beanpublic Queue secondQueue() {return new Queue(woman);}/*** 创建一个主题交换机** @return TopicExchange*/@BeanTopicExchange exchange() {return new TopicExchange(TOPIC_EXCHANGE);}/*** //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man* //这样只要是消息携带的路由键是topic.man,才会分发到该队列** @return*/@BeanBinding bindingExchangeMessageForFirstQueue() {return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);}/*** //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#* // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列** @return*/@BeanBinding bindingExchangeMessageForSecondQueue() {return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");}}

然后添加多2个接口，用于推送消息到主题交换机：


//    然后添加多2个接口，用于推送消息到主题交换机找那个，再主题交换机中通过设置的路由键来推送到主题为topic.man的队列中以供消费
//    https://blog.csdn.net/qq_35387940/article/details/100514134/*** 用于向MQ发送携带topic.man路由键的消息* @return*/@GetMapping("/sendTopicMessageToMan")public String sendTopicMessageToMan(){String messageId = String.valueOf(UUID.randomUUID());String messageData ="send topic message to man";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String,Object> map=new HashMap<>();map.put(QueueConstant.MESSAGE_ID,messageId);map.put(QueueConstant.MESSAGE_DATA,messageData);map.put(QueueConstant.MESSAGE_TIME,createTime);rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,TopicRabbitConfig.man,map);System.out.println("sendTopicMessageToMan() 执行成功");return "sendTopicMessageToMan is ok";}/*** 用于向MQ发送携带topic.woman路由键的消息。 这样会在exchange中去找绑定中这个路由键绑定的队列，并向其中进行转发* topic.# 这个是通用的绑定规则，只要是携带着topic.开头的就会转发到绑定的这个队列中* https://blog.csdn.net/qq_35387940/article/details/100514134* @return*/@GetMapping("/sendTopicMessageToTotal")public String sendTopicMessageToTotal(){String messageId = String.valueOf(UUID.randomUUID());String messageData ="send topic message to woman";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String,Object> map=new HashMap<>();map.put(QueueConstant.MESSAGE_ID,messageId);map.put(QueueConstant.MESSAGE_DATA,messageData);map.put(QueueConstant.MESSAGE_TIME,createTime);
//        rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,TopicRabbitConfig.woman,map);rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,"topic.woman1",map); //测试携带路由键符合topic.#的是否能转发到topic.woman的队列System.out.println("sendTopicMessageToTotal() 执行成功");return "sendTopicMessageToTotal is ok";}

生产者这边已经完事，先不急着运行，在rabbitmq-consumer-true项目上，创建TopicManListener.java：

package com.atguigu.gulimall.consumertrue.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/** 
主题交换机 监听topic.man队列* @author: jd* @create: 2024-06-25*/
@Component
@Slf4j
@RabbitListener(queues = "topic.man")//监听的队列名称 TestDirectQueue
public class TopicManListener {@RabbitHandlerpublic void process(Map testMessage) {System.out.println("TopicManListener主题消费者收到消息  : "+testMessage.toString());}}

再创建一个TopicTotalListener.java：

package com.atguigu.gulimall.consumertrue.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/*** @author: jd* @create: 2024-06-25*/@Component
@Slf4j
@RabbitListener(queues = "topic.woman")
public class TopicTotalListener {@RabbitHandlerpublic void process(Map testMessage){System.out.println("TopicTotalListener主题消费者收到消息  : "+testMessage.toString());}
}

同样，加主题交换机的相关配置，TopicRabbitConfig.java（消费者一定要加这个配置吗？不需要的其实，理由在前面已经说过了。）：

package com.atguigu.gulimall.rabbitmqconsumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 使用Topic Exchange 主题交换机。** @author: jd* @create: 2024-06-25*/
@Configuration
public class TopicRabbitConfig {//设置绑定键public static final String man = "topic.man";public static final String woman = "topic.woman";public static final String TOPIC_EXCHANGE = "topicExchange";//创建队列/*** 第一个主题队列** @return*/@Beanpublic Queue firstQueue() {return new Queue(man);}/*** 第二个主题队列** @return*/@Beanpublic Queue secondQueue() {return new Queue(woman);}/*** 创建一个主题交换机** @return TopicExchange*/@BeanTopicExchange exchange() {return new TopicExchange(TOPIC_EXCHANGE);}/*** //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man* //这样只要是消息携带的路由键是topic.man,才会分发到该队列** @return*/@BeanBinding bindingExchangeMessageForFirstQueue() {return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);}/*** //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#* // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列** @return*/@BeanBinding bindingExchangeMessageForSecondQueue() {return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");}}

然后把rabbitmq-consumer，rabbitmq-consumer-true两个项目都跑起来，先调用/sendTopicMessage1 接口：
在这里插入图片描述

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver监听队列1，绑定键为：topic.man
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.man

所以可以看到两个监听消费者receiver都成功消费到了消息，因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。
在这里插入图片描述
接下来调用接口/sendTopicMessage2:

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver监听队列1，绑定键为：topic.man
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.woman

所以可以看到两个监听消费者只有TopicTotalReceiver成功消费到了消息。
在这里插入图片描述

（3）使用Fanout Exchang 扇型交换机。

同样地，先在rabbitmq-provider项目上创建FanoutRabbitConfig.java：

package com.atguigu.gulimall.rabbitmqconsumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 使用Fanout Exchang 扇型交换机* @author: jd* @create: 2024-06-25*/
@Configuration
public class FanoutRabbitConfig {//队列名称public static final String  FANOUT_QUEUE_A ="fanout.a";public static final String  FANOUT_QUEUE_B ="fanout.b";public static final String  FANOUT_QUEUE_C ="fanout.c";public static final String  FANOUT_EXCHANGE = "fanout.exchange";//创建队列 FANOUT_QUEUE_A@Beanpublic Queue queueA(){return new Queue(FANOUT_QUEUE_A,true);}//创建队列 FANOUT_QUEUE_B@Beanpublic Queue queueB(){return new Queue(FANOUT_QUEUE_B);}//创建队列 FANOUT_QUEUE_C@Beanpublic Queue queueC(){return new Queue(FANOUT_QUEUE_C);}//创建交换机@Beanpublic FanoutExchange  fanoutExchange(){return new FanoutExchange(FANOUT_EXCHANGE);}//绑定将多有的队列都绑定到这个交换机@BeanBinding bindingExchangeA() {return BindingBuilder.bind(queueA()).to(fanoutExchange());}@BeanBinding bindingExchangeB() {return BindingBuilder.bind(queueB()).to(fanoutExchange());}@BeanBinding bindingExchangeC() {return BindingBuilder.bind(queueC()).to(fanoutExchange());}}

然后是写一个接口用于推送消息，

 /*** 发送消息给扇形交换机 扇型交换机* @return*/@GetMapping("/sendFanoutMessage")public String sendFanoutMessage(){String messageId = String.valueOf(UUID.randomUUID());String messageData = "message: testFanoutMessage ";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String, Object> map = new HashMap<>();map.put(QueueConstant.MESSAGE_ID,messageId);map.put(QueueConstant.MESSAGE_DATA,messageData);map.put(QueueConstant.MESSAGE_TIME,createTime);rabbitTemplate.convertAndSend(FanoutRabbitConfig.FANOUT_EXCHANGE,null,map);System.out.println("sendFanoutMessage() 执行成功");return "sendFanoutMessage is ok";}

接着在rabbitmq-consumer-true项目里加上消息消费类，
在这里插入图片描述

FanoutReceiverA.java：
FanoutReceiverB.java：
FanoutReceiverC.java：

package com.atguigu.gulimall.consumertrue.listener;import com.atguigu.gulimall.consumertrue.config.FanoutRabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/*** 扇形交换机-队列A的监听器,及监听到消息后的处理器* @author: jd* @create: 2024-06-25*/
@Component
@Slf4j
@RabbitListener(queues = FanoutRabbitConfig.FANOUT_QUEUE_A)
public class FanoutReceiverA {@RabbitHandlerpublic void process(Map message){System.out.println("FanoutReceiverA消费者收到消息  : "+message.toString());}}

package com.atguigu.gulimall.consumertrue.listener;import com.atguigu.gulimall.consumertrue.config.FanoutRabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/*** 扇形交换机-队列B的监听器,及监听到消息后的处理器* @author: jd* @create: 2024-06-25*/
@Component
@Slf4j
@RabbitListener(queues = FanoutRabbitConfig.FANOUT_QUEUE_B)
public class FanoutReceiverB {@RabbitHandlerpublic void process(Map message){System.out.println("FanoutReceiverB消费者收到消息  : "+message.toString());}
}

package com.atguigu.gulimall.consumertrue.listener;/*** @author: jd* @create: 2024-06-25*/import com.atguigu.gulimall.consumertrue.config.FanoutRabbitConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;/*** 扇形交换机-队列B的监听器,及监听到消息后的处理器* @author: jd* @create: 2024-06-25*/
@Component
@Slf4j
@RabbitListener(queues = FanoutRabbitConfig.FANOUT_QUEUE_C)
public class FanoutReceiverC {@RabbitHandlerpublic void process(Map message){System.out.println("FanoutReceiverC消费者收到消息  : "+message.toString());}
}

然后加上扇型交换机的配置类，FanoutRabbitConfig.java（消费者真的要加这个配置吗？不需要的其实，理由在前面已经说过了）

package com.atguigu.gulimall.consumertrue.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 使用Fanout Exchang 扇型交换机* @author: jd* @create: 2024-06-25*/
@Configuration
public class FanoutRabbitConfig {//队列名称public static final String  FANOUT_QUEUE_A ="fanout.a";public static final String  FANOUT_QUEUE_B ="fanout.b";public static final String  FANOUT_QUEUE_C ="fanout.c";public static final String  FANOUT_EXCHANGE = "fanout.exchange";//创建队列 FANOUT_QUEUE_A@Beanpublic Queue queueA(){return new Queue(FANOUT_QUEUE_A,true);}//创建队列 FANOUT_QUEUE_B@Beanpublic Queue queueB(){return new Queue(FANOUT_QUEUE_B);}//创建队列 FANOUT_QUEUE_C@Beanpublic Queue queueC(){return new Queue(FANOUT_QUEUE_C);}//创建交换机@Beanpublic FanoutExchange  fanoutExchange(){return new FanoutExchange(FANOUT_EXCHANGE);}//绑定将多有的队列都绑定到这个交换机@BeanBinding bindingExchangeA() {return BindingBuilder.bind(queueA()).to(fanoutExchange());}@BeanBinding bindingExchangeB() {return BindingBuilder.bind(queueB()).to(fanoutExchange());}@BeanBinding bindingExchangeC() {return BindingBuilder.bind(queueC()).to(fanoutExchange());}}

最后将rabbitmq-provider和rabbitmq-consumer项目都跑起来，调用下接口/sendFanoutMessage ：
在这里插入图片描述

可以看到只要发送到 fanoutExchange 这个扇型交换机的消息，三个队列都绑定这个交换机，所以三个消息接收类都监听到了这条消息。

在这里插入图片描述

到了这里其实三个常用的交换机的使用我们已经完毕了，那么接下来我们继续讲讲消息的回调，其实就是消息确认（生产者推送消息成功，消费者接收消息成功）。

三、消息确认种类

RabbitMQ的消息确认有两种。

一种是消息发送确认。这种是用来确认生产者将消息发送给交换器，交换器传递给队列的过程中，消息是否成功投递。发送确认分为两步，一是确认是否到达交换器，二是确认是否到达队列。

第二种是消费接收确认。这种是确认消费者是否成功消费了队列中的消息。

消息确认的作用是什么？

为了防止消息丢失。消息丢失分为发送丢失和消费者处理丢失，相应的也有两种确认机制。

先来一起学习一下：

A：消息发送确认

在rabbitmq-consumer项目的application.yml文件上，加上消息确认的配置项后：

server:port: 8021#数据源配置
spring:datasource:username: rootpassword: rooturl: jdbc:mysql://192.168.56.10:3306/gulimall_umsdriver-class-name: com.mysql.cj.jdbc.Driver#注册到注册中心cloud:nacos:discovery:server-addr: 127.0.0.1:8848application:name: rabbitmq-consumer#配置rabbitMq 服务器rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guest#虚拟host 可以不设置,使用server默认hostvirtual-host: /publisher-returns: true  #确认消息已发送到队列(Queue)  这个在生产者模块配置 这个后期再配置，这会还用不到publisher-confirm-type: correlated   #确认消息已发送到交换机(Exchange) 这个在生产者模块配置 这个后期再配置，这会还用不到logging:level:com.atguigu.gulimall: debug   #调整product模块日志的输出模式是debug级别，这样就能在控制台看到dao包下的输出日志了。

然后是配置相关的消息确认回调函数，RabbitConfig.java：

package com.atguigu.gulimall.rabbitmqconsumer.config;import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 配置相关的消息确认回调函数，RabbitConfig.java：* https://blog.csdn.net/qq_35387940/article/details/100514134** 先从总体的情况分析，推送消息存在四种情况：** ①消息推送到server，但是在server里找不到交换机* ②消息推送到server，找到交换机了，但是没找到队列* ③消息推送到sever，交换机和队列啥都没找到* ④消息推送成功* 具体哪些会触发回调，分别又会触发哪个函数，看下面的测试** @author: jd* @create: 2024-06-25*/
@Configuration
public class RabbitConfig {@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate =new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);System.out.println("ConfirmCallback:     "+"确认情况："+ack);System.out.println("ConfirmCallback:     "+"原因："+cause);}});rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {System.out.println("ReturnCallback:     "+"消息："+returnedMessage.getMessage());System.out.println("ReturnCallback:     "+"回应码："+returnedMessage.getReplyCode());System.out.println("ReturnCallback:     "+"回应信息："+returnedMessage.getReplyText());System.out.println("ReturnCallback:     "+"交换机："+returnedMessage.getExchange());System.out.println("ReturnCallback:     "+"路由键："+returnedMessage.getRoutingKey());}});return rabbitTemplate;}
}

到这里，生产者推送消息的消息确认调用回调函数已经完毕。
可以看到上面写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？

先从总体的情况分析，推送消息存在四种情况：

①消息推送到server，但是在server里找不到交换机
②消息推送到server，找到交换机了，但是没找到队列
③消息推送到sever，交换机和队列啥都没找到
④消息推送成功

那么我先写几个接口来分别测试和认证下以上4种情况，消息确认触发回调函数的情况：

①消息推送到server，但是在server里找不到交换机 （是否到达交换机）
写个测试接口，把消息推送到名为‘non-existent-exchange’的交换机上（这个交换机是没有创建没有配置的）：

/*** ①消息推送到server，但是在server里找不到交换机** 写个测试接口，把消息推送到名为‘non-existent-exchange’的交换机上（这个交换机是没有创建没有配置的）* 调用接口，查看rabbitmq-provuder项目的控制台输出情况（原因里面有说，没有找到交换机'non-existent-exchange'）：*在控制台中* 调用后返回：http://localhost:8021/TestMessageAck*ConfirmCallback:     相关数据：null* ConfirmCallback:     确认情况：false* ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404,* reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/', class-id=60, method-id=40)**  结论： ①这种情况触发的是 ConfirmCallback 回调函数* @return*/@GetMapping("/TestMessageAck")public String TestMessageAck() {String messageId = String.valueOf(UUID.randomUUID());String messageData = "message: non-existent-exchange test message ";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String, Object> map = new HashMap<>();map.put("messageId", messageId);map.put("messageData", messageData);map.put("createTime", createTime);rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);return "ok";}

调用接口，查看rabbitmq-provuder项目的控制台输出情况（原因里面有说，没有找到交换机’non-existent-exchange’）：
在这里插入图片描述
结论： ①这种情况触发的是 ConfirmCallback 回调函数。

②消息推送到server，找到交换机了，但是没找到队列（是否到达队列）
这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在DirectRabitConfig里面新增一个直连交换机，名叫‘lonelyDirectExchange’，但没给它做任何绑定配置操作：

    @BeanDirectExchange lonelyDirectExchange() {return new DirectExchange("lonelyDirectExchange");}

然后写个测试接口，把消息推送到名为‘lonelyDirectExchange’的交换机上（这个交换机是没有任何队列配置的）：

/*** ②消息推送到server，找到交换机了，但是没找到队列* 这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，* 我来简单地在DirectRabitConfig里面新增一个直连交换机，名叫‘lonelyDirectExchange’，但没给它做任何绑定配置操作：** 然后写个测试接口，把消息推送到名为‘lonelyDirectExchange’的交换机上（这个交换机是没有任何队列配置的）：**可以看到这种情况，在控制台中 两个函数都被调用了；* 这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；* 而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。** 调用后返回：http://localhost:8021/TestMessageAck2* ReturnCallback:     回应码：312* ReturnCallback:     回应信息：NO_ROUTE* ReturnCallback:     交换机：lonelyDirectExchange* ReturnCallback:     路由键：TestDirectRouting* ConfirmCallback:     相关数据：null* ConfirmCallback:     确认情况：true* ConfirmCallback:     原因：null**   结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。* @return*/@GetMapping("/TestMessageAck2")public String TestMessageAck2() {String messageId = String.valueOf(UUID.randomUUID());String messageData = "message: lonelyDirectExchange test message ";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String, Object> map = new HashMap<>();map.put("messageId", messageId);map.put("messageData", messageData);map.put("createTime", createTime);rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);  //lonelyDirectExchange这个交换机没有和任何队列做绑定，return "ok";}

调用接口，查看rabbitmq-provuder项目的控制台输出情况：
在这里插入图片描述

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null
ReturnCallback:     消息：(Body:'[serialized object]' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：lonelyDirectExchange
ReturnCallback:     路由键：TestDirectRouting

可以看到这种情况，两个函数都被调用了；
这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；
而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。
结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。

③消息推送到sever，交换机和队列啥都没找到
这种情况其实一看就觉得跟①很像，没错，③和①情况回调是一致的，所以不做结果说明了。
结论： ③这种情况触发的是 ConfirmCallback 回调函数。

④消息推送成功
那么测试下，按照正常调用之前消息推送的接口就行，就调用下 /sendFanoutMessage接口，可以看到控制台输出：

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

结论： ④这种情况触发的是 ConfirmCallback 回调函数。

总结：
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){}通过设置这个参数，其中使用内部类进行实现，来记录消息发送到交换器Exchange后触发回调。
（使用该功能需要开启确认， publisher-confirm-type: correlated #确认消息已发送到交换机(Exchange) 这个在生产者模块配置
）

rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback(){}）通过设置这个参数，如果消息从交换器发送到对应队列失败时触发（比如根据发送消息时指定的routingKey找不到队列时会触发）
（ publisher-returns: true #确认消息已发送到队列(Queue) 这个在生产者模块配置 ）

以上是生产者推送消息的消息确认回调函数的使用介绍（可以在回调函数根据需求做对应的扩展或者业务数据处理）。

B：消费接收确认

接下来我们继续，消费者接收到消息的消息确认机制。

（1）确认模式

AcknowledgeMode.NONE：不确认
AcknowledgeMode.AUTO：自动确认
AcknowledgeMode.MANUAL：手动确认
spring-boot中配置方法：

spring.rabbitmq.listener.simple.acknowledge-mode = manual

（2）手动确认
在这里插入图片描述
未确认的消息数

上图为channel中未被消费者确认的消息数。

通过RabbitMQ的host地址加上默认端口号15672访问管理界面。

（2.1）成功确认

void basicAck(long deliveryTag, boolean multiple) throws IOException;

deliveryTag:该消息的index

multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。

消费者成功处理后，调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false)方法对消息进行确认。

（2.2）失败确认

void basicNack(long deliveryTag, boolean multiple, boolean requeue)

throws IOException;

deliveryTag:该消息的index。

multiple：是否批量. true：将一次性拒绝所有小于deliveryTag的消息。

requeue：被拒绝的是否重新入队列。

void basicReject(long deliveryTag, boolean requeue) throws IOException;

deliveryTag:该消息的index。

requeue：被拒绝的是否重新入队列。

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息，而basicReject一次只能拒绝一条消息。

①自动确认， 这也是默认的消息确认情况。  AcknowledgeMode.NONE
RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。
一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。② 根据情况确认， 这个不做介绍
③ 手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
basic.ack用于肯定确认 
basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展） 
basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息 消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理：着重讲下reject，因为有时候一些场景是需要重新入列的。channel.basicReject(deliveryTag, true);  拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。 下次不想再消费这条消息了。使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。channel.basicNack(deliveryTag, false, true);
第一个参数依然是当前消息到的数据的唯一id;
第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

看了上面这么多介绍，接下来我们一起配置下，看看一般的消息接收手动确认是怎么样的。

方式一：通过配置类的方式实现

此时还不需要加下面的配置，因为这种方式是通过配置类注解来配置的手动消费者确认，再下面的方式二则是通过yml的配置来设置的消费者手动确认，我们先来看方式一是怎么实现的
在这里插入图片描述

在消费者项目里，
新建MessageListenerConfig.java上添加代码相关的配置代码：

package com.atguigu.gulimall.consumertrue.config;import com.atguigu.gulimall.consumertrue.listener.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 一般的消息接收 手动确认是怎么样的，消费者的手动消息确认，配置类* https://blog.csdn.net/qq_35387940/article/details/100514134* @author: jd* @create: 2024-06-25*/
//@Configuration     //注释掉这个注解，这样第一种MQ消费者的确认模式就失效了，以为你这个里面配置着对某个队列的监控呢。 第二种MQ的配置方式的话和这个的区别，不用这种配置类，而是在yml中配置东西
public class MessageListenerConfig {@Autowiredprivate CachingConnectionFactory connectionFactory;@Autowiredprivate MyAckReceiver myAckReceiver;//消息接收处理类@Beanpublic SimpleMessageListenerContainer simpleMessageListenerContainer(){SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);container.setConcurrentConsumers(1);container.setMaxConcurrentConsumers(1);container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息//设置一个队列,在这里设置了队列，container.setQueueNames("TestDirectQueue");//如果同时设置多个如下： 前提是队列都是必须已经创建存在的//  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");//另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues//container.setQueues(new Queue("TestDirectQueue",true));//container.addQueues(new Queue("TestDirectQueue2",true));//container.addQueues(new Queue("TestDirectQueue3",true));//这里设置了监听器，因为上面设置了队列，所以在监听器中就不需要用监听器的注解了 。container.setMessageListener(myAckReceiver);return container;}
}

对应的手动确认消息监听类，MyAckReceiver.java（手动确认模式需要实现 ChannelAwareMessageListener）：
//之前的相关监听器可以先注释掉，以免造成多个同类型监听器都监听同一个队列。【比如我之前用的RabbitMQListener 、RabbitMQListener2 为了让其失效，直接注释掉其中的//@RabbitListener(queues = “TestDirectQueue”)//监听的队列名称 TestDirectQueue】这个注解即可，这样这个监听器就无法监听相关队列了。

在这里插入图片描述
MyAckReceiver.java

package com.atguigu.gulimall.consumertrue.listener;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.Map;/*** 对应的手动确认消息监听类，MyAckReceiver.java（手动确认模式需要实现 ChannelAwareMessageListener）：* //之前的相关监听器可以先注释掉，以免造成多个同类型监听器都监听同一个队列。** 注意：因为这里是在MessageListenerConfig 类中指定了是要监听哪个队列，以及消息的确认机制，所以这里不需要使用* @RabbitListener(queues = "TestDirectQueue")  和 @RabbitHandler(isDefault = true)注解了* @author: jd* @create: 2024-06-25*/@Component
public class MyAckReceiver implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {byte[] body = message.getBody();ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(body));Map<String,String> msgMap  = (Map<String,String>)objectInputStream.readObject();String messageId = msgMap.get("messageId");String messageData = msgMap.get("messageData");String createTime = msgMap.get("createTime");objectInputStream.close();System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);System.out.println("消费的主题队列来自："+message.getMessageProperties().getConsumerQueue());
//        消费者成功处理后，调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false)方法对消息进行确认。channel.basicAck(deliveryTag, true); // deliveryTag:该消息的index   multiple：是否批量. true：将一次性ack所有小于deliveryTag的消息。    第二个参数，手动确认可以被批处理， 当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//		channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}
}

这时，先调用接口/sendDirectMessage，给直连交换机TestDirectExchange 的队列TestDirectQueue 推送一条消息，可以看到监听器正常消费了下来：
在这里插入图片描述

第一次验证我们发现，消费者没有消费掉直流交换机中的消息，而且也在直流队列中积压了起来，

这是由于我们的配置类忘记加了 @Configuration 注解了，所以此时这个不是配置类，也就是这里对MQ的配置不会生效，所以加上之后，我们再去试试：
在这里插入图片描述
可看到下图消费成功

配置类中 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息是发挥作用的关键；

方式二：通过yml配置来完成消费者确认

特别注意：因为这里我们要使用yml配置来实现，所以我们需要关闭配置类的作用，使之失效，我这里直接把@Configuration 给注释掉了，这样配置类不会起作用了！！^_
在这里插入图片描述
第二种方式正式开始啦 (#^.#)
首先我们来在yml中开启手动确认的配置

server:port: 8022#数据源配置
spring:datasource:url: jdbc:mysql://192.168.56.10:3306/gulimall_umsusername: rootpassword: rootdriver-class-name:  com.mysql.cj.jdbc.Driver#配置nacoscloud:nacos:discovery:server-addr: 127.0.0.1#配置服务名称application:name: rabbitmq-consumer-true# 配置rabbitMq 服务器#spring.application.name=rabbitmq-consumer-truerabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guest#虚拟host 可以不设置,使用server默认hostvirtual-host: /listener:  #这个在测试消费多个消息的时候，不能有下面这些配置，否则只能消费一个消息后就不继续消费了simple:acknowledge-mode: manual  #指定MQ消费者的确认模式是手动确认模式  这个在消费者者模块配置prefetch: 1 #一次只能消费一条消息   这个在消费者者模块配置#配置日志输出级别
logging:level:com.atguigu.gulimall: debug#配置日志级别

其中的几行是开启的关键
listener: #这个在测试消费多个消息的时候，不能有下面这些配置，否则只能消费一个消息后就不继续消费了
simple:
acknowledge-mode: manual #指定MQ消费者的确认模式是手动确认模式这个在消费者者模块配置
prefetch: 1 #一次只能消费一条消息这个在消费者者模块配置

此处直接用接口来当生产者了；
然后我们在生产者模块用于放消息的controller中增加一个放消息的请求方法，用于往队列里面连续放入5个放消息
SendMessageController.java

    /*** 原文链接：https://blog.csdn.net/weixin_45724872/article/details/119655638* 将信号放入MQ* @param message* @return*/@PostMapping("/msg/muscle")public String receiveMuscleSign(@RequestBody String message) {//处理业务for (int i = 1; i <= 5; i++) {rabbitTemplate.convertAndSend("muscle_fanout_exchange","",message+i);}return " receiveMuscleSign ok";}

开发消费者
此处用一个类下的两个方法来模拟2个消费者

package com.atguigu.gulimall.consumertrue.listener;import com.rabbitmq.client.Channel;
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;/****此处用一个类下的两个方法来模拟2个消费者*原文链接：https://blog.csdn.net/weixin_45724872/article/details/119655638原文链接：https://blog.csdn.net/weixin_45724872/article/details/119655638* @author: jd* @create: 2024-06-25*/
@Component
public class MyConsumerListener {@RabbitListener(bindings = {@QueueBinding(value = @Queue("consumer_queue_1"),//绑定交换机exchange = @Exchange(value = "muscle_fanout_exchange", type = "fanout"))})public void consumer1(String msg, Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("消费者1 => " + msg);//channel.basicAck(deliveryTag, false); // 因为 yml中 prefetch 设置为 1（或未设置，因为默认可能是 0，表示无限制，但这不是推荐的做法），RabbitMQ 将只发送一个消息给消费者，并等待该消息的确认。在这种情况下，// 如果你注释掉了 channel.basicAck，消费者将只能消费一个消息，并且不会收到下一个消息，直到你发送确认或关闭连接。 所以对于消息队列中的五个消息只能销费一个，除非你手动确认，否则不会再消费其他的消息} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}@RabbitListener(bindings = {@QueueBinding(value = @Queue("consumer_queue_2"),//绑定交换机exchange = @Exchange(value = "muscle_fanout_exchange", type = "fanout"))})public void consumer2(String msg,Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("消费者2 => " + msg);channel.basicAck(deliveryTag, false);} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}}

注意一点，消费者1的手动ACK我们是注释掉了

而消费者2的手动ACK我们是开着的

原因是为了对照试验

我们期望的情况是：一共5条消息，消费者1和2都一一处理；

处理完毕后再取下一条，否则不让取；

那么按我们代码这样写；

消费者1只能取一条 (只是处理一条的原因，)
而消费者2则能取满5条（因为消费者1的手动ACK被我们注释了，此处又不是自动ACK）

消费者1只是处理一条的原因：下图中的perfetchCount有问题，我们实际上配置的是prefetch: 1 ，我们直接按照这个配置来理解就行
在这里插入图片描述
消费者一，就是注释了对消息消费之后的确认回馈给RabbitMQ的设置，所以消费者对五条消息中消费到第一个之后，因为我们在yml中又配置了每次消费一条，而且也是手动确认的，所以MQ消费到这一条之后，就在那等着手动调用ack方法来完成的确认ack的反馈，结果我们这里注释了，所以就一直等不到第一条消息的回馈，所以就会一直等待，下面的4条消息也就无法继续消费了，

相反，消费者二就不一样了，他有消费完每一条消息之后，都调用了手动ack的回馈，所以可以消费5条消息，都消息完。

以下是实验截图
MQ 的初始状态：
在这里插入图片描述

首先用postman发送请求
在这里插入图片描述

看下图，生产者发送了5条消息，并得到了成功推送到了交换机和队列的回馈
在这里插入图片描述

接下来我们步入正题：看消费者里面，消费者1只是消费了一条，消费者2消费了全部的5条消息；
在这里插入图片描述
结果和我们预想的是一致的；

我们在看看MQ的管理页面来确认
在这里插入图片描述
可以看到，消费者2已经搞完了，而消费者1那边卡住了（消费者一消费了一条，但是在等待回馈，还剩余4条都没被消费，在等待消费）

我在实验的过程中，因为消费者1中的消息堆积了，如果再次发送5条消息到扇形交换机中，那队列1中会积累到9条待消费的，1条等待反馈的，10条总共的，我们可以实验一下子：
在这里插入图片描述
结果和我们预想的一样，那我们如何将这些积压的消息给去掉呢？
我自己试出了两种方式，最初试的直接重启服务，这样是无效的，因为进入队列的不被消费会一直在队列里面。
下面是2种处理方法：
第一种是最直接的方法，直接把确认那行的代码给放开，这样这个消费者1 就会把队列1中积压的那些给消费掉了
第二种我们将yml中的手动确认配置注释掉，这样就默认是自动确认了，这样我每次从postman中发送5条消息到扇形交换机，分发到两个队列之后，两个消费者都会一直可以消费，因为没消费一个都会自动确认回馈，不用等待了，这样也是可以的
我们实验如下：
实验1：
我们先把消费者1中注释的手动回馈给放开
在这里插入图片描述
可见console中，对于积压的消息直接给消费掉了。

实验2：
我们将消费者1中的手动反馈，给继续注释掉，发送2次 postman；

在这里插入图片描述
造成积压

我把yml中的手动消费者确认，改成自动的，也就是注释掉，可以看到，重启消费者模块后，积压的也被消费了
注释配置：

重启后，看控制台：很明显启动后，积压的消息也被消费了，

在MQ控制台中也可以看到，积压消息被消费啦
在这里插入图片描述

关于手动确认的一些方法
细心的小伙伴可能发现了我们在消费者的catch处写了这样一行代码

channel.basicReject(deliveryTag, false);

以下是解释

一般是有3种确认的，其中1种是正确确认，另外2种是错误确认；

reject:只能否定一条消息
nack:可以否定一条或者多条消息

而错误确认的这两个，都有一个属性
boolean requeue

当它是true的时候，表示重新入队；
当它是false的时候，则表示抛弃掉；

使用拒绝后重新入列这个确认模式要谨慎，因为触发错误确认一般都是出现异常的时候，那么就可能导致死循环，即不断的入队-消费-报错-重新入队…；这将导致消息积压，万一就炸了…

实验错误确认
我们将上述的消费者代码加一行代码；

此处只改动了消费者1，消费者2不变

新增一条抛异常的语句
int num = 1/0;

package com.tubai;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;@Component
public class MyConsumer {@RabbitListener(bindings = {@QueueBinding(value = @Queue("consumer_queue_1"),//绑定交换机exchange = @Exchange(value = "muscle_fanout_exchange", type = "fanout"))})public void consumer1(String msg,Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("消费者1 => " + msg);int num = 1/0;channel.basicAck(deliveryTag, false); //第二个参数，手动确认可以被批处理，当该参数为 true 时} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}@RabbitListener(bindings = {@QueueBinding(value = @Queue("consumer_queue_2"),//绑定交换机exchange = @Exchange(value = "muscle_fanout_exchange", type = "fanout"))})public void consumer2(String msg,Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("消费者2 => " + msg);channel.basicAck(deliveryTag, false);} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}
}

运行结果
在这里插入图片描述
可以看到我们的消费者1也正常了，因为我们是先打印后确认，因此1~5也会被打印出来；

如果重复入队…那么我们的程序就会死循环了，疯狂打印，各位可以自己试试；但是容易把内存占满O。。

本篇文章书写不易，自己打了好久，大家认可的话，或者开启了新认知，请给个点赞。收藏哦 (#^.#) 谢谢大家！
参考文章也写的超级好，大家也可都学习学习，一起进步
Springboot 整合RabbitMq ，用心看完这一篇就够了
RabbitMQ的消息确认机制
SpringBoot集成RabbitMq 手动ACK
RabbitMQ控制界面详解