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

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

文章目录

  • 前言
  • 一、对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: root
    password: root
    url: jdbc:mysql://192.168.56.10:3306/gulimall_ums
    driver-class-name: com.mysql.cj.jdbc.Driver

  #注册到注册中心
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
  application:
    name: rabbitmq-consumer


  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #虚拟host 可以不设置,使用server默认host
    virtual-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";


    // 声明交换机，需要几个声明几个，这里就一个
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange(DEFAULT_EXCHANGE);
    }

    //创建队列
    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue(){
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue(DEFAULT_QUEUE,true);
    }

    //绑定交换机和队列，并指定路由键
    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    Binding bindingDirect(){
        return BindingBuilder.bind(TestDirectQueue()).to(directExchange()).with(DEFAULT_ROUTE);
    }

    /**
     * 这个是做什么用的 ,为了后面 生产者确认那，找到交换机，找不到队列用的，
     * @return
     */
    @Bean
    DirectExchange 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 {
 
    @Autowired
    RabbitTemplate 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 发送到交换机TestDirectExchange
        rabbitTemplate.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_ums
    username: root
    password: root
    driver-class-name:  com.mysql.cj.jdbc.Driver
  #配置nacos
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1
  #配置服务名称
  application:
    name: rabbitmq-consumer-true
  # 配置rabbitMq 服务器
  #spring.application.name=rabbitmq-consumer-true
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #虚拟host 可以不设置,使用server默认host
    virtual-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
    @Bean
    public Queue TestDirectQueue() {
        return new Queue(DEFAULT_QUEUE,true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange(DEFAULT_EXCHANGE);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding 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
     */
    @Bean
    public Queue firstQueue() {
        return new Queue(man);
    }

    /**
     * 第二个主题队列
     *
     * @return
     */
    @Bean
    public Queue secondQueue() {
        return new Queue(woman);
    }

    /**
     * 创建一个主题交换机
     *
     * @return TopicExchange
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }


    /**
     * //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
     * //这样只要是消息携带的路由键是topic.man,才会分发到该队列
     *
     * @return
     */
    @Bean
    Binding bindingExchangeMessageForFirstQueue() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }


    /**
     * //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
     * // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
     *
     * @return
     */
    @Bean
    Binding 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 {

    @RabbitHandler
    public 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 {

    @RabbitHandler
    public 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
     */
    @Bean
    public Queue firstQueue() {
        return new Queue(man);
    }

    /**
     * 第二个主题队列
     *
     * @return
     */
    @Bean
    public Queue secondQueue() {
        return new Queue(woman);
    }

    /**
     * 创建一个主题交换机
     *
     * @return TopicExchange
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }


    /**
     * //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
     * //这样只要是消息携带的路由键是topic.man,才会分发到该队列
     *
     * @return
     */
    @Bean
    Binding bindingExchangeMessageForFirstQueue() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }


    /**
     * //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
     * // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
     *
     * @return
     */
    @Bean
    Binding 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
    @Bean
    public Queue queueA(){
       return new Queue(FANOUT_QUEUE_A,true);
    }

    //创建队列 FANOUT_QUEUE_B
    @Bean
    public Queue queueB(){
        return new Queue(FANOUT_QUEUE_B);
    }

    //创建队列 FANOUT_QUEUE_C
    @Bean
    public Queue queueC(){
        return new Queue(FANOUT_QUEUE_C);
    }

    //创建交换机
    @Bean
    public FanoutExchange  fanoutExchange(){
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    //绑定将多有的队列都绑定到这个交换机
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    @Bean
    Binding 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 {

    @RabbitHandler
    public 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 {

    @RabbitHandler
    public 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 {

    @RabbitHandler
    public 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
    @Bean
    public Queue queueA(){
       return new Queue(FANOUT_QUEUE_A,true);
    }

    //创建队列 FANOUT_QUEUE_B
    @Bean
    public Queue queueB(){
        return new Queue(FANOUT_QUEUE_B);
    }

    //创建队列 FANOUT_QUEUE_C
    @Bean
    public Queue queueC(){
        return new Queue(FANOUT_QUEUE_C);
    }

    //创建交换机
    @Bean
    public FanoutExchange  fanoutExchange(){
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    //绑定将多有的队列都绑定到这个交换机
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    @Bean
    Binding 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: root
    password: root
    url: jdbc:mysql://192.168.56.10:3306/gulimall_ums
    driver-class-name: com.mysql.cj.jdbc.Driver

  #注册到注册中心
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
  application:
    name: rabbitmq-consumer


  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #虚拟host 可以不设置,使用server默认host
    virtual-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 {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate =new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){

            @Override
            public 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() {
            @Override
            public 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’，但没给它做任何绑定配置操作：

    @Bean
    DirectExchange 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 {

    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收处理类

    @Bean
    public 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 {

    @Override
    public 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_ums
    username: root
    password: root
    driver-class-name:  com.mysql.cj.jdbc.Driver
  #配置nacos
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1
  #配置服务名称
  application:
    name: rabbitmq-consumer-true
  # 配置rabbitMq 服务器
  #spring.application.name=rabbitmq-consumer-true
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #虚拟host 可以不设置,使用server默认host
    virtual-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控制界面详解

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

【网络安全】修改Host文件实现域名解析

【网络安全】修改Host文件实现域名解析

场景 开发一个网站或者服务&#xff0c;需要在本地测试时&#xff0c;可以将线上的域名指向本地开发环境的IP地址。从而模拟真实环境中的域名访问&#xff0c;方便调试和开发。 步骤 1、以管理员身份打开命令提示符 2、编辑hosts文件&#xff1a; 输入以下命令打开hosts文…
阅读更多...
【Python数据分析及环境搭建】：教程详解1(第23天)

【Python数据分析及环境搭建】：教程详解1(第23天)

系列文章目录 Python进行数据分析的优势常用Python数据分析开源库介绍启动Jupyter服务Jupyter Notebook的使用 文章目录 系列文章目录前言学习目标1. Python进行数据分析的优势2. 常用Python数据分析开源库介绍2.1 NumPy2.2 Pandas2.3 Matplotlib2.4 Seaborn2.5 Sklearn2.6 Ju…
阅读更多...
【PB案例学习笔记】-26制作一个带浮动图标的工具栏

【PB案例学习笔记】-26制作一个带浮动图标的工具栏

写在前面 这是PB案例学习笔记系列文章的第26篇&#xff0c;该系列文章适合具有一定PB基础的读者。 通过一个个由浅入深的编程实战案例学习&#xff0c;提高编程技巧&#xff0c;以保证小伙伴们能应付公司的各种开发需求。 文章中设计到的源码&#xff0c;小凡都上传到了gite…
阅读更多...
怎样使用js技术实现Chrome投屏功能？

怎样使用js技术实现Chrome投屏功能？

在Web前端技术中&#xff0c;直接控制浏览器窗口或标签页从主屏投屏到副屏&#xff08;如PPT的演讲者模式&#xff09;并不简单&#xff0c;而且直接控制浏览器窗口从主屏投屏到副屏的功能超出了Web标准的范畴&#xff0c;并且涉及到用户系统级别的设置和权限&#xff0c;因此不…
阅读更多...
正确认识手机NFC，安全无风险

正确认识手机NFC，安全无风险

在数字化生活日益普及的今天&#xff0c;NFC&#xff08;近场通信&#xff09;技术以其独特的便捷性和高效性&#xff0c;逐渐成为了我们日常生活中不可或缺的一部分。然而&#xff0c;面对新技术的崛起&#xff0c;总有一些用户对于其安全性心存疑虑&#xff0c;尤其是关于“N…
阅读更多...
（一）Docker基本介绍

（一）Docker基本介绍

部署项目的发展 传统部署适合需要最大性能和可靠性的场景&#xff0c;但在资源利用和管理方面有显著劣势。虚拟化部署提供了良好的资源利用率和隔离性&#xff0c;适用于需要灵活扩展和多租户环境的场景&#xff0c;但存在性能开销。容器部署在轻量级、可移植性和资源利用率方面…
阅读更多...
JVM线上监控环境搭建Grafana+Prometheus+Micrometer

JVM线上监控环境搭建Grafana+Prometheus+Micrometer

架构图 一: SpringBoot自带监控Actuator SpringBoot自带监控功能Actuator&#xff0c;可以帮助实现对程序内部运行情况监控&#xff0c;比如监控内存状况、CPU、Bean加载情况、配置属性、日志信息、线程情况等。 使用步骤&#xff1a; 1. 导入依赖坐标 <dependency><…
阅读更多...
Omni3D目标检测

Omni3D目标检测

Omni3D是一个针对现实场景中的3D目标检测而构建的大型基准和模型体系。该项目旨在推动从单一图像中识别3D场景和物体的能力&#xff0c;这对于计算机视觉领域而言是一个长期的研究目标&#xff0c;并且在机器人、增强现实&#xff08;AR&#xff09;、虚拟现实&#xff08;VR&a…
阅读更多...
7.1.SQL注入-基于函数报错的方式来利用updatexml()

7.1.SQL注入-基于函数报错的方式来利用updatexml()

基于函数报错的方式来进行利用-字符型&#xff08;本页updatexml()&#xff09; 前提条件是后台数据库没有屏蔽数据库语法报错信息 updatexml()方法详解 注释&#xff1a; 第一个参数&#xff0c;意思就是xml文档的名称 第二个参数&#xff0c;意思就是定位到xml文档中指定…
阅读更多...
PriorityQueue底层你了解多少？（带你彻底掌握优先级队列）

PriorityQueue底层你了解多少？（带你彻底掌握优先级队列）

1. 概念 队列是一种先进先出(FIFO)的数据结构&#xff0c;但有些情况下&#xff0c;操作的数据可能带有优先级&#xff0c;一般出队列时&#xff0c;可能需要优先级高的元素先出队列&#xff0c;该中场景下&#xff0c;使用队列显然不合适&#xff0c;比如:在手机上玩游戏的时…
阅读更多...
[图解]SysML和EA建模住宅安全系统-05-参数图

[图解]SysML和EA建模住宅安全系统-05-参数图

1 00:00:01,140 --> 00:00:03,060 这是实数没错&#xff0c;这是分钟 2 00:00:03,750 --> 00:00:07,490 但是你在这里选&#xff0c;选不了的 3 00:00:07,500 --> 00:00:09,930 因为它这里不能够有那个 4 00:00:11,990 --> 00:00:13,850 但是我们前面这里 5 00…
阅读更多...
D - Intersecting Intervals（abc355）

D - Intersecting Intervals（abc355）

题意&#xff1a;有n个区间&#xff0c;找出俩俩区间相交的个数 分析&#xff1a; 设初始俩俩相交&#xff0c;找出不相交的&#xff08;不同区间l>r)&#xff0c;减去即可 #include<bits/stdc.h> using namespace std; typedef long long ll; int main(){ ios:…
阅读更多...
大力出奇迹：大语言模型的崛起与挑战

大力出奇迹：大语言模型的崛起与挑战

随着人工智能&#xff08;AI&#xff09;技术的迅猛发展&#xff0c;特别是在自然语言处理&#xff08;NLP&#xff09;领域&#xff0c;大语言模型&#xff08;LLM&#xff09;的出现与应用&#xff0c;彻底改变了我们与机器互动的方式。本文将探讨ChatGPT等大语言模型的定义、…
阅读更多...
Hive-存储-文件格式

Hive-存储-文件格式

一、前言 数据存储是Hive的基础&#xff0c;选择合适的底层数据存储格式&#xff0c;可以在不改变Hql的前提下得到大的性能提升。类似mysql选择适合场景的存储引擎。 Hive支持的存储格式有 文本格式&#xff08;TextFile&#xff09; 二进制序列化文件 &#xff08;SequenceF…
阅读更多...
期末复习---程序填空

期末复习---程序填空

注意&#xff1a; 1.数组后移 *p *(p-1) //把前一个数赋值到后一个数的位置上来覆盖后一个数 2.指针找最大字符 max *p while( *p){ if( max< *p) { max*p; qp;/ 用新的指针指向这个已经找到的最大位置&#xff1b;!!!!!!!!! } p; //因为开始没有next &#xff…
阅读更多...
Fragment+Viewpage2+FragmentStateAdapter实现滑动式标签布局

Fragment+Viewpage2+FragmentStateAdapter实现滑动式标签布局

大家好&#xff0c;我是网创有方&#xff0c;今天记录下标签布局的实现方法&#xff0c;先看下效果图。 第一步&#xff1a;编写一个activity或者fragment。内含有一个viewpager2的适配器&#xff0c;适配器类型为FragmentStateAdapter。 ​ public class MediaCreateFragment…
阅读更多...
计算机图形学入门22：双向反射分布函数(BRDF)

计算机图形学入门22：双向反射分布函数(BRDF)

1.定义 所谓BRDF&#xff08;Bidirectional Reflectance Distribution Function&#xff0c;双向反射分布函数&#xff09;&#xff0c;指的是从辐射度量学的角度去理解光线的反射&#xff0c;如下图所示。 所谓反射就是一个点从ωi方向发出的Radiance转化为dA接收到的功率E&am…
阅读更多...
在jeesite框架中增加一个收藏夹功能-- V1.0版本

在jeesite框架中增加一个收藏夹功能-- V1.0版本

Jeesite简介&#xff1a;JeeSite 快速开发平台&#xff0c;不仅仅是一个后台开发框架&#xff0c;它是一个企业级快速开发解决方案&#xff0c;提供在线数据源管理、数据表建模、代码生成等功能。 正文&#xff1a;Jeesite是一个方便实用&#xff0c;适合敏捷开发&#xff0c;提…
阅读更多...
openmetadata1.3.1 自定义连接器 开发教程

openmetadata1.3.1 自定义连接器 开发教程

openmetadata自定义连接器开发教程 一、开发通用自定义连接器教程 官网教程链接&#xff1a; 1.https://docs.open-metadata.org/v1.3.x/connectors/custom-connectors 2.https://github.com/open-metadata/openmetadata-demo/tree/main/custom-connector &#xff08;一&…
阅读更多...
Webpack: 其他性能优化

Webpack: 其他性能优化

概述 前面章节我们已经详细探讨 Webpack 中如何使用分包、代码压缩提升应用执行性能。除此之外&#xff0c;还有不少普适、细碎的方法&#xff0c;能够有效降低应用体积&#xff0c;提升网络分发性能&#xff0c;包括&#xff1a; 使用动态加载&#xff0c;减少首屏资源加载量&…
阅读更多...
最新文章

Copyright @ 2022~2023