springboot整合rabbitmq（一）

痛定思痛。 2023-10-04 20:01 7阅读 0赞

最近看到一篇文章，讲的是mqtt在物联网的使用，在测试使用时，选择的是rabbitmq的mqtt协议，因此测试使用完mqtt后，闲来无事再整合一下rabbitmq，mqtt可以看如下文章

[springboot+rabbitmq搭建实现mqtt协议订阅发布\_qianQueen的博客-CSDN博客][springboot_rabbitmq_mqtt_qianQueen_-CSDN]

#### 1.rabbitmq基本讲解 ####

RabbitMQ简单来说就是一个消息队列中间件,用来保存消息和传递消息的一个容器。在此过程中充当一个中间人的作用。 是一种程序对程序的通信方法，其服务器也是以高性能、健壮以及可伸缩性出名的Erlang语言编写而成。

rabbitmq的搭建很简单，在此就不再做展示了。这是安装好的rabbitmq服务端界面

![d1de5b846cf141d9a1449a9cbeba0108.png][]

在网上找了个根据rabbitmq功能大概制成的图：

![f4f023724da14bcea669e4bb0e16ee33.jpeg][]

生产者发送消息，再由rabbitmq服务端的交换机以及路由和队列一系列操作，再由监听的消费者消费。 主要的交换机分为三种（还有其他交换机，这里只讲主要的三种）：

**1.Direct Exchange直连型交换机：**

大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。

**2.Fanout Exchange扇形交换机：**

扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

**3.Topic Exchange主题交换机：**

主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的

**\* (星号) 用来表示一个单词 (必须出现的)**

**\# (井号) 用来表示任意数量（零个或多个）单词**

**注：主题交换机可以实现直连型交换机和扇形交换机的功能。**

### 2.接下来就展示代码，需要一个provider项目和consumer项目。 ###

首先是项目:

pom文件所需依赖：

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

application.yml配置文件：

spring:
        application:
          rabbitmq-provider
        rabbitmq:
          host: 127.0.0.1
          port: 5672
          username: guest
          password: guest

#### **2.1 Direct Exchange直连型交换机代码：**  ####

配置类文件代码：

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @Data
    @Configuration
    public class DirectRabbitConfig {
    
        public static String topic = "TestDirectQueue";
    
        public static String exchange = "TestDirectExchange";
    
        public static String routingKey = "TestDirectRouting";
    
        //队列 起名：TestDirectQueue
        @Bean
        public Queue TestDirectQueue() {
            // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            //   return new Queue("TestDirectQueue",true,true,false);
     
            //一般设置一下队列的持久化就好,其余两个就是默认false
            return new Queue(topic,true);
        }
     
        //Direct交换机 起名：TestDirectExchange
        @Bean
        DirectExchange TestDirectExchange() {
            return new DirectExchange(exchange,true,false);
        }
     
        //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
        @Bean
        Binding bindingDirect() {
            return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(routingKey);
        }
     
        @Bean
        DirectExchange lonelyDirectExchange() {
            return new DirectExchange("lonelyDirectExchange");
        }
     
     
    }

发送消息接口：

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @RequestMapping("rabbitmq")
    @RestController
    public class RabbitmqSendController {
     
        @Resource
        RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
     
        @GetMapping("/send-direct-message")
        public String sendDirectMessage() {
            String messageId = String.valueOf(UUID.randomUUID());
            String messageData = "directMessage, hello!" + System.currentTimeMillis();
            Map<String,Object> map=new HashMap<>();
            map.put("messageId",messageId);
            map.put("messageData",messageData);
            map.put("createTime", DateUtils.getTime());
            //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
            rabbitTemplate.convertAndSend(DirectRabbitConfig.exchange, DirectRabbitConfig.routingKey, map);
            return "ok";
        }
    
    
    }

接下来启动provider项目，可在rabbitmq服务端查看发送过去的消息。

![ead3b800f4d84458b2992c0ce1f70a28.png][]

