springboot中rabbitmq同一个消费者类可同时监听多个队列以及可使用方法重载消费队列中不同类型数据

迈不过友情╰ 2023-01-16 14:30 277阅读 0赞

## 1、概述 ##

SpringBoot项目中RabbitMq的使用可参考这篇博客：  
[https://blog.csdn.net/qq\_35387940/article/details/100514134][https_blog.csdn.net_qq_35387940_article_details_100514134]  
写的非常的细致。

**本文着重讲解：**

*  同一个消费者类可同时监听消费**多个队列**
 *  同一个消费者类中可使用方法重载来实现消费队列中**不同类型的数据**

## 2、代码示例 ##

#### 第一步： ####

创建springboot项目，引入依赖，添加rabbitmq相关配置，具体步骤参考文首的博客链接。

#### 第二步： ####

创建rabbitmq配置类：RabbitConfig.java（名字随便起，记得类名上面添加@Configuration注解）

**配置类中创建:**  
**两个队列**：队列A`TestQueueA`，队列B`TestQueueB`  
**两个交换机**：一个直连交换机`TestDirectExchange`、一个扇形交换机`TestFanoutExchange`  
**绑定规则**：直连交换机绑定队列A，扇形交换机绑定队列B  
如图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RBYXJvbjY2Ng_size_16_color_FFFFFF_t_70_pic_center]

package com.tzq.testRabbitMq.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** * @Author : Tangzhiqian * @CreateTime : 2021/4/26 * **/
    
    @Configuration
    public class DirectRabbitConfig { 
    
        //队列A
        @Bean
        public Queue testQueueA() { 
            // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            // return new Queue("TestDirectQueue",true,true,false);
    
            //一般设置一下队列的持久化就好,其余两个就是默认false
            return new Queue("TestQueueA",true);
        }
         //队列B
        @Bean
        public Queue testQueueB() { 
            return new Queue("TestQueueB",true);
        }
    
    
        //Direct交换机 起名：TestDirectExchange
        @Bean
        DirectExchange testDirectExchange() { 
            return new DirectExchange("TestDirectExchange",true,false);
        }
        //扇形交换机 起名：testFanoutExchange
        @Bean
        FanoutExchange testFanoutExchange() { 
            return new FanoutExchange("TestFanoutExchange",true,true);
        }
    
    
        //绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
        @Bean
        Binding bindingDirect() { 
            return BindingBuilder.bind(testQueueA()).to(testDirectExchange()).with("Test");
        }
        @Bean
        Binding bindingTopic() { 
            return BindingBuilder.bind(testQueueB()).to(testFanoutExchange());
        }
    
    
    }

#### 第三步: ####

创建定时任务类：向交换机定时发送数据；  
给`TestDirectExchange`发送String格式数据  
给`TestFanoutExchange`发送Map格式数据

package com.tzq.testRabbitMq.schedule;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    import java.util.HashMap;
    import java.util.Map;
    
    @Component
    public class ScheduleTestRabbitMqSender { 
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        @Scheduled(cron = "0/6 * * * * ?")
        public void SendRabbitmqMessage() { 
    
            Map<String, Object> mapA = new HashMap<>();
            mapA.put("message", "mapA的数据,string格式");
    
            Map<String, Object> mapB = new HashMap<>();
            mapB.put("message", "mapB的数据，map格式");
    		
    		//发送string类型数据
            rabbitTemplate.convertAndSend("TestDirectExchange","Test", mapA.toString());
            //发送map类型数据
            rabbitTemplate.convertAndSend("TestFanoutExchange",null, mapB);
            System.out.println("数据发送成功！！！");
        }
    }

> 定时任务：及记得在启动类上加`@EnableScheduling`  
> 注解

#### 第四步： ####

创建队列消费的类。

*  监听多个队列，在类上添加多个@RabbitListener(queues = $\{QueueName\})
 *  监听多种类型数据，重载被@RabbitHandler注解的方法，参数不同即可。

package com.tzq.testRabbitMq.schedule;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.util.Map;
    
    /** * @Author tangzhiqian * @CreateTime 2021/4/26 9:53 */
    @Component
    @RabbitListener(queues = "TestQueueA")//监听的队列名称 TestQueueA
    @RabbitListener(queues = "TestQueueB")//监听的队列名称 TestQueueB
    public class DirectReceiver { 
    
        @RabbitHandler
        public void process(String message) { 
            System.out.println("消费者收到String类型消息 : " + message);
        }
    
        @RabbitHandler
        public void process(Map message) { 
            System.out.println("消费者收到map类型消息 : " + message);
        }
    }

#### 第五步： ####

运行程序：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RBYXJvbjY2Ng_size_16_color_FFFFFF_t_70]  
消费者同时消费到了两个队列中的数据，不同的数据类型，也可以用不同的方法形参来接收。

> 源码：  
> [https://download.csdn.net/download/TAaron666/18141264][https_download.csdn.net_download_TAaron666_18141264]

[https_blog.csdn.net_qq_35387940_article_details_100514134]: https://blog.csdn.net/qq_35387940/article/details/100514134
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RBYXJvbjY2Ng_size_16_color_FFFFFF_t_70_pic_center]: /images/20221022/646e37ed3c1f4be3ab6d964909f6cee6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RBYXJvbjY2Ng_size_16_color_FFFFFF_t_70]: /images/20221022/314eb7ca61fa4ec698ba96f6e5a43830.png
[https_download.csdn.net_download_TAaron666_18141264]: https://download.csdn.net/download/TAaron666/18141264