SpringBoot——整合RabbitMQ收发消息

目录
RabbitMQ消息队列 
项目总结
新建一个SpringBoot项目
pom.xml
application.properties配置文件
index.html前端页面
RabbitMQConfig配置类
RabbitMQProducer生产者
RabbitMQConsumer消费者
IndexController控制器
SpringbootRabbitmqApplication启动类
 测试

---

RabbitMQ消息队列 

   

• 消息中间件是位于两个或多个应用程序之间的中介软件，它允许应用程序相互通讯和数据交换，而无需直接连接和交互。
  
• RabbitMQ 作为消息中间件，提供了这种消息传递的机制和基础架构，使得差别系统之间能够通过异步消息队列举行有效通讯。 
  
• 它是一个基于AMQP（Advanced Message Queuing Protocol，高级消息队列协议）协议的消息代理，提供了跨应用和跨服务的异步通讯本领。
  
• RabbitMQ的核心概念包括：
  
  
  
  • 消息（Message）：数据的最小单位，包罗了消息头和消息体。
    
  • 队列（Queue）：用于存储消息，等待被消费者吸收和处理。
    
  • 交换器（Exchange）：负责吸收消息并将其路由到合适的队列，基于预界说的规则。
    
    
    
    • 直接交换（Direct Exchange）：基于路由键的精确匹配。
      
    • 主题交换（Topic Exchange）：基于路由键的模式匹配。
      
    • 扇形交换（Fanout Exchange）：广播消息到全部绑定的队列。
      
    • 首选交换（Headers Exchange）：基于消息头举行匹配。（headers交换机和direct交换机完全一致，但性能差很多，目前几乎用不到了）
      
    
    
  • 绑定（Binding）：界说交换器与队列之间的关系，确定消息的路由规则。
    
  • 路由键（Routing Key）：用于在交换器和队列之间路由消息。
    
  • 消费者（Consumer）：从队列中吸收和处理消息的应用程序或服务。
    
  • 发布者（Producer）：向交换器发送消息的应用程序或服务。
    
  
  
• RabbitMQ广泛用于构建分布式系统、微服务架构和事件驱动架构，以实现异步通讯、负载分担、数据解耦和高可用性。
  

  项目总结

   

• 添加依赖：在Spring Boot项目的pom.xml文件中添加RabbitMQ的依赖
  
• 配置RabbitMQ连接信息：在application.properties或application.yml文件中配置RabbitMQ的连接信息，包括主机名、端口号、用户名、密码等信息。
  
• 创建配置类：
  
  
  
  • 使用@Bean注解界说Exchange和Queue，并绑定它们的关系，以确保消息能够被正确路由和传递。
    
  • 使用@Bean注入CachingConnectionFactory缓存 连接工厂对象，用于开启RabbitMQ的消息发送确认模式
    
  
  
• 创建生产者：编写一个发送消息的生产者（Producer），可以是一个服务类或控制器方法。使用RabbitTemplate来发送消息到RabbitMQ的Exchange。
  
• 创建消费者：编写一个吸收消息的消费者（Consumer），使用@RabbitListener监听指定的Queue，并处理吸收到的消息，比如打印在控制台上。
  
• 启动项目和RabbitMQ服务后，项目的工作流程：
  
  
  
  • 打开欣赏器访问页面，在文本域填写要发送信息，点击“发送”，
    
  • 后端IndexController控制器吸收到用户提交的消息后，由生产者将消息发送给RabbitMQ的交换机，RabbitMQ再将消息根据路由键路由到队列中，而且将结果反馈给服务器，服务器将打印”消息发送乐成“的日记
    
  • 消费者中监听此队列就会立刻收到并处理吸收到的消息，打印到控制台上
    
  
  

  自己从填写要发送的信息的地方开始分析，顺藤摸瓜，将项目的几个文件串连起来
  新建一个SpringBoot项目

项目结构：

