RabbitMQ（七）【SpringBoot案例】

矫情吗；* 2024-04-08 13:16 77阅读 0赞

### 七、RabbitMQ - SpringBoot案例 ###

--------------------

上一篇文章《[基础入门案例][Link 1]》

> 整体核心

![在这里插入图片描述][1568796e74e842f1afe030b82e378b5a.png_pic_center]

#### 7.1 fanout 模式 ####

生产者：交换机绑定队列

![在这里插入图片描述][31e8a8c952f34b0285d9fef76af49cf6.png_pic_center]

1.  创建一个 springboot 项目**springboot-order-rabbitmq-producer**

![在这里插入图片描述][e38da15d0cab4dcebb25c20b41fc15bf.png_pic_center]

1.  配置文件**application.yml**

# 服务端口
    server:
      port: 8080
    
    # 配置rabbitmq
    spring:
      rabbitmq:
        username: admin
        password: admin
        virtual-host: /
        host: 192.168.159.100
        port: 5672

1.  添加依赖

<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>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>

1.  编写一个**OrderServiceImpl.java**

1.  编写**FanoutRabbitMQConfiguration.java**

package com.vinjcent.rabbitmq.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class FanoutRabbitMQConfiguration {
        
    
        // 1.声明注册fanout模式交换机
        @Bean
        public FanoutExchange fanoutExchange(){
        
            return new FanoutExchange("fanout_order_exchange",true,false);
        }
    
    
        // 2.声明队列 account.fanout.queue、express.fanout.queue、sms.fanout.queue
        @Bean
        public Queue fanout_accountQueue(){
        
            return new Queue("account.fanout.queue",true);
        }
        @Bean
        public Queue fanout_expressQueue(){
        
            return new Queue("express.fanout.queue",true);
        }
        @Bean
        public Queue fanout_smsQueue(){
        
            return new Queue("sms.fanout.queue",true);
        }
        // 3.完成绑定关系(队列)
        @Bean
        public Binding fanout_accountBinding(){
        
            return BindingBuilder.bind(fanout_accountQueue()).to(fanoutExchange());
        }
        @Bean
        public Binding fanout_expressBinding(){
        
            return BindingBuilder.bind(fanout_expressQueue()).to(fanoutExchange());
        }
        @Bean
        public Binding fanout_smsBinding(){
        
            return BindingBuilder.bind(fanout_smsQueue()).to(fanoutExchange());
        }
    
    }

1.  在web界面中将队列 queue 和交换机 exchange 清空

![在这里插入图片描述][d2200ddcf02640d88354ebfa83ee0a11.png_pic_center]

![在这里插入图片描述][0ecfcdb4527f4a0f89fb13e31c5448c6.png_pic_center]

1.  运行测试

1.  观察web界面中的消息队列和交换机

![在这里插入图片描述][261be68e04fe4c15b188e1a795eb8aa1.png_pic_center]

![在这里插入图片描述][2c784e28c20249aa9153a9f773a6d92a.png_pic_center]

![在这里插入图片描述][75b9ae49988246198c8ee3b88f719108.png_pic_center]

消费者接收消息

1.  创建一个 springboot 项目**springboot-order-rabbitmq-consumer**

![在这里插入图片描述][e4da027d72bd453a94f954ec4ade1904.png_pic_center]

1.  配置文件

# 应用服务 WEB 访问端口
    server:
      port: 80
    
    # 配置rabbitmq
    spring:
      rabbitmq:
        username: admin
        password: admin
        virtual-host: /
        host: 192.168.159.100
        port: 5672
      application:
        # 应用名称
        name: springboot-order-rabbitmq-consumer

1.  在`service`包下添加一个`fanout`包，添加以下类，注入Spring容器中

（1）**FanoutAccountConsumer.java**监听队列`account.fanout.queue`、`express.fanout.queue`、`sms.fanout.queue`

