springboot simple （12） springboot RabbitMQ

怼烎@ 2024-03-30 13:12 187阅读 0赞

这里首先简单的介绍了RabbitMQ ，然后实现了springboot集成RabbitMQ ，包含两个工程：
1 Producer 生产者
2 Consumer 消费者

1 RabbitMQ 简介

AMQP ：Advanced Message Queue，高级消息队列协议。
RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。

RabbitMQ是一个生产者和消费者模型，用于实现消息的接收、存储、转发。
模型如下图所示：
在这里插入图片描述
Producer 生产者
Consumer 消费者
Exchange 交换机
Queue :队列,存储消息

常用的交换机Exchange
1）fanout Exchange（扇形交换机）
2）direct Exchange（直连交换机）
3）topic Exchange（主题交换机）
4）headers Exchange（头交换机）

2 springboot 集成RabbitMQ

第1步：pom中引入依赖的包：

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4.             <version>3.0.1</version>
  5.         </dependency>

第2步：application.properties设置相关参数：

  1. #rabbitmq相关配置
  2. spring.rabbitmq.host=127.0.0.1
  3. spring.rabbitmq.port=5672
  4. spring.rabbitmq.username=guest
  5. spring.rabbitmq.password=guest
  6. spring.rabbitmq.publisher-confirm-type=correlated
  7. spring.rabbitmq.publisher-returns=true
  8. spring.main.allow-bean-definition-overriding = true

2.1 fanout Exchange（扇形交换机）

特点：将消息路由到所有与该交换机绑定的队列中。
在这里插入图片描述
第1步：创建配置类FanoutConfig，创建了：
交换机（exchange）：fanout_exchange
队列（Queue）：fanout_queue_A
绑定（Binding）：fanoutBindA
队列（Queue）：fanout_queue_B
绑定（Binding）：fanoutBindB

  1. @Configuration
  2. public class FanoutConfig {
  4.     @Bean("fanout_exchange")
  5.     public FanoutExchange fanoutExchange() {
  7.         return ExchangeBuilder.fanoutExchange("fanout_exchange").durable(true).build();
  8.     }
  10.     @Bean("fanout_queue_A")
  11.     public Queue fanoutQueueA(){
  13.         return QueueBuilder.durable("fanout_queue_A").build();
  14.     }
  16.     @Bean
  17.     public Binding fanoutBindA(){
  19.         return BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
  20.     }
  22.     @Bean("fanout_queue_B")
  23.     public Queue fanoutQueueB(){
  25.         return QueueBuilder.durable("fanout_queue_B").build();
  26.     }
  28.     @Bean
  29.     public Binding fanoutBindB(){
  31.         return BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
  32.     }
  33. }

第2步：Producer 项目创建发送消息的类FanoutService ：

  1. @Slf4j
  2. @Service
  3. public class FanoutService {
  5.     String exchange = "fanout_exchange";
  7.     @Resource
  8.     private RabbitTemplate rabbitTemplate;
  10.     public void sendData(String data) {
  12.         log.info("------------fanout producer begin------------");
  13.         rabbitTemplate.convertAndSend(exchange, "", data);
  14.         log.info("exchange:"+exchange);
  15.         log.info("send data:"+data);
  16.         log.info("------------fanout producer end------------");
  17.     }
  18. }

