创建SpringBoot和RabbitMQ的整合项目

ops/2024/9/22 16:46:14/

文章目录

  • 创建SpringBoot和RabbitMQ的整合项目
    • 首先快速创建一个maven项目
    • 引入SpringBoot整合rabbitMQ的依赖
    • 在src/main目录下创建resources目录并引入配置文件
    • 写消息发送者MessageSender
    • 写消息接收者MessageReceiver
    • 写RabbitMQConfig配置类
    • 写SpringBoot启动主类
    • CommandLineRunner接口的作用
    • 启动程序 查看效果
    • 注意要点
      • 15672和5672的区别
      • 消息发送的队列怎么找
    • 我们的消息发送者和消息接收者可以只开启一个
      • 只开启消息发送者先存一些消息到消息队列
      • 只开启消息接收者从消息队列里面读取消息
    • 接下来我们来看一下RabbitMQ中怎么控制一个线程一次消费的消息数,主要是看下concurrency关键属性

创建SpringBoot和RabbitMQ的整合项目

首先快速创建一个maven项目

如下图:
在这里插入图片描述

引入SpringBoot整合rabbitMQ的依赖

如下图:
在这里插入图片描述
这里要SpringBoot和RabbitMQ整合的依赖对应的就是spring-boot-starter-amqp,那么amqp是什么意思呢?是Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,是面向消息的中间件设计。比如RabbitMQ消息中间件使用的就是这个协议。
因为我们要启动服务,因此需要有tomcat服务器,因此这里还要引入spring-boot-starter-web模块依赖,因为SpringBoot自带的tomcat服务器在web模块里面。

<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 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>rabbitMQ</artifactId><version>1.0-SNAPSHOT</version><packaging>jar</packaging><name>rabbitMQ</name><url>http://maven.apache.org</url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.7</version></parent><dependencies><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>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency></dependencies>
</project>

在src/main目录下创建resources目录并引入配置文件

如下图:
在这里插入图片描述

写消息发送者MessageSender

如下图:
在这里插入图片描述

/*** @author xuan* @create 2024/4/25*/
@Component
public class MessageSender {@Autowiredprivate AmqpTemplate rabbitTemplate;public void sendMessage(String exchange, String routingKey, Object message) {rabbitTemplate.convertAndSend(exchange, routingKey, message);}
}

写消息接收者MessageReceiver

如下图:
在这里插入图片描述

/*** @author xuan* @create 2024/4/25*/
@Component
public class MessageReceiver {@RabbitListener(queues = "test.queue")public void handleMessage(Object message) {System.out.println("Received message: " + message);}
}

写RabbitMQConfig配置类

如下图:
在这里插入图片描述

/*** @author xuan* @create 2024/4/25*/
@Configuration
public class RabbitMQConfig {@Beanpublic DirectExchange testExchange() {return new DirectExchange("test.exchange");}@Beanpublic Queue testQueue() {return new Queue("test.queue");}@Beanpublic Binding testBinding() {return BindingBuilder.bind(testQueue()).to(testExchange()).with("test.routingKey");}
}

写SpringBoot启动主类

如下图:
在这里插入图片描述

/*** @author xuan* @create 2024/4/25*/
@SpringBootApplication
public class Application implements CommandLineRunner {@Autowiredprivate MessageSender messageSender;public static void main(String[] args) {SpringApplication.run(Application.class, args);}@Overridepublic void run(String... args) throws Exception {messageSender.sendMessage("test.exchange", "test.routingKey", "Hello, RabbitMQ!");}
}

CommandLineRunner接口的作用

从上面的SpringBoot启动类可以看到,它实现了CommandLineRunner接口,实现了这个接口中的抽象方法run,那么CommandLineRunner接口又什么作用呢?它主要是用在SpringBoot的启动主类上的,主要作用是在SpringBoot应用程序启动之后执行一些任务,比如:

  • 数据初始化:加载初始数据到数据库或其他存储系统中。
  • 发送通知:向用户或系统发送启动通知,例如通过邮件或消息队列。
  • 执行检查:进行应用程序启动前的检查,如数据库连接检查、文件系统检查等。

比如我们上面的主类在成功启动SpringBoot的应用程序之后 就会给RabbitMQ消息队列发送一个消息。