接下来就是创建consumer项目，pom文件和application.yml文件和provider项目一致，这里不展示了，**如果消费者只单纯消费消息，则只需要监听队列即可，如果需要作为生产者发送消息，则需要一个DirectRabbitConfig文件（其他交换机示例代码则不再展示消费者的该配置文件）**。

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @Configuration
    public class DirectRabbitConfig {
     
        //队列 起名：TestDirectQueue
        @Bean
        public Queue TestDirectQueue() {
            return new Queue("TestDirectQueue",true);
        }
     
        //Direct交换机 起名：TestDirectExchange
        @Bean
        DirectExchange TestDirectExchange() {
            return new DirectExchange("TestDirectExchange");
        }
     
        //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
        @Bean
        Binding bindingDirect() {
            return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
        }
        
    }

接下来就是consumer消费者监听类：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Map;
    
    @Component
    @RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
    public class DirectReceiver {
     
        @RabbitHandler
        public void process(Map testMessage) {
            System.out.println("DirectReceiver消费者1收到消息  : " + testMessage.toString());
        }
     
    }

接下来启动consumer端，将收到provider发送的消息。

![f2aec11dd9714ca09f6b1d2fcf536d0f.png][]

直连型交换机是一对一，多个消费者监听了同一个队列，那么会轮询消费，且不会重复消费。

![531dd2d963c04cf183ef33f069edecac.png][]

####  **2.2 Fanout Exchange扇形交换机代码：**  ####

provider配置类文件代码：

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
     
    @Configuration
    public class FanoutRabbitConfig {
     
        /**
         *  创建三个队列 ：fanout.test1   fanout.test2
         *  将三个队列都绑定在交换机 fanoutExchange 上
         *  因为是扇型交换机, 路由键无需配置,配置也不起作用
         */
     
     
        @Bean
        public Queue queueA() {
            return new Queue("fanout.test1");
        }
     
        @Bean
        public Queue queueB() {
            return new Queue("fanout.test2");
        }
     
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");
        }
     
        @Bean
        Binding bindingExchangeA() {
            return BindingBuilder.bind(queueA()).to(fanoutExchange());
        }
     
        @Bean
        Binding bindingExchangeB() {
            return BindingBuilder.bind(queueB()).to(fanoutExchange());
        }
    
    }

provider发送消息接口

@GetMapping("/send-fanout-message")
        public String sendFanoutMessage() {
            String messageId = String.valueOf(UUID.randomUUID());
            String messageData = "message: testFanoutMessage ";
            Map<String, Object> map = new HashMap<>();
            map.put("messageId", messageId);
            map.put("messageData", messageData);
            rabbitTemplate.convertAndSend("fanoutExchange", null, map);
            return "ok";
        }