第3步：Consumer 创建接收消息的类FanoutListener ：

  1. @Component
  2. @Slf4j
  3. public class FanoutListener {
  5.     /**
  6.      * .接收队列fanout_queue_A的数据
  7.      */
  8.     @RabbitListener(queues = {
  9.     "fanout_queue_A"})
  10.     @RabbitHandler
  11.     public void receiveDataFromQueueA(Message message, Channel channel) throws IOException {
  13.         long tag = message.getMessageProperties().getDeliveryTag();
  14.         channel.basicAck( tag, true);
  15.         log.info("------------fanout consumer begin------------");
  16.         log.info("queue: fanout_queue_A");
  17.         log.info("receive Data: " + new String(message.getBody()));
  18.         log.info("------------fanout consumer end------------");
  19.     }
  21.     /**
  22.      * .接收队列fanout_queue_B的数据
  23.      */
  24.     @RabbitListener(queues = {
  25.     "fanout_queue_B"})
  26.     @RabbitHandler
  27.     public void receiveDataFromQueueB(Message message, Channel channel) throws IOException {
  29.         long tag = message.getMessageProperties().getDeliveryTag();
  30.         channel.basicAck( tag, true);
  31.         log.info("------------fanout consumer begin------------");
  32.         log.info("queue: fanout_queue_B");
  33.         log.info("receive Data: " + new String(message.getBody()));
  34.         log.info("------------fanout consumer end------------");
  35.     }
  36. }

2.2 direct Exchange（直连交换机）

特点：根据消息默认的路由键（队列名称）将消息投递到相应的队列中。
在这里插入图片描述
第1步：创建配置类DirectConfig，创建了：
交换机（exchange）：direct_exchange_A
队列（Queue）：direct_queue_A
绑定（Binding）：directBindA

交换机（exchange）：direct_exchange_B
队列（Queue）：direct_queue_B
绑定（Binding）：directBindB

  1. @Configuration
  2. public class DirectConfig {
  4.     @Bean("direct_exchange_A")
  5.     public DirectExchange directExchangeA() {
  7.         return ExchangeBuilder.directExchange("direct_exchange_A").durable(true).build();
  8.     }
  10.     @Bean("direct_queue_A")
  11.     public Queue directQueueA() {
  13.         return QueueBuilder.durable("direct_queue_A").build();
  14.     }
  16.     @Bean
  17.     public Binding directBindA(@Qualifier("direct_queue_A")Queue queue,
  18.                               @Qualifier("direct_exchange_A")DirectExchange directExchange){
  20.         return BindingBuilder.bind(queue)
  21.                 .to(directExchange).withQueueName();
  22.     }
  24.     @Bean("direct_exchange_B")
  25.     public DirectExchange directExchangeB() {
  27.         return ExchangeBuilder.directExchange("direct_exchange_B").durable(true).build();
  28.     }
  30.     @Bean("direct_queue_B")
  31.     public Queue directQueueB() {
  33.         return QueueBuilder.durable("direct_queue_B").build();
  34.     }
  36.     @Bean
  37.     public Binding directBindB(@Qualifier("direct_queue_B")Queue queue,
  38.                                @Qualifier("direct_exchange_B")DirectExchange directExchange){
  40.         return BindingBuilder.bind(queue)
  41.                 .to(directExchange).withQueueName();
  42.     }
  43. }

.withQueueName() 可见绑定key为queue名称。

第2步：Producer 项目创建发送消息的类DirectService ：

  1. @Slf4j
  2. @Service
  3. public class DirectService {
  5.     String routeKey = "direct_queue_A";
  7.     @Autowired
  8.     private RabbitTemplate rabbitTemplate;
  10.     public void sendData(String data) {
  12.         CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
  13.         log.info("------------direct producer begin------------");
  14.         rabbitTemplate.convertAndSend(routeKey, data);
  15.         log.info("routeKey:"+routeKey);
  16.         log.info("send data:"+data);
  17.         log.info("------------direct producer end------------");
  18.     }
  19. }