启动程序 查看效果

启动主类之后 控制台输出如下图:
在这里插入图片描述
这里我们主要看消息体Body的内容,可以发现是Hello RabbitMQ。后面是消息的一些属性 类似如我们的http协议中的request请求头的信息,包括消息的内容类型了,消息的内容编码了,消息的长度了等信息。

可以发现我们这里测试成功了。

注意要点

15672和5672的区别

RabbitMQ前端管理界面使用的端口号是15672,比如我们启动rabbitmq之后,在浏览器中输入127.0.0.1:15672可以进入rabbitmq的前端页面管理器中进行相关操作,如下图:
在这里插入图片描述
但是后端服务器的rabbitmq在电脑上启动的端口号是5672,如下图:
在这里插入图片描述
这两个端口号特别容易搞错,千万要注意,如果配置文件里的端口号写15672,那么我们的idea里面的java程序就连接不上我们本地已经启动的rabbitmq服务器了。

消息发送的队列怎么找

通过RabbitMQ中的交换机Exchange,和路由RouteingKey可以唯一的定位到一个消息队列Queue,如下图:
在这里插入图片描述
在这里插入图片描述

因此我们代码里面发送者发送消息的时候,一定要指定使用的交换机和使用的路由,但不用指定使用的队列,因为有交换机和路由之后就会自动的映射到使用的队列了,看下代码如下图:
在这里插入图片描述
但是我们需要通过一个配置类配置一下 需要告诉java程序,如下图:
在这里插入图片描述
然后接收者接收消息的时候 可以直接指定test.queue队列,就可以自动的去这个队列中取出消息了。不用写交换机和路由的具体信息。下面看下接收者的处理 如下图:
在这里插入图片描述

但是我觉得RabbitMQ应该是会有线程安全问题的,比如说因为目前看来同一个队列 两个不同的线程都是可以访问的,那这样的话有共享资源问题 就肯定会出现线程安全问题。

我们的消息发送者和消息接收者可以只开启一个

只开启消息发送者先存一些消息到消息队列

比如我们可以先用消息发送者往RabbitMQ的消息队列test.queue里面发送三条消息,此时注意关闭消息接收者,以免接收者直接读取队列里面的消息,我们在页面上就看不到效果了,如下图:
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

只开启消息接收者从消息队列里面读取消息

接下来我们把消息发送者关了,把消息接收者打开,去消费队列里面的消息,如下图:
在这里插入图片描述
在这里插入图片描述
看下启动应用程序之后的效果 如下图:
在这里插入图片描述

接下来我们来看一下RabbitMQ中怎么控制一个线程一次消费的消息数,主要是看下concurrency关键属性

通过concurrency属性来控制线程从队列中一次性读取的消息个数。如果我们把属性concurrency的值设置为1 那么就表示一个线程一次性会从队列里面读取一条消息;如果把这个值设置成3 那么就表示一个线程一次性会从队列里面读取3条消息;concurrency的书写位置如下图:
在这里插入图片描述
据我的观察,RabbitMQ内部应该有线程安全机制,就是在当前线程读取消费队列消息没有完成的时候,另外的线程是不能消费队列消息的。

现在我们模拟两个线程,而我们具体的模拟方式其实是重新复制一个项目,然后改下端口号为8081,接着我们把给队列发送消息的代码注释掉 只开启接收者从消息队列里面读取消息 过程如下图:
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

上面是我们模拟的第一个线程,其实也就是copy一个相同的项目 改下端口号 启动的时候就是不同的线程了。

然后启动第二个线程,发送6条消息到消息队列中,如下图:
在这里插入图片描述
那么这个时候呢?线程一就会先读取消息队列的第一条消息,线程二此时会在线程一后面排队,当线程一处理完消息之后,线程二会去读取消息队列里面的消息,然后线程一读取…。因此正常情况的输出是 其中一个应用程序中 输出的是编号1,3,5的消息,另外一个应用程序中输出的是编号2,4,6的消息。如下图:
在这里插入图片描述
在这里插入图片描述
确实是正常的。

