【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-1RabbitMQ扩展) 
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控制界面详解

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

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

相关文章

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

场景 开发一个网站或者服务&#xff0c;需要在本地测试时&#xff0c;可以将线上的域名指向本地开发环境的IP地址。从而模拟真实环境中的域名访问&#xff0c;方便调试和开发。 步骤 1、以管理员身份打开命令提示符 2、编辑hosts文件&#xff1a; 输入以下命令打开hosts文…

【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篇&#xff0c;该系列文章适合具有一定PB基础的读者。 通过一个个由浅入深的编程实战案例学习&#xff0c;提高编程技巧&#xff0c;以保证小伙伴们能应付公司的各种开发需求。 文章中设计到的源码&#xff0c;小凡都上传到了gite…

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

在Web前端技术中&#xff0c;直接控制浏览器窗口或标签页从主屏投屏到副屏&#xff08;如PPT的演讲者模式&#xff09;并不简单&#xff0c;而且直接控制浏览器窗口从主屏投屏到副屏的功能超出了Web标准的范畴&#xff0c;并且涉及到用户系统级别的设置和权限&#xff0c;因此不…

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

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

(一)Docker基本介绍

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

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

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

Omni3D目标检测

Omni3D是一个针对现实场景中的3D目标检测而构建的大型基准和模型体系。该项目旨在推动从单一图像中识别3D场景和物体的能力&#xff0c;这对于计算机视觉领域而言是一个长期的研究目标&#xff0c;并且在机器人、增强现实&#xff08;AR&#xff09;、虚拟现实&#xff08;VR&a…

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

基于函数报错的方式来进行利用-字符型&#xff08;本页updatexml()&#xff09; 前提条件是后台数据库没有屏蔽数据库语法报错信息 updatexml()方法详解 注释&#xff1a; 第一个参数&#xff0c;意思就是xml文档的名称 第二个参数&#xff0c;意思就是定位到xml文档中指定…

PriorityQueue底层你了解多少?(带你彻底掌握优先级队列)

1. 概念 队列是一种先进先出(FIFO)的数据结构&#xff0c;但有些情况下&#xff0c;操作的数据可能带有优先级&#xff0c;一般出队列时&#xff0c;可能需要优先级高的元素先出队列&#xff0c;该中场景下&#xff0c;使用队列显然不合适&#xff0c;比如:在手机上玩游戏的时…

[图解]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)

题意&#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的基础&#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实现滑动式标签布局

大家好&#xff0c;我是网创有方&#xff0c;今天记录下标签布局的实现方法&#xff0c;先看下效果图。 第一步&#xff1a;编写一个activity或者fragment。内含有一个viewpager2的适配器&#xff0c;适配器类型为FragmentStateAdapter。 ​ public class MediaCreateFragment…

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

1.定义 所谓BRDF&#xff08;Bidirectional Reflectance Distribution Function&#xff0c;双向反射分布函数&#xff09;&#xff0c;指的是从辐射度量学的角度去理解光线的反射&#xff0c;如下图所示。 所谓反射就是一个点从ωi方向发出的Radiance转化为dA接收到的功率E&am…

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

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

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 中如何使用分包、代码压缩提升应用执行性能。除此之外&#xff0c;还有不少普适、细碎的方法&#xff0c;能够有效降低应用体积&#xff0c;提升网络分发性能&#xff0c;包括&#xff1a; 使用动态加载&#xff0c;减少首屏资源加载量&…