接下来是consumer消费者监听类：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.util.Map;
    /**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @Component
    @RabbitListener(queues = "fanout.test1")
    public class FanoutReceiverTest1 {
     
        @RabbitHandler
        public void process(Map testMessage) {
            System.out.println("FanoutReceiverTest1消费者收到消息  : " +testMessage.toString());
        }
     
    }

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.util.Map;
    /**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @Component
    @RabbitListener(queues = "fanout.test2")
    public class FanoutReceiverTest2 {
     
        @RabbitHandler
        public void process(Map testMessage) {
            System.out.println("FanoutReceiverTest2消费者收到消息  : " +testMessage.toString());
        }
     
    }

先启动provider项目，再启动consumer项目，发送消息：

![b08a817feaa2420584f617de57ab7cec.png][]

####  **2.3 Topic Exchange主题交换机代码：** ####

provider配置类：

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;
     
    /**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
     
    @Configuration
    public class TopicRabbitConfig {
        //绑定键
        public final static String student = "topic.school.student";
        public final static String teacher = "topic.school.teacher";
    
        public static String exchange = "topicExchange";
     
        @Bean
        public Queue firstQueue() {
            return new Queue(TopicRabbitConfig.student);
        }
     
        @Bean
        public Queue secondQueue() {
            return new Queue(TopicRabbitConfig.teacher);
        }
     
        @Bean
        TopicExchange exchange() {
            return new TopicExchange(exchange);
        }
     
     
        //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
        //这样只要是消息携带的路由键是topic.man,才会分发到该队列
        @Bean
        Binding bindingExchangeMessage() {
            return BindingBuilder.bind(firstQueue()).to(exchange()).with(student);
        }
     
        //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
        // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
        @Bean
        Binding bindingExchangeMessage2() {
            return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
        }
     
    }

provider新增测试接口:

@GetMapping("/send-topic-student")
        public String sendTopicMessage1(@RequestParam String message) {
            String messageId = String.valueOf(UUID.randomUUID());
            Map<String, Object> manMap = new HashMap<>();
            manMap.put("messageId", messageId);
            manMap.put("messageData", message+": student ");
            manMap.put("createTime", DateUtils.getTime());
            rabbitTemplate.convertAndSend("topicExchange", "topic.school.student", manMap);
            return "ok";
        }
    
        @GetMapping("/send-topic-teacher")
        public String sendTopicMessage2(@RequestParam String message) {
            String messageId = String.valueOf(UUID.randomUUID());
            Map<String, Object> womanMap = new HashMap<>();
            womanMap.put("messageId", messageId);
            womanMap.put("messageData", message+": teacher ");
            womanMap.put("createTime", DateUtils.getTime());
            rabbitTemplate.convertAndSend("topicExchange", "topic.school.teacher", womanMap);
            return "ok";
        }

consumer端新增消费者监听：

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
    @Component
    @RabbitListener(queues = "topic.school.student")
    public class TopicStudentReceiver {
     
        @RabbitHandler
        public void process(Map testMessage) {
            System.out.println("TopicStudentReceiver消费者收到消息  : " + testMessage.toString());
        }
    }

/**
     * @Author : liuqian
     * @CreateTime : 2022/11/22
     * @Description :
     **/
     
    @Component
    @RabbitListener(queues = "topic.school.teacher")
    public class TopicTeacherReceiver {
     
        @RabbitHandler
        public void process(Map testMessage) {
            System.out.println("TopicTeacherReceiver消费者收到消息  : " + testMessage.toString());
        }
    }

consumer端控制台展示：

![9c66d8ec9cd64b2d8ebd785036e1543a.png][]

TopicTeacherReceiver绑定的teacher队列绑定的routingKey为topic.#,不管是student生产者还是teacher生产者发送的消息，TopicTeacherReceiver都收到了消息。

以上就是rabbitmq三个交换机的简单使用，其他的使用在下篇文章继续讲解。

[springboot_rabbitmq_mqtt_qianQueen_-CSDN]: https://blog.csdn.net/qq_36090537/article/details/127923057
[d1de5b846cf141d9a1449a9cbeba0108.png]: https://img-blog.csdnimg.cn/d1de5b846cf141d9a1449a9cbeba0108.png
[f4f023724da14bcea669e4bb0e16ee33.jpeg]: https://img-blog.csdnimg.cn/f4f023724da14bcea669e4bb0e16ee33.jpeg
[ead3b800f4d84458b2992c0ce1f70a28.png]: https://img-blog.csdnimg.cn/ead3b800f4d84458b2992c0ce1f70a28.png
[f2aec11dd9714ca09f6b1d2fcf536d0f.png]: https://img-blog.csdnimg.cn/f2aec11dd9714ca09f6b1d2fcf536d0f.png
[531dd2d963c04cf183ef33f069edecac.png]: https://img-blog.csdnimg.cn/531dd2d963c04cf183ef33f069edecac.png
[b08a817feaa2420584f617de57ab7cec.png]: https://img-blog.csdnimg.cn/b08a817feaa2420584f617de57ab7cec.png
[9c66d8ec9cd64b2d8ebd785036e1543a.png]: https://img-blog.csdnimg.cn/9c66d8ec9cd64b2d8ebd785036e1543a.png