现在我们把concurrency线程一次性最多读取的消息个数改成4,看下控制台输出结果,如下图:
在这里插入图片描述
在这里插入图片描述
可以看到第一个线程一次性会读取消息队列的前四条消息,而第二个线程本来也会读取四条 但是队列中只剩两条消息,因此第二个线程就一次性的读取两条消息了。
但是有个线程安全问题。就是你会发现第一个线程虽然读取的确实是队列里面的前四条消息,但是呢?消费消息的时候却不是按队列里面添加消息的顺序,这可能就会出现问题了。比如假如说队列里面按照先后顺序添加了两条消息,1穿装,2脱装。但是线程读取消息之后 先消费处理的可能是2脱装,那是不是就会出现问题了?因为你玩家身上现在根本就没有装备 你要怎么脱呢?因此当concurrency设置的值超过1的时候都可能会出现消息消费顺序有误的问题。

因此我们如果想要保证消费的消息严格的按照队列里面添加的消息顺序来执行,那么必须要把concurrency设置成1


http://www.ppmy.cn/ops/26818.html

相关文章

git解决冲突问题

冲突问题&#xff1a;开发者的俩个分支都修改了同一个文件的同一部分时, Git 无法自动决定哪个版本是正确的, 因此会产生冲突。 解决冲突&#xff1a; 我们git在合并的过程中指出冲突文件&#xff0c;通过git status命令查看那个文件存在冲突 解决方法&#xff1a; 1&#xff0…

【嵌入式笔试题】网络编程笔试题

非常经典的笔试题。 2.网络编程(29道) 2.1列举一下OSI协议的各种分层。说说你最熟悉的一层协议的功能。 ( 1 )七层划分为:应用层、表示层、会话层、传输层、网络层、数据链路层、物理 层。 ( 2 )五层划分为:应用层、传输层、网络层、数据链路层、物理层。 ( 3 )…

基于Springboot的web后端开发三层架构上手实操

引言 我们拿到了xml文件 我们要将将xml文件中的数据加载并解析 完成数据的处理 并且返回给前端页面(result格式) 1.将xml文件放在resources目录下 xml是我们需要解析的文件 查看xml文件 2.在springboot中引入dom4j依赖 解析xml需要在springboot中引入dom4j依赖 这边我们…

使用 LlamaIndex 和 Llama 2-Chat 构建知识驱动的对话应用程序

文章目录 使用 LlamaIndex 和 Llama 2-Chat 构建知识驱动的对话应用程序Llama 2-70B-聊天LlamaIndex 解决方案概述先决条件使用 SageMaker JumpStart 部署 GPT-J 嵌入模型使用 SageMaker Python SDK 进行部署在 SageMaker Studio 中使用 SageMaker JumpStart 进行部署使用 Sage…

【C语言】动态内存分配

即使行动导致错误&#xff0c;却也带来了学习与成长;不行动则是停滞与萎缩。&#x1f493;&#x1f493;&#x1f493; 目录 •&#x1f319;知识回顾 &#x1f34b;知识点一&#xff1a;为什么要有动态内存分配 &#x1f34b;知识点二&#xff1a;malloc和free • &#x1…

ElasticSearch教程入门到精通——第二部分(基于ELK技术栈elasticsearch 7.x新特性)

ElasticSearch教程入门到精通——第二部分&#xff08;基于ELK技术栈elasticsearch 7.x新特性&#xff09; 1. JavaAPI-环境准备1.1 新建Maven工程——添加依赖1.2 HelloElasticsearch 2. 索引2.1 索引——创建2.2 索引——查询2.3 索引——删除 3. 文档3.1 文档——重构3.2 文…

Java_从入门到JavaEE_06

一、方法 1.静态方法 理解&#xff1a;特定功能的代码块 好处&#xff1a;解决代码的冗余 语法结构&#xff1a; 访问修饰符 static 返回值类型 方法名([参数列表]){ ​ …代码块… } 分类&#xff1a; 无参数无返回值的方法 带参数的方法 带返回值的方法 2.无参数无返回值…

【已解决】pandas读excel中长数字变成科学计数法的问题

pandas 读excel中的长数字时&#xff0c;即使excel中已经设置为文本&#xff0c;读进df后也会自动变成科学计数法。 在日常的数据分析和处理工作中&#xff0c;Excel和pandas是数据分析师们不可或缺的得力助手。然而&#xff0c;在使用pandas读取Excel文件时&#xff0c;我们有…