第3步：Consumer 创建接收消息的类DirectListener ：

  1. @Component
  2. @Slf4j
  3. public class DirectListener {
  6.     /**
  7.      * .接收队列direct_queue_A的数据
  8.      */
  9.     @RabbitListener(queues = {
  10.     "direct_queue_A"})
  11.     public void receiveDataFromQueueA(Message message, Channel channel) throws IOException {
  13.         long tag = message.getMessageProperties().getDeliveryTag();
  14.         channel.basicAck(tag, true);
  15.         log.info("------------direct consumer begin------------");
  16.         log.info("queue: fanout_queue_A");
  17.         log.info("receive Data: " + new String(message.getBody()));
  18.         log.info("------------direct consumer end------------");
  19.     }
  21.     /**
  22.      * .接收队列direct_queue_B的数据
  23.      */
  24.     @RabbitListener(queues = {
  25.     "direct_queue_B"})
  26.     public void receiveDataFromQueueB(Message message, Channel channel) throws IOException {
  28.         long tag = message.getMessageProperties().getDeliveryTag();
  29.         channel.basicAck(tag, true);
  30.         log.info("------------direct consumer begin------------");
  31.         log.info("queue: fanout_queue_B");
  32.         log.info("receive Data: " + new String(message.getBody()));
  33.         log.info("------------direct consumer end------------");
  34.     }
  35. }

2.3 topic Exchange（主题交换机）

特点：将消息路由到路由键（RoutingKey）和绑定键（BindKey）相匹配的队列中。
在这里插入图片描述

第1步：创建配置类TopicConfig，创建了：
交换机（exchange）：topic_exchange
队列（Queue）：topic_queue_A
绑定（Binding）：topicBindA
绑定key（BindKey）：project1.station1.*
交换机（exchange）：topic_exchange
队列（Queue）：topic_queue_B
绑定（Binding）：topicBindB
绑定key（BindKey）：project1..“

  1. @Configuration
  2. public class TopicConfig {
  4.     String routeKeyA = "project1.station1.*";
  5.     String routeKeyB = "project1.*.*";
  7.     @Bean("topic_exchange")
  8.     public TopicExchange topicExchange() {
  10.         return ExchangeBuilder.topicExchange("topic_exchange").durable(true).build();
  11.     }
  13.     @Bean("topic_queue_A")
  14.     public Queue topicQueueA(){
  16.         return QueueBuilder.durable("topic_queue_A").build();
  17.     }
  19.     @Bean
  20.     public Binding topicBindA(){
  22.         return BindingBuilder.bind(topicQueueA()).to(topicExchange()).with(routeKeyA);
  23.     }
  25.     @Bean("topic_queue_B")
  26.     public Queue topicQueueB(){
  28.         return QueueBuilder.durable("topic_queue_B").build();
  29.     }
  31.     @Bean
  32.     public Binding topicBindB(){
  34.         return BindingBuilder.bind(topicQueueB()).to(topicExchange()).with(routeKeyB);
  35.     }
  36. }

第2步：Producer 项目创建发送消息的类TopicService ：

  1. @Slf4j
  2. @Service
  3. public class TopicService {
  5.     String exchange = "topic_exchange";
  6.     //String routeKey = "project1.station1.device1";
  7.     String routeKey = "project1.station2.device1";
  8.     @Resource
  9.     private RabbitTemplate rabbitTemplate;
  11.     public void sendData(String data) {
  13.         CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
  14.         log.info("------------topic producer begin------------");
  15.         rabbitTemplate.convertAndSend(exchange, routeKey, data);
  16.         log.info("routeKey:"+routeKey);
  17.         log.info("send data:"+data);
  18.         log.info("------------topic producer end------------");
  19.     }
  20. }

第3步：Consumer 创建接收消息的类TopicListener ：

  1. @Component
  2. @Slf4j
  3. public class TopicListener {
  5.     /**
  6.      * .接收队列topic_queue_A的数据
  7.      */
  8.     @RabbitListener(queues = {
  9.     "topic_queue_A"})
  10.     public void receiveDataFromQueueA(Message message, Channel channel) throws IOException {
  12.         long tag = message.getMessageProperties().getDeliveryTag();
  13.         channel.basicAck( tag, true);
  14.         log.info("------------topic_ consumer begin------------");
  15.         log.info("queue: topic_queue_A");
  16.         log.info("receive Data: " + new String(message.getBody()));
  17.         log.info("------------topic_ consumer end------------");
  18.     }
  20.     /**
  21.      * .接收队列topic_queue_B的数据
  22.      */
  23.     @RabbitListener(queues = {
  24.     "topic_queue_B"})
  25.     public void receiveDataFromQueueB(Message message, Channel channel) throws IOException {
  27.         long tag = message.getMessageProperties().getDeliveryTag();
  28.         channel.basicAck( tag, true);
  29.         log.info("------------topic_ consumer begin------------");
  30.         log.info("queue: topic_queue_B");
  31.         log.info("receive Data: " + new String(message.getBody()));
  32.         log.info("------------topic_ consumer end------------");
  33.     }
  34. }