package com.vinjcent.rabbitmq.service.fanout;
    
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(queues = {
        "account.fanout.queue"})
    public class FanoutAccountConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("account.fanout.queue===>" + message);
        }
    
    }

（2）**FanoutExpressConsumer.java**

package com.vinjcent.rabbitmq.service.fanout;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(queues = {
        "express.fanout.queue"})
    public class FanoutExpressConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("express.fanout.queue===>" + message);
        }
    
    }

（3）**FanoutSMSConsumer.java**

package com.vinjcent.rabbitmq.service.fanout;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(queues = {
        "sms.fanout.queue"})
    public class FanoutSMSConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("sms.fanout.queue===>" + message);
        }
    
    }

1.  运行**springboot-order-rabbitmq-consumer**工程
2.  查看控制台打印结果

![在这里插入图片描述][293c8efa9669499f9a00cd8b064d4446.png_pic_center]

![在这里插入图片描述][10d1171ced854d078e127b1d1bed2773.png_pic_center]

#### 7.2 direct 模式 ####

1）**生产者**：交换机绑定队列

1.  在**springboot-order-rabbitmq-producer**工程的`service`包下**OrderServiceImpl.java**添加如下；并配置一个**DirectRabbitMQConfiguration.java**类

**OrderServiceImpl.java**

package com.vinjcent.rabbitmq;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.UUID;
    
    @SuppressWarnings("all")
    @Service
    public class OrderServiceImpl {
        
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        // fanout模式...
    
    
        // direct模式
        public void createOrderDirect(String userId, String productId,int num){
        
            // 1.根据商品id查询库存是否充足
            // 2.保存订单
            String orderId = UUID.randomUUID().toString();
            System.out.println("订单生产成功: " + orderId);
            // 3.通过MQ来完成消息的分发
            // 参数1: 交换机   参数2: 路由key/queue队列名称   参数3: 消息内容
            String exchangeName = "direct_order_exchange";
            String routingKey = "";
            // 根据路由推送消息
            rabbitTemplate.convertAndSend(exchangeName, "account", orderId);
            rabbitTemplate.convertAndSend(exchangeName, "express", orderId);
    
        }
    
    
    }

**DirectRabbitMQConfiguration.java**

package com.vinjcent.rabbitmq.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class DirectRabbitMQConfiguration {
        
    
    
        // 1.声明注册fanout模式交换机
        @Bean
        public DirectExchange directExchange(){
        
            return new DirectExchange("direct_order_exchange",true,false);
        }
    
    
        // 2.声明队列 account.direct.queue、express.direct.queue、sms.direct.queue
        @Bean
        public Queue direct_accountQueue(){
        
            return new Queue("account.direct.queue",true);
        }
        @Bean
        public Queue direct_expressQueue(){
        
            return new Queue("express.direct.queue",true);
        }
        @Bean
        public Queue direct_smsQueue(){
        
            return new Queue("sms.direct.queue",true);
        }
        // 3.完成绑定关系(队列)
        // direct模式比fanout模式多了一个路由key
        @Bean
        public Binding direct_accountBinding(){
        
            return BindingBuilder.bind(direct_accountQueue()).to(directExchange()).with("account");
        }
        @Bean
        public Binding direct_expressBinding(){
        
            return BindingBuilder.bind(direct_expressQueue()).to(directExchange()).with("express");
        }
        @Bean
        public Binding direct_smsBinding(){
        
            return BindingBuilder.bind(direct_smsQueue()).to(directExchange()).with("sms");
        }
    
    }

1.  测试类

@SpringBootTest
    class SpringbootOrderRabbitmqProducerApplicationTests {
        
    
        @Autowired
        OrderServiceImpl orderService;
    
        @Test
        void testFanout() {
        
            orderService.createOrderFanout("1","1",12);
        }
    
        // 运行该测试用例
        @Test
        void testDirect() {
        
            orderService.createOrderDirect("1","1",12);
        }
    
    }