pom.xml

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3.         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4.         <modelVersion>4.0.0</modelVersion>
5.         <parent>
6.                 <groupId>org.springframework.boot</groupId>
7.                 <artifactId>spring-boot-starter-parent</artifactId>
8.                 <version>2.3.12.RELEASE</version>
9.                 <relativePath/> <!-- lookup parent from repository -->
10.         </parent>
11.         <groupId>com.study</groupId>
12.         <artifactId>springboot_rabbitmq</artifactId>
13.         <version>0.0.1-SNAPSHOT</version>
14.         <name>springboot_rabbitmq</name>
15.         <description>Demo project for Spring Boot</description>
16.         <properties>
17.                 <java.version>8</java.version>
18.         </properties>
19.         <dependencies>
20.                 <dependency>
21.                         <groupId>org.springframework.boot</groupId>
22.                         <artifactId>spring-boot-starter-amqp</artifactId>
23.                 </dependency>
24.                 <dependency>
25.                         <groupId>org.springframework.boot</groupId>
26.                         <artifactId>spring-boot-starter-thymeleaf</artifactId>
27.                 </dependency>
28.                 <dependency>
29.                         <groupId>org.springframework.boot</groupId>
30.                         <artifactId>spring-boot-starter-web</artifactId>
31.                 </dependency>
33.                 <dependency>
34.                         <groupId>org.springframework.boot</groupId>
35.                         <artifactId>spring-boot-starter-test</artifactId>
36.                         <scope>test</scope>
37.                 </dependency>
38.                 <dependency>
39.                         <groupId>org.springframework.amqp</groupId>
40.                         <artifactId>spring-rabbit-test</artifactId>
41.                         <scope>test</scope>
42.                 </dependency>
43.         </dependencies>
45.         <build>
46.                 <plugins>
47.                         <plugin>
48.                                 <groupId>org.springframework.boot</groupId>
49.                                 <artifactId>spring-boot-maven-plugin</artifactId>
50.                         </plugin>
51.                 </plugins>
52.         </build>
54. </project>

复制代码

application.properties配置文件

1. spring.rabbitmq.host=192.168.40.128
2. spring.rabbitmq.port=5672
3. spring.rabbitmq.username=admin
4. spring.rabbitmq.password=123456
6. # 指定队列,交换机,路由键的名称
7. #队列用于存储消息，生产者发送消息到队列，消费者从队列接收消息进行处理。
8. rabbit.queue.name=springboot.queue.test
9. #交换机负责将消息路由到一个或多个队列中，根据指定的路由键来确定消息的路由规则。
10. rabbit.exchange.name=springboot.exchange.test
11. #在消息发送时，会指定消息的路由键，交换机根据这个路由键来决定将消息路由到哪些队列中。
12. rabbit.routing.key=springboot.routingkey.test

复制代码

index.html前端页面

1. <!DOCTYPE html>
2. <html lang="en">
3. <head>
4.     <meta charset="UTF-8">
5.     <title>Title</title>
6. </head>
7. <body>
8.     <!--发送后,会触发IndexController的sendMessage方法-->
9.     <form action="/sendMessage" method="post">
10.         <!--在文本域填写要发送的消息-->
11.         <textarea rows="4" cols="40" name="message"></textarea>
12.         <!--点击"发送"按钮-->
13.         <br><input type="submit" value="发送"/>
14.     </form>
16. </body>
17. </html>

复制代码

RabbitMQConfig配置类

1. package com.study.springboot_rabbitmq.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.DirectExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.beans.factory.annotation.Value;
8. import org.springframework.context.annotation.Bean;
9. import org.springframework.context.annotation.Configuration;
11. /**
12. * 配置类: 绑定队列和交换器
13. */
14. @Configuration
15. public class RabbitMQConfig {
17.     @Value("${rabbit.queue.name}")
18.     String queueName;
20.     @Value("${rabbit.exchange.name}")
21.     String exchangeName;
23.     @Value("${rabbit.routing.key}")
24.     String routingKey;
26.     @Bean()
27.     public Queue initQueue(){//创建队列
28.         return new Queue(queueName);
29.     }
31.     @Bean()
32.     public DirectExchange initDirectExchange(){//创建交换器
33.         return new DirectExchange(exchangeName);
34.     }
36.     @Bean
37.     public Binding bindingDirect(){//将队列与交换器绑定
38.         return BindingBuilder.bind(initQueue()).to(initDirectExchange()).with(routingKey);
39.     }
40. }

复制代码

RabbitMQProducer生产者

1. package com.study.springboot_rabbitmq.service;
3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.beans.factory.annotation.Value;
6. import org.springframework.stereotype.Service;
8. /**
9. * 消息生产者
10. */
11. @Service
12. public class RabbitMQProducer {
14.     @Autowired
15.     RabbitTemplate rabbitTemplate;
17.     //读取配置文件中的交换器名称和路由键名称
18.     @Value("${rabbit.exchange.name}")
19.     String exchangeName;
21.     @Value("${rabbit.routing.key}")
22.     String routingKey;
24.     //将消息发送给RabbitMQ的交换机,交换机通过路由键决定将消息路由到哪些队列
25.     public void send(String message){
26.         rabbitTemplate.convertAndSend(exchangeName,routingKey,message);
27.     }
28. }