2.4 headers Exchange（头交换机）

特点：路由规则是建立在头属性值之上，而不是路由键。
在这里插入图片描述

第1步：创建配置类TopicConfig，创建了：
交换机（exchange）：headers_exchange
队列（Queue）：headers_queue_A
绑定（Binding）：headersBindA
头（headers）：
headerValues.put(“type”, “gass”);
headerValues.put(“data”, “realTime”);
匹配规则：whereAll

队列（Queue）：headers_queue_B
绑定（Binding）：headersBindB
头（headers）：
headerValues.put(“type”, “gass”);
headerValues.put(“data”, “realTime”);
匹配规则：whereAny

  1. @Configuration
  2. public class HeadersConfig {
  5.     @Bean("headers_exchange")
  6.     public HeadersExchange headersExchange() {
  8.         return ExchangeBuilder.headersExchange("headers_exchange").durable(true).build();
  9.     }
  11.     @Bean("headers_queue_A")
  12.     public Queue headerQueueA(){
  14.         return QueueBuilder.durable("headers_queue_A").build();
  15.     }
  17.     @Bean
  18.     public Binding headersBindA(){
  20.         Map<String,Object> headerValues = new HashMap<>();
  21.         headerValues.put("type", "gass");
  22.         headerValues.put("data", "realTime");
  23.         return BindingBuilder.bind(headerQueueA()).to(headersExchange()).whereAll(headerValues).match();
  24.     }
  26.     @Bean("headers_queue_B")
  27.     public Queue headerQueueB(){
  29.         return QueueBuilder.durable("headers_queue_B").build();
  30.     }
  32.     @Bean
  33.     public Binding headersBindB(){
  35.         Map<String,Object> headerValues = new HashMap<>();
  36.         headerValues.put("type", "gass");
  37.         headerValues.put("data", "realTime");
  38.         return BindingBuilder.bind(headerQueueB()).to(headersExchange()).whereAny(headerValues).match();
  39.     }
  40. }

第2步：Producer 项目创建发送消息的类HeadersService ：

  1. @Slf4j
  2. @Service
  3. public class HeadersService {
  5.     String exchange = "headers_exchange";
  6.     String routeKey = "";
  8.     @Resource
  9.     private RabbitTemplate rabbitTemplate;
  11.     public void sendData(String data) {
  13.         log.info("------------headers producer begin------------");
  14.         MessageProperties messageProperties = new MessageProperties();
  15.         messageProperties.setHeader("type", "gass");
  16.         messageProperties.setHeader("data", "realTime");
  17.         //messageProperties.setHeader("data", "history");
  18.         Message message = new Message(data.getBytes(), messageProperties);
  19.         rabbitTemplate.convertAndSend(exchange, routeKey, message);
  20.         log.info("send data:"+data);
  21.         log.info("------------headers producer end------------");
  22.     }
  23. }

