SpringBoot整合RabbitMQ之Exchange、Queue参数详解

以你之姓@ 2023-05-31 07:57 24阅读 0赞

### 前言 ###

之前一直用的activemq，因为项目需要，需要使用rabbitmq，所以这里学习总结一下。

**先看一下RabbitMQ中的成员：**

> Producer(生产者)： 将消息发送到Exchange  
> Exchange(交换器)：将从生产者接收到的消息路由到Queue  
> Queue(队列)：存放供消费者消费的消息  
> BindingKey(绑定键)：建立Exchange与Queue之间的关系  
> RoutingKey(路由键)：Producer发送消息与路由键给Exchange，Exchange将判断RoutingKey是否符合BindingKey，如何则将该消息路由到绑定的Queue  
> Consumer(消费者)：从Queue中获取消息

引用一张网上的图来表示各个成员的图解：

![å¨è¿éæå¥å¾çæè¿°][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoNTU0MTEyMDc1_size_16_color_FFFFFF_t_70]

**先介绍Exchange和Queue的各个参数解释 :**

**1.Exchange:**

1.  exchange：名称
2.  type：类型
3.  durable：是否持久化，RabbitMQ关闭后，没有持久化的Exchange将被清除
4.  autoDelete：是否自动删除，如果没有与之绑定的Queue，直接删除
5.  internal：是否内置的，如果为true，只能通过Exchange到Exchange
6.  arguments：结构化参数

**2.Queue:**

queue多了一个exclusive

exclusive：是否排他，如果未true，则只在第一次创建它的Connection中有效，当Connection关闭，该Queue也会被删除

### exchange ###

**exchange：交换机（常用有三种），用于接收生产者发来的消息，并通过binding-key 与 routing-key 的匹配关系来决定将消息分发到指定queue**

1.Direct（路由模式）：完全匹配 > 当消息的routing-key 与 exchange和queue间的binding-key完全匹配时，将消息分发到该queue

2.Fanout （订阅模式）：与binding-key和routing-key无关，将接受到的消息分发给有绑定关系的所有队列（不论binding-key和routing-key是什么）

3.Topic （通配符模式）：用消息的routing-key 与 exchange和queue间的binding-key 进行模式匹配，当满足规则时，分发到满足规则的所有队列  
Topic Exchange交换机也叫通配符交换机，我们在发送消息到Topic Exchange的时候不能随意指定route key（应该是由一系列点号连接的字符串，一般会与binding key有关联，route key的长度一般不能超过255个字节)。同理，交换机与队列之间的binding key也应该是点号连接成的字符串，当消息发送者发送信息到Topic Exchage交换机的时候，这时候发送消息的route key会与binding key进行通配符匹配，所有匹配成功的消息都会发送到消息接受者。

Topic Exchange主要有两种通配符：\# 和 \*

*  \*（星号）：可以（只能）匹配一个单词
 *  \#（井号）：可以匹配多个单词（或者零个）

下面给出例子：

config：

package com.zoo.rabbitmq.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @author: 谢飞
     */
    @Slf4j
    @Configuration
    public class RabbitConfig {
        public final static String SEND_MAIL_QUEUE = "send-mail-queue";
        public final static String SEND_MESSAGE_QUEUE = "send-message-queue";
        //交换器
        public final static String TOPIC_EXCHANGE = "topicExchange";
        public final static String FANOUT_EXCHANGE = "fanoutExchange";
    /队列
    
        /**
         * 测试队列实例
         */
        @Bean
        public Queue testQueue() {
            log.info("【【【测试队列实例创建成功】】】");
            return new Queue("test-queue");
        }
    
        /**
         * 发送邮件队列实例，并持久化
         */
        @Bean
        public Queue sendMailQueue() {
            Queue sendMailQueue = new Queue(SEND_MAIL_QUEUE, true);
            log.info("【【【发送邮件队列实例创建成功】】】");
            return sendMailQueue;
        }
    
        /**
         * 发送消息队列实例，并持久化
         */
        @Bean
        public Queue sendMessageQueue() {
            Queue sendMessageQueue = new Queue(SEND_MESSAGE_QUEUE, true);
            log.info("【【【发送消息队列实例创建成功】】】");
            return sendMessageQueue;
        }
    
        @Bean
        public Queue A() {
            return new Queue("fanout.A");
        }
    
        @Bean
        public Queue B() {
            return new Queue("fanout.B");
        }
    
    
    /交换机
    
        /**
         * 创建Topic Exchange交换机也叫通配符交换机
         * <p>
         * Topic Exchange主要有两种通配符：# 和 *
         * *（星号）：可以（只能）匹配一个单词
         * #（井号）：可以匹配多个单词（或者零个）
         */
        @Bean
        TopicExchange topicExchange() {
            return new TopicExchange(TOPIC_EXCHANGE);
        }
    
        /**
         * Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout转发器发送消息，绑定了这个转发器的所有队列都收到这个消息。
         */
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange(FANOUT_EXCHANGE);
        }
    
    
    /绑定
    
        /**
         * 绑定发送邮件队列到交换机
         * 完全匹配,就是routingKey必须是send-mail-queue
         */
        @Bean
        public Binding sendMailBinding() {
            Binding binding = BindingBuilder.bind(sendMailQueue()).to(topicExchange()).with(SEND_MAIL_QUEUE);
            log.info("【【【发送邮件队列与交换机绑定成功】】】");
            return binding;
        }
    
        /**
         * 绑定发送消息队列到交换机
         */
        @Bean
        public Binding sendMessageBinding() {
            Binding binding = BindingBuilder.bind(sendMessageQueue()).to(topicExchange()).with("message.*");
            log.info("【【【发送消息队列与交换机绑定成功】】】");
            return binding;
        }
    
    
        @Bean
        Binding bindingA() {
            return BindingBuilder.bind(A()).to(fanoutExchange());
        }
    
        @Bean
        Binding bindingB() {
            return BindingBuilder.bind(B()).to(fanoutExchange());
        }
    
    }

consumer：

/**
     * @author: 谢飞
     */
    @Component
    @RabbitListener(queues = "fanout.A")
    public class FanoutReceiverA {
        @RabbitHandler
        @Transactional
        public void handler(String msg) {
            System.out.println("ReceiverA  : " + msg);
        }
    }
    
    /**
     * @author: 谢飞
     */
    @Component
    @RabbitListener(queues = "fanout.B")
    public class FanoutReceiverB {
        @RabbitHandler
        @Transactional
        public void handler(String msg) {
            System.out.println("ReceiverB  : " + msg);
        }
    }
    
    /**
     * @author: 谢飞
     */
    @Component
    @RabbitListener(queues = RabbitConfig.SEND_MAIL_QUEUE)
    public class MailReceiver {
    
        @RabbitHandler
        @Transactional
        public void handler(String msg) {
            System.out.println("Receiver  : " + msg);
        }
    }
    
    /**
     * @author: 谢飞
     */
    @Component
    @RabbitListener(queues = RabbitConfig.SEND_MESSAGE_QUEUE)
    public class MessageReceiver {
    
        @RabbitHandler
        @Transactional
        public void handler(String msg) {
            System.out.println("Receiver  : " + msg);
        }
    }
    
    /**
     * @author: 谢飞
     */
    @Component
    @RabbitListener(queues = "test-queue")
    public class TestReceiver {
    
        @RabbitHandler
        public void process(User msg) {
            System.out.println("Receiver  : " + msg);
        }
    }

测试:

package com.zoo.rabbitmq.controller;
    
    
    import com.zoo.rabbitmq.config.RabbitConfig;
    import com.zoo.rabbitmq.entity.User;
    import com.zoo.rabbitmq.service.RabbitmqService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author: 谢飞
     */
    @RestController
    public class TestController {
    
        @Autowired
        private RabbitmqService rabbitmqService;
    
        @GetMapping("test")
        public void test() {
            Map<String, Object> map = new HashMap<>();
    
            map.put("userId", 1);
            rabbitmqService.send("test-queue", User.builder().id(1L).name("谢飞").build());
        }
    
        @GetMapping("exchange")
        public void exchange() {
            rabbitmqService.send(RabbitConfig.TOPIC_EXCHANGE,RabbitConfig.SEND_MAIL_QUEUE, "hello world!");
        }
    
        @GetMapping("fanout")
        public void fanout() {
            //与binding-key和routing-key无关，将接受到的消息分发给有绑定关系的所有队列（不论binding-key和routing-key是什么）
            rabbitmqService.send(RabbitConfig.FANOUT_EXCHANGE,"abcd.ee", "hello world!");
        }
    }

(完)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoNTU0MTEyMDc1_size_16_color_FFFFFF_t_70]: /images/20230531/6718439381124816ae899515672993bb.png