复制代码

RabbitMQConsumer消费者

1. package com.study.springboot_rabbitmq.service;
3. import org.slf4j.Logger;
4. import org.slf4j.LoggerFactory;
5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
6. import org.springframework.stereotype.Service;
8. /**
9. * 消息消费者
10. */
11. @Service
12. public class RabbitMQConsumer {
14.     private static final Logger log= LoggerFactory.getLogger(RabbitMQConsumer.class);
16.     //使用@RabbitListener监听配置文件中的队列,当收到消息后,将其打印在控制台上
17.     @RabbitListener(queues = "${rabbit.queue.name}")
18.     public void getMessage(String message){
19.         log.info("消费者收到消息:{}",message);
20.     }
21. }

复制代码

IndexController控制器

   

• 发送确认模式：在生产者向RabbitMQ发送消息后，RabbitMQ可以给生产者一个反馈消息，这个反馈消息中会包罗吸收是否乐成、失败原因等一系列内容
  

1. package com.study.springboot_rabbitmq.config;
3. import org.slf4j.Logger;
4. import org.slf4j.LoggerFactory;
5. import org.springframework.amqp.core.*;
6. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
7. import org.springframework.amqp.rabbit.connection.CorrelationData;
8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
9. import org.springframework.beans.factory.annotation.Autowired;
10. import org.springframework.beans.factory.annotation.Value;
11. import org.springframework.cache.annotation.Caching;
12. import org.springframework.context.annotation.Bean;
13. import org.springframework.context.annotation.Configuration;
15. /**
16. * 配置类: 绑定队列和交换器
17. */
18. @Configuration
19. public class RabbitMQConfig {
21.     private static final Logger log= LoggerFactory.getLogger(RabbitMQConfig.class);
23.     @Autowired
24.     private CachingConnectionFactory connectionFactory;
26.     @Value("${rabbit.queue.name}")
27.     String queueName;
29.     @Value("${rabbit.exchange.name}")
30.     String exchangeName;
32.     @Value("${rabbit.routing.key}")
33.     String routingKey;
35.     @Bean()
36.     public Queue initQueue(){//创建队列
37.         return new Queue(queueName);
38.     }
40.     @Bean()
41.     public DirectExchange initDirectExchange(){//创建交换器
42.         return new DirectExchange(exchangeName);
43.     }
45.     @Bean
46.     public Binding bindingDirect(){//将队列与交换器绑定
47.         return BindingBuilder.bind(initQueue()).to(initDirectExchange()).with(routingKey);
48.     }
50.     //RabbitMQ收到消息后,把结果反馈给服务器,服务器将打印日志
51.     @Bean
52.     public RabbitTemplate rabbitTemplate(){
53.         //消息发送成功后触发确认方法
54.         connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
55.         //消息发送失败后触发回调方法
56.         connectionFactory.setPublisherReturns(true);
57.         //通过连接工厂对象创建RabbitTemplate对象
58.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
59.         //若交换器无法匹配到指定队列,则取消发送消息
60.         rabbitTemplate.setMandatory(true);
61.         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
62.             @Override
63.             public void confirm(CorrelationData correlationData, boolean ack, String cause) {//ack:RabbitMQ返回的应答
64.                 if(ack){
65.                     log.info("消息发送成功");
66.                 }else{
67.                     log.info("消息发送失败,原因: {}",cause);
68.                 }
69.             }
70.         });
71.         rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
72.             @Override
73.             public void returnedMessage(Message message, int i, String s, String s1, String s2) {
74.                 log.info("消息发送失败: {}",message);
76.             }
77.         });
78.         return rabbitTemplate;
79.     }
80. }

复制代码

SpringbootRabbitmqApplication启动类

1. package com.study.springboot_rabbitmq;
3. import org.springframework.boot.SpringApplication;
4. import org.springframework.boot.autoconfigure.SpringBootApplication;
6. @SpringBootApplication
7. public class SpringbootRabbitmqApplication {
9.         public static void main(String[] args) {
10.                 SpringApplication.run(SpringbootRabbitmqApplication.class, args);
11.         }
13. }

复制代码

 测试

开启RabbitMQ服务，启动项目
   从Spring Boot项目的角度来看，无论RabbitMQ运行在Windows还是Linux上，整合RabbitMQ的方式都是基本一致的，只要你的Spring Boot应用程序连接到RabbitMQ服务器，并使用RabbitMQ客户端库举行通讯即可
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。