【SpringBoot】SpringBoot 集成 RabbitMQ（二）（其它模式）

朱雀 2023-10-02 10:24 22阅读 0赞

这篇文章内容：SpringBoot 集成 RabbitMQ 的其它模式：工作队列模式、

### 1. 工作队列模式 WorkQueue ###

工作队列模式：一个生产者对应多个消费者

#### 1.1 生产者 ####

生产者工程结构图：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_14_color_FFFFFF_t_70_g_se_x_16]  
application.yml

server:
      port: 8081
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        virtual-host: /

1》、新配置一个工作队列

@Configuration
    public class WorkQueueConfig {
        
    
        // 创建队列
        @Bean
        public Queue workQueue() {
        
            return new Queue(RabbitMqConstant.WORK_QUEUE_NAME);
        }
    }

2》、编写发送消息逻辑

@Component
    public class WorkProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "工作队列模式下的消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.WORK_QUEUE_NAME, msg);
            }
        }
    }

常量类

public interface RabbitMqConstant {
        
        String SIMPLE_QUEUE_NAME = "simple_queue";
        String WORK_QUEUE_NAME = "work_queue";
    }

4》、添加一个发送消息的接口

@RestController
    @RequestMapping("/work")
    public class WorkController {
        
    
        @Autowired
        private WorkProducer workProducer;
    
        @RequestMapping("/sendMsg")
        public String sendMsg() {
        
            workProducer.sendMsg();
            return "WORK-QUEUE";
        }
    }

使用 PostMan 调用此接口：`http://localhost:8081/work/sendMsg`

#### 1.2 消费者 ####

消费者工程结构图  
![在这里插入图片描述][fa64d13397cc49c7ae61f0a6e491a091.png]  
application.yml

server:
      port: 8080
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        virtual-host: /
        listener:
          simple:
            acknowledge-mode: manual
            concurrency: 1
            max-concurrency: 2
            retry:
              enabled: true

编写两个消费者：

Component
    public class WorkConsumer {
        
    
        @RabbitListener(queues = {
        "work_queue"})
        public void receive(Message message, Channel channel) throws Exception{
        
            Thread.sleep(1000);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息1：" + new String(message.getBody()));
        }
    
        @RabbitListener(queues = {
        "work_queue"})
        public void receive2(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息2：" + new String(message.getBody()));
        }
    }

控制台打印，发现10条消息由两个消费者消费，并且平均分配。

通过配置可控分配数实现按需分配，即谁的性能强，谁优先原则，实现负载均衡。

### 2. 发布/订阅模式（广播模式） Fanout Queue ###

需要新建一个 Exchange，它的类型为 fanout，不需要路由键。

application.yml 配置文件不变

#### 2.1 生产者 ####

1》、配置队列、交换机、队列与交换机之间的关系

@Configuration
    public class FanoutQueueConfig {
        
    
        // 声明队列
        @Bean
        public Queue fanoutQueue1() {
        
            return new Queue(RabbitMqConstant.FAN_OUT_QUEUE_NAME_1);
        }
    
        @Bean
        public Queue fanoutQueue2() {
        
            return new Queue(RabbitMqConstant.FAN_OUT_QUEUE_NAME_2);
        }
    
        // 声明交换机
        @Bean
        public FanoutExchange exchange() {
        
            return new FanoutExchange(RabbitMqConstant.FAN_OUT_EXCHANGE);
        }
    
        // 声明交换机与队列之间的关系
        @Bean
        public Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange exchange) {
        
            return BindingBuilder.bind(fanoutQueue1).to(exchange);
        }
    
        @Bean
        public Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange exchange) {
        
            return BindingBuilder.bind(fanoutQueue2).to(exchange);
        }
    }

2》、发送消息

注意：这里是往交换机中发送消息

@Component
    public class FanoutProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "发布/订阅模式下的消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.FAN_OUT_EXCHANGE, "", msg);
            }
        }
    }

3》、常量类

public interface RabbitMqConstant {
        
        String SIMPLE_QUEUE_NAME = "simple_queue";
        String WORK_QUEUE_NAME = "work_queue";
        /**
         * fanout
         */
        String FAN_OUT_QUEUE_NAME_1 = "fanout_queue_1";
        String FAN_OUT_QUEUE_NAME_2 = "fanout_queue_2";
        String FAN_OUT_EXCHANGE = "fanout_exchange";
    }

4》、调用接口，发送消息

@RestController
    @RequestMapping("/fanout")
    public class FanoutController {
        
    
        @Autowired
        private FanoutProducer fanoutProducer;
    
        @RequestMapping("/sendMsg")
        public String sendMsg() {
        
            fanoutProducer.sendMsg();
            return "FAN-OUT-QUEUE";
        }
    }

#### 2.2 消费者 ####

@Component
    public class FanoutConsumer {
        
    
        @RabbitListener(queues = {
        "fanout_queue_1"})
        public void receive(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息1：" + new String(message.getBody()));
        }
    
        @RabbitListener(queues = {
        "fanout_queue_2"})
        public void receive2(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息2：" + new String(message.getBody()));
        }
    }

### 3. 路由模式 Direct Queue ###

路由模式 = 发布/订阅模式 + 指定 RoutingKey

#### 3.1 生产者 ####

1》、配置队列、交换机、队列与交换机之间的关系

@Configuration
    public class DirectQueueConfig {
        
    
        // 声明队列
        @Bean
        public Queue directQueue1() {
        
            return new Queue(RabbitMqConstant.DIRECT_QUEUE_NAME_1);
        }
    
        @Bean
        public Queue directQueue2() {
        
            return new Queue(RabbitMqConstant.DIRECT_QUEUE_NAME_2);
        }
    
        // 声明交换机
        @Bean
        public DirectExchange exchange() {
        
            return new DirectExchange(RabbitMqConstant.DIRECT_EXCHANGE);
        }
    
        // 声明交换机与队列之间的关系
        @Bean
        public Binding bindingDirectQueue1(Queue directQueue1, DirectExchange exchange) {
        
            return BindingBuilder.bind(directQueue1).to(exchange).with(RabbitMqConstant.DIRECT_ROUTING_KEY_1);
        }
    
        @Bean
        public Binding bindingDirectQueue2(Queue directQueue2, DirectExchange exchange) {
        
            return BindingBuilder.bind(directQueue2).to(exchange).with(RabbitMqConstant.DIRECT_ROUTING_KEY_2);
        }
    }

2》、发送消息

注意：这里是往交换机中发送消息

@Component
    public class DirectProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsgA() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "路由模式下的消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.DIRECT_EXCHANGE, RabbitMqConstant.DIRECT_ROUTING_KEY_1, msg);
            }
        }
    
        public void sendMsgB() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "路由模式下的消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.DIRECT_EXCHANGE, RabbitMqConstant.DIRECT_ROUTING_KEY_2, msg);
            }
        }
    }

3》、常量类

4》、调用接口，发送消息

@RestController
    @RequestMapping("/direct")
    public class DirectController {
        
    
        @Autowired
        private DirectProducer directProducer;
    
        @RequestMapping("/sendMsgA")
        public String sendMsgA() {
        
            directProducer.sendMsgA();
            return "DIRECT-QUEUE";
        }
    
        @RequestMapping("/sendMsgB")
        public String sendMsgB() {
        
            directProducer.sendMsgB();
            return "DIRECT-QUEUE";
        }
    }

#### 3.2 消费者 ####

@Component
    public class DirectConsumer {
        
    
        @RabbitListener(queues = {
        "direct_queue_1"})
        public void receive(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息1：" + new String(message.getBody()));
        }
    
        @RabbitListener(queues = {
        "direct_queue_2"})
        public void receive2(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息2：" + new String(message.getBody()));
        }
    }

### 4. 主题模式 Topic Queue ###

#### 4.1 生产者 ####

1》、配置队列、交换机、队列与交换机之间的关系

@Configuration
    public class TopicQueueConfig {
        
    
        // 声明队列
        @Bean
        public Queue topicQueue1() {
        
            return new Queue(RabbitMqConstant.TPOIC_QUEUE_NAME_1);
        }
    
        @Bean
        public Queue topicQueue2() {
        
            return new Queue(RabbitMqConstant.TPOIC_QUEUE_NAME_2);
        }
    
        // 声明交换机
        @Bean
        public TopicExchange exchange() {
        
            return new TopicExchange(RabbitMqConstant.TOPIC_EXCHANGE);
        }
    
        // 声明交换机与队列之间的关系
        @Bean
        public Binding bindingDirectQueue1(Queue topicQueue1, TopicExchange exchange) {
        
            return BindingBuilder.bind(topicQueue1).to(exchange).with(RabbitMqConstant.TOPIC_ROUTING_KEY_1);
        }
    
        /**
        *
        * 功能描述:
        * *：匹配不多不少一个词   #：匹配一个或多个词
        *
        * @param topicQueue2
        * @param exchange
        */
        @Bean
        public Binding bindingDirectQueue2(Queue topicQueue2, TopicExchange exchange) {
        
            return BindingBuilder.bind(topicQueue2).to(exchange).with(RabbitMqConstant.TOPIC_ROUTING_KEY_2);
        }

2》、发送消息

注意：这里是往交换机中发送消息

@Component
    public class TopicProducer {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsgA() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "通配符模式--routingKey=topic.key消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE, RabbitMqConstant.TOPIC_ROUTING_KEY_1, msg);
            }
        }
    
        public void sendMsgB() {
        
            for (int i = 0; i < 10; i++) {
        
                String msg = "通配符模式--routingKey=topic.#消息 " + i;
                rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE, RabbitMqConstant.TOPIC_ROUTING_KEY_3, msg);
            }
        }
    }

3》、常量类

public interface RabbitMqConstant {
        
        // simple
        String SIMPLE_QUEUE_NAME = "simple_queue";
    
        // work
        String WORK_QUEUE_NAME = "work_queue";
    
        // fanout
        String FAN_OUT_QUEUE_NAME_1 = "fanout_queue_1";
        String FAN_OUT_QUEUE_NAME_2 = "fanout_queue_2";
        String FAN_OUT_EXCHANGE = "fanout_exchange";
    
       // direct
        String DIRECT_QUEUE_NAME_1 = "direct_queue_1";
        String DIRECT_QUEUE_NAME_2 = "direct_queue_2";
        String DIRECT_EXCHANGE = "direct_exchange";
        String DIRECT_ROUTING_KEY_1 = "direct_add";
        String DIRECT_ROUTING_KEY_2 = "direct_update";
    
        // topic
        String TPOIC_QUEUE_NAME_1 = "topic_queue_1";
        String TPOIC_QUEUE_NAME_2 = "topic_queue_2";
        String TOPIC_EXCHANGE = "topic_exchange";
        String TOPIC_ROUTING_KEY_1 = "topic.key";
        String TOPIC_ROUTING_KEY_2 = "topic.#";
        String TOPIC_ROUTING_KEY_3 = "topic.key2.key3";
    }

4》、调用接口，发送消息

@RestController
    @RequestMapping("/topic")
    public class TopicController {
        
    
        @Autowired
        private TopicProducer topicProducer;
    
        @RequestMapping("/sendMsgA")
        public String sendMsgA() {
        
            topicProducer.sendMsgA();
            return "TOPIC-QUEUE";
        }
    
        @RequestMapping("/sendMsgB")
        public String sendMsgB() {
        
            topicProducer.sendMsgB();
            return "TOPIC-QUEUE";
        }
    }

#### 4.2 消费者 ####

@Component
    public class TopicConsumer {
        
    
        @RabbitListener(queues = {
        "topic_queue_1"})
        public void receive(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息1：" + new String(message.getBody()));
        }
    
        @RabbitListener(queues = {
        "topic_queue_2"})
        public void receive2(Message message, Channel channel) throws Exception{
        
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("我是消费信息2：" + new String(message.getBody()));
        }
        
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAc2NvNTI4Mg_size_14_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/68601d9bce3540cea519c31cb761aa6a.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAc2NvNTI4Mg==,size_14,color_FFFFFF,t_70,g_se,x_16
[fa64d13397cc49c7ae61f0a6e491a091.png]: https://img-blog.csdnimg.cn/fa64d13397cc49c7ae61f0a6e491a091.png