第3步：Consumer 创建接收消息的类HeadersListener ：

  1. @Component
  2. @Slf4j
  3. public class HeadersListener {
  5.     /**
  6.      * .接收队列headers_queue_A的数据
  7.      */
  8.     @RabbitListener(queues = {
  9.     "headers_queue_A"})
  10.     public void receiveDataFromQueueA(Message message, Channel channel) throws IOException {
  12.         long tag = message.getMessageProperties().getDeliveryTag();
  13.         channel.basicAck( tag, true);
  14.         log.info("------------headers consumer begin------------");
  15.         log.info("queue: headers_queue_A");
  16.         log.info("receive Data: " + new String(message.getBody()));
  17.         log.info("------------headers consumer begin------------");
  18.     }
  20.     /**
  21.      * .接收队列headers_queue_B的数据
  22.      */
  23.     @RabbitListener(queues = {
  24.     "headers_queue_B"})
  25.     public void receiveDataFromQueueB(Message message, Channel channel) throws IOException {
  27.         long tag = message.getMessageProperties().getDeliveryTag();
  28.         channel.basicAck( tag, true);
  29.         log.info("------------headers consumer begin------------");
  30.         log.info("queue: headers_queue_B");
  31.         log.info("receive Data: " + new String(message.getBody()));
  32.         log.info("------------headers consumer begin------------");
  33.     }
  34. }

3 测试验证

3.1 fanout Exchange验证

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByFanout?msg=hello,fanout exchangge!
在这里插入图片描述
第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以看到consumer接收消息：
在这里插入图片描述
可见：与交换机fanout_exchange绑定的两个队列fanout_queue_A、fanout_queue_B都收到了消息。

3.2 direct Exchange验证

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByDirect?msg=hello,direct exchangge!
在这里插入图片描述
第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以看到consumer接收消息：
在这里插入图片描述
可见：routeKey 和BindKey都为queue名称direct_queue_A，队列direct_queue_A收到了消息。

3.3 topic Exchange验证

3.3.1 topic_queue_A、topic_queue_B都匹配

路由键（routeKey）：project1.station1.device1
绑定key（BindKey）：project1.station1.*
绑定key（BindKey）：project1..“

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByTopic?msg=hello,topic exchangge!
在这里插入图片描述
第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以导刊consumer接收消息：
在这里插入图片描述
可以看到两个队列 topic_queue_A、topic_queue_B都收到了消息。

3.3.2 只有topic_queue_B匹配

路由键（routeKey）：project1.station2.device1
绑定key（BindKey）：project1.station1.*
绑定key（BindKey）：project1..“

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByTopic?msg=hello,topic exchangge!
在这里插入图片描述
第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以看到consumer接收消息：
在这里插入图片描述
可以看到只有队列topic_queue_B收到了消息。

3.4 headers Exchange验证

3.4.1 headers_queue_A、headers_queue_B都匹配

发送消息的header：

  1. messageProperties.setHeader("type", "gass");
  2. messageProperties.setHeader("data", "realTime");

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByHeaders?msg=hello,headers exchangge!
在这里插入图片描述
第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以看到consumer接收消息：
在这里插入图片描述
可见headers_queue_A、headers_queue_B都收到了消息。

3.4.2 只有headers_queue_B匹配

发送消息的header：

  1. messageProperties.setHeader("type", "gass");
  2. messageProperties.setHeader("data", "history");

第1步：postman发送请求：
http://127.0.0.1:8081/rabbitMq/sendMessageByHeaders?msg=hello,headers exchangge!
在这里插入图片描述

第2步：可以看到producer发送消息：
在这里插入图片描述
第3步：可以看到consumer接收消息：
在这里插入图片描述
可见：只有headers_queue_B收到了消息。

代码详见：
https://gitee.com/linghufeixia/springboot-simple
chapter6-4 producer项目
chapter6-5 consumer项目

教程列表：
springboot simple（0） springboot简介
springboot simple（1） springboot Helloworld
springboot simple（2） springboot Starter
springboot simple（3 ）springboot Web开发
springboot simple（4）springboot 数据持久化
springboot simple （5） springboot Nosql
springboot simple （6） springboot mqtt
springboot simple （7） springboot thrift
springboot simple （8） springboot kafka
springboot simple （9） springboot jpa（Hibernate）
springboot simple （10） springboot protobuf
springboot simple （11） springboot protostuff
springboot simple （12） springboot RabbitMQ

发表评论

表情：
评论列表 （有 0 条评论，187人围观）

还没有评论，来说两句吧...

相关阅读