1.  在web界面观察交换机与队列情况

![在这里插入图片描述][d8ddaf802ab14898b1836f84468f2be4.png_pic_center]

![在这里插入图片描述][0fa6ddc5342b402dab1ebc666bd983f5.png_pic_center]

![在这里插入图片描述][b57e21c524af4128b706549a309b3bad.png_pic_center]

【**注**】**先要启动生产者再自动消费者，不然交换机不存在会报错**

2）**消费者接收消息**

1.  在项目**springboot-order-rabbitmq-consumer**工程中的`service`包下添加一个`direct`包，添加以下类，注入Spring容器中

（1）**DirectAccountConsumer.java**监听队列`account.direct.queue`

@Service
    @RabbitListener(queues = {
        "account.direct.queue"})
    public class DirectAccountConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("account.direct.queue===>" + message);
        }
    
    }

（2）**DirectExpressConsumer.java**监听队列`express.direct.queue`

@Service
    @RabbitListener(queues = {
        "express.direct.queue"})
    public class DirectExpressConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("express.direct.queue===>" + message);
        }
    
    }

（3）**DirectSMSConsumer.java**监听队列`sms.direct.queue`

@Service
    @RabbitListener(queues = {
        "sms.direct.queue"})
    public class DirectSMSConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("sms.direct.queue===>" + message);
        }
    
    }

1.  运行**springboot-order-rabbitmq-consumer**
2.  查看控制台打印结果

![在这里插入图片描述][9a14be26f3ce46a6815fb4694d408a1b.png_pic_center]

#### 7.3 topic 模式 ####

> 使用注解方式实现绑定

消费者

1.  在**springboot-order-rabbitmq-consumer**工程下的`service`包下添加一个`topic`包，添加以下类，注入Spring容器中

（1）**TopicAccountConsumer.java**

package com.vinjcent.rabbitmq.service.topic;
    
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "account.topic.queue", durable = "true", autoDelete = "false"),
            exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
            key = "#.account.#"
    ))
    public class TopicAccountConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("account.topic.queue===>" + message);
        }
    
    }

（2）**TopicExpressConsumer.java**

package com.vinjcent.rabbitmq.service.topic;
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "express.topic.queue", durable = "true", autoDelete = "false"),
            exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
            key = "*.express.#"
    ))
    public class TopicExpressConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("express.topic.queue===>" + message);
        }
    
    }

（3）**TopicSMSConsumer.java**

package com.vinjcent.rabbitmq.service.topic;
    
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Service;
    
    @Service
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "sms.topic.queue", durable = "true", autoDelete = "false"),
            exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC),
            key = "sms.#"
    ))public class TopicSMSConsumer {
        
    
        @RabbitHandler
        public void receiveMessage(String message){
        
            System.out.println("sms.topic.queue===>" + message);
        }
    
    }

1.  启动**springboot-order-rabbitmq-consumer**
2.  查看web界面

![在这里插入图片描述][4af04cbf4f4f424a8360ca196d93ae67.png_pic_center]

![在这里插入图片描述][a6abb7fe0d2746549747b5242024823b.png_pic_center]

![在这里插入图片描述][8b10ec5a43404f888559774c9631739e.png_pic_center]

**推荐使用配置类的方式实现交换机与队列的绑定，以及路由key规则**

生产者

1.  在**springboot-order-rabbitmq-producer**工程下，`service`包下添加以下内容

package com.vinjcent.rabbitmq;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.UUID;
    
    @SuppressWarnings("all")
    @Service
    public class OrderServiceImpl {
        
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        // fanout模式
        public void createOrderFanout(String userId, String productId,int num){
        
            //......
        }
    
        // direct模式
        public void createOrderDirect(String userId, String productId,int num){
        
            //......
        }
    
        // topic
        public void createOrderTopic(String userId, String productId,int num){
        
            // 1.根据商品id查询库存是否充足
            // 2.保存订单
            String orderId = UUID.randomUUID().toString();
            System.out.println("订单生产成功: " + orderId);
            // 3.通过MQ来完成消息的分发
            // 参数1: 交换机   参数2: 路由key/queue队列名称   参数3: 消息内容
            String exchangeName = "topic_order_exchange";
            // 路由给三个消息队列推送消息
            String routingKey = "sms.express.account.xxxxx";
    
            /*
             *  #.account.#
             *  *.express.#
             *  sms.#
             */
            rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
    
        }
    
    
    }

1.  运行测试用例，生产消息

@SpringBootTest
    class SpringbootOrderRabbitmqProducerApplicationTests {
        
    
        @Autowired
        OrderServiceImpl orderService;
    
        @Test
        void contextLoads() {
        
            orderService.createOrderFanout("1","1",12);
        }
    
        @Test
        void testDirect() {
        
            orderService.createOrderDirect("1","1",12);
        }
    
        // 运行该测试用例
        @Test
        void testTopic() {
        
            orderService.createOrderTopic("1","1",12);
        }
    }

![在这里插入图片描述][da6b8522dbd347aa9855a3abca9a8f07.png_pic_center]

下一篇文章《[RabbitMQ高级 - 过期时间 TTL][RabbitMQ_ - _ TTL]》

[Link 1]: https://blog.csdn.net/Wei_Naijia/article/details/126168642
[1568796e74e842f1afe030b82e378b5a.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/21db2136d7f14e2a9ead17ca8bdb0689.png
[31e8a8c952f34b0285d9fef76af49cf6.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/a2ce6c5496bc4e17935b502efb2ffb34.png
[e38da15d0cab4dcebb25c20b41fc15bf.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/045aee262e074063b91d0e07b113edb3.png
[d2200ddcf02640d88354ebfa83ee0a11.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/f896a2984aba40ef87042407f8781faf.png
[0ecfcdb4527f4a0f89fb13e31c5448c6.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/cb2ecc66ad4348f4a18d1189206e2992.png
[261be68e04fe4c15b188e1a795eb8aa1.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/d5f8494b0e8349de935fe5c550f2ac6c.png
[2c784e28c20249aa9153a9f773a6d92a.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b097faa355f84a6cbdfd1f141d7625c8.png
[75b9ae49988246198c8ee3b88f719108.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/8204c37fee884dbcad90034f554c3217.png
[e4da027d72bd453a94f954ec4ade1904.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/bb8a933e027b4b85a90c29104a8e0c66.png
[293c8efa9669499f9a00cd8b064d4446.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e2fc69fe941d45d4aedc25e023959236.png
[10d1171ced854d078e127b1d1bed2773.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/ac759d7f326a45dd88d5c7ec5e360288.png
[d8ddaf802ab14898b1836f84468f2be4.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/975846eb9f0c4a28abf90fa149cfbb32.png
[0fa6ddc5342b402dab1ebc666bd983f5.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/feb8b73668d649f2a8a770f0d9622f5a.png
[b57e21c524af4128b706549a309b3bad.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/44da7feeb62442a895d8b878da259a68.png
[9a14be26f3ce46a6815fb4694d408a1b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/6f258be2582e4ac18a25d256560ee2f4.png
[4af04cbf4f4f424a8360ca196d93ae67.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/eb2c0f26843c45c69387fcfad6af97b2.png
[a6abb7fe0d2746549747b5242024823b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/adf5a05fbbe044efb58ee984f04614eb.png
[8b10ec5a43404f888559774c9631739e.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/ee051f622b5f483cac436ddb3af10fed.png
[da6b8522dbd347aa9855a3abca9a8f07.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/c3c51de50f9c4f0b816049112b0d3aab.png
[RabbitMQ_ - _ TTL]: https://blog.csdn.net/Wei_Naijia/article/details/126564240