RabbitMQ交换器类型（Direct、Topic、Fanout、Headers）与消息的发送/接收模式

忘是亡心i 2021-09-07 06:03 439阅读 0赞

# 1、RabbitMQ 交换器类型 #

RabbitMQ 常用的交换器类型有 Direct、Topic、Fanout、Headers 这四种。AMQP 协议里还提到另外两种类型：System 和自定义，这里不予描述。对于这四种类型下面一一阐述。

## 1.1 Direct 类型 ##

Direct 类型的交换器由路由规则很简单，它会把消息路由到那些 BindingKey 和 RoutingKey 完全匹配的队列中。

Direct Exchange 是 RabbitMQ 默认的交换器模式，也是最简单的模式。它根据 RoutingKey 完全匹配去寻找队列。

## 1.2 Topic 类型 ##

上面讲到 Direct 类型的交换器由规则是完全匹配 BindingKey 和 RoutingKey，但是这种严格的匹配方式在很多情况下不能满足实际业务的需求。Topic 类型的交换器在匹配规则上进行了扩展，它与 Direct 类型的交换器相似，也是将消息路由到 BindingKey 和 RoutingKey 相匹配的队列中，但这里的匹配规则有些不同，它约定：

（1）RoutingKey 为一个点号 “.” 分隔的字符串（被点号 “.” 分隔开的每一段独立的字符串称为一个单词），如：com.rabbitmq.client、java.util.concurrent、com.hidden.client；

（2）BindingKey 和 RoutingKey 一样也是点号 “.” 分隔的字符串；

（3）BindingKey 中可以存在两种特殊字符串星号 “\*” 和井号 “\#”，用于做模糊匹配，其中星号 “\*” 用于匹配一个单词，井号 “\#”用于匹配多个规则单词（0个或者多个单词）；

## 1.3 Fanout 类型 ##

消息广播的模式，即将消息广播到所有绑定到它的队列中，而不考虑 RoutingKey 的值（不管路键或是路由模式）。如果设置了 RoutingKey ，则 RoutingKey 依然被忽略。

## 1.4 Headers 类型 ##

Headers 类型的交换器不依赖于路由键的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。在绑定队列和交换器时制定一组键值对，当发送消息到交换器时，RabbitMQ 会获取到该消息的 headers（也是一个键值对的形式），对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果完全匹配则消息会路由到该队列，否则不会路由到该队列。Headers 类型的交换器性能会很差，而且不实用，基本上不会看到它的存在。

# 2、RabbitMQ 消息的发送/接收模式 #

下面将通过示例来讲解 RabbitMQ 的发送/接收模式。首先需要创建两个 SpringBoot 项目并整合 RabbitMQ 客户端。

**（1）创建第一个 SpringBoot 项目（ rabbitmq-provider 消息推送项目），项目结构如下图：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70][]

在pom.xml配置信息文件中，添加相关依赖文件：

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

在 application.yml 配置文件中配置 RabbitMQ 服务：

spring:
      # 项目名称
      application:
        name: rabbitmq-provider
      # RabbitMQ服务配置
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest

**（2）创建第二个 SpringBoot 项目（ rabbitmq-consumer 消息接收项目），项目结构如下图：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1][]

在pom.xml配置信息文件中，添加相关依赖文件：

在 application.yml 配置文件中配置 RabbitMQ 服务：

spring:
      # 项目名称
      application:
        name: rabbitmq-consumer
      # RabbitMQ服务配置
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest

## 2.1 实现发送和接收队列 ##

使用 Direct 模式，实现消息的发送和接收队列。

**（1）配置队列**

在 rabbitmq-provider（消息推送项目）中，配置队列名称，并将队列交由 IoC 管理，代码如下：

package com.pjb.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * RabbitMQ配置类
     * @author pan_junbiao
     **/
    @Configuration
    public class DirectRabbitMqConfig
    {
        public static final String DIRECT_QUEUE_NAME = "direct_queue_name"; //队列名称
        public static final String DIRECT_EXCHANGE_NAME = "direct_exchange_name"; //交换器名称
        public static final String DIRECT_ROUTING_KEY = "direct_routing_key"; //路由键
    
        /**
         * 队列
         */
        @Bean
        public Queue directQueue()
        {
            /**
             * 创建队列，参数说明：
             * String name：队列名称。
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。
             * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             * 当没有生产者或者消费者使用此队列，该队列会自动删除。
             * Map<String, Object> arguments：设置队列的其他一些参数。
             */
            return new Queue(DIRECT_QUEUE_NAME, true, false, false, null);
        }
    
        /**
         * Direct交换器
         */
        @Bean
        public DirectExchange directExchange()
        {
            /**
             * 创建交换器，参数说明：
             * String name：交换器名称
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             */
            return new DirectExchange(DIRECT_EXCHANGE_NAME, true, false);
        }
    
        /**
         * 绑定
         */
        @Bean
        Binding bindingDirect(DirectExchange directExchange,Queue directQueue)
        {
            //将队列和交换机绑定, 并设置用于匹配键：routingKey
            return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);
        }
    }

**（2）创建发送者**

在 rabbitmq-provider（消息推送项目）中，创建发送者，利用 rabbitTemplate.convertAndSend() 方法发送消息，代码如下：

package com.pjb;
    
    import com.pjb.config.DirectRabbitMqConfig;
    import org.junit.jupiter.api.Test;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * RabbitMQ测试类
     * @author pan_junbiao
     **/
    @SpringBootTest
    public class DirectRabbitMqTest
    {
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        @Test
        public void sendDirectMessage()
        {
            //创建用户信息
            Map<String, Object> userMap = new HashMap<>();
            userMap.put("userId", "1");
            userMap.put("userName", "pan_junbiao的博客");
            userMap.put("blogUrl", "https://blog.csdn.net/pan_junbiao");
            userMap.put("userRemark", "您好，欢迎访问 pan_junbiao的博客");
    
            /**
             * 发送消息，参数说明：
             * String exchange：交换器名称。
             * String routingKey：路由键。
             * Object object：发送内容。
             */
            rabbitTemplate.convertAndSend(DirectRabbitMqConfig.DIRECT_EXCHANGE_NAME, DirectRabbitMqConfig.DIRECT_ROUTING_KEY, userMap);
            System.out.println("消息发送成功！");
        }
    }

**（3）创建接收者**

在 rabbitmq-consumer（消息接收项目）中，创建创建接收者，注意，发送者和接收者的 Queue 名称必须一致，否则不能接收消息。代码如下：

package com.pjb.receiver;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    import java.util.Map;
    
    /**
     * 接收者
     * @author pan_junbiao
     **/
    @Component
    @RabbitListener(queues="direct_queue_name")
    public class DirectReceiver
    {
        @RabbitHandler
        public void process(Map message)
        {
            System.out.println("接收者收到消息：");
            System.out.println("用户编号：" + message.get("userId"));
            System.out.println("用户名称：" + message.get("userName"));
            System.out.println("博客地址：" + message.get("blogUrl"));
            System.out.println("博客信息：" + message.get("userRemark"));
        }
    }

运行 rabbitmq-provider（消息推送项目）中的发送方法，然后运行 rabbitmq-consumer（消息接收项目），将从控制台中看到执行结果。

**执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2][]

## 2.2 实现用接收器接收多个主题 ##

Topic 模式是 RabbitMQ 中最灵活的一种模式，可以根据 RoutingKey 自由地绑定不同的队列。

**（1）配置 Topic 模式**

在 rabbitmq-provider（消息推送项目）中，配置处理消息的队列，代码如下：

package com.pjb.config;
    
    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;
    
    /**
     * RabbitMQ配置类
     * @author pan_junbiao
     **/
    @Configuration
    public class TopicRabbitMqConfig
    {
        public static final String TOPIC_QUEUE_NAME_A = "topic_queue_name_a"; //队列名称A
        public static final String TOPIC_QUEUE_NAME_B = "topic_queue_name_b"; //队列名称B
        public static final String TOPIC_EXCHANGE_NAME = "topic_exchange_name"; //交换器名称
    
        /**
         * 队列A
         */
        @Bean
        public Queue topicQueueA()
        {
            /**
             * 创建队列，参数说明：
             * String name：队列名称。
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。
             * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             * 当没有生产者或者消费者使用此队列，该队列会自动删除。
             * Map<String, Object> arguments：设置队列的其他一些参数。
             */
            return new Queue(TOPIC_QUEUE_NAME_A, true);
        }
    
        /**
         * 队列B
         */
        @Bean
        public Queue topicQueueB()
        {
            /**
             * 创建队列，参数说明：
             * String name：队列名称。
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。
             * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             * 当没有生产者或者消费者使用此队列，该队列会自动删除。
             * Map<String, Object> arguments：设置队列的其他一些参数。
             */
            return new Queue(TOPIC_QUEUE_NAME_B, true);
        }
    
        /**
         * Topic交换器
         */
        @Bean
        TopicExchange exchange()
        {
            /**
             * 创建交换器，参数说明：
             * String name：交换器名称
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             */
            return new TopicExchange(TOPIC_EXCHANGE_NAME, true, false);
        }
    
        /**
         * 绑定1
         */
        @Bean
        Binding bindingExchangeMessage1(Queue topicQueueA, TopicExchange exchange)
        {
            //将队列和交换机绑定, 并设置用于匹配键：routingKey
            return BindingBuilder.bind(topicQueueA).to(exchange).with("topic.routingKey.a");
        }
    
        /**
         * 绑定2
         */
        @Bean
        Binding bindingExchangeMessage2(Queue topicQueueB, TopicExchange exchange)
        {
            //将队列和交换机绑定, 并设置用于匹配键：routingKey
            return BindingBuilder.bind(topicQueueB).to(exchange).with("topic.routingKey.#");
        }
    }

**（2）创建发送者**

在 rabbitmq-provider（消息推送项目）中，创建发送者，通过发送不同的 RoutingKey 来测试效果。代码如下：

package com.pjb;
    
    import com.pjb.config.TopicRabbitMqConfig;
    import org.junit.jupiter.api.Test;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    /**
     * RabbitMQ测试类
     * @author pan_junbiao
     **/
    @SpringBootTest
    public class TopicRabbitMqTest
    {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void sendTopicMessage1()
        {
            String context = "pan_junbiao的博客_01";
            System.out.println("Sender：" + context);
            this.rabbitTemplate.convertAndSend(TopicRabbitMqConfig.TOPIC_EXCHANGE_NAME,"topic.routingKey.cc",context);
        }
    
        @Test
        public void sendTopicMessage2()
        {
            String context = "pan_junbiao的博客_02";
            System.out.println("Sender：" + context);
            this.rabbitTemplate.convertAndSend(TopicRabbitMqConfig.TOPIC_EXCHANGE_NAME,"topic.routingKey.a",context);
        }
    
        @Test
        public void sendTopicMessage3()
        {
            String context = "pan_junbiao的博客_03";
            System.out.println("Sender：" + context);
            this.rabbitTemplate.convertAndSend(TopicRabbitMqConfig.TOPIC_EXCHANGE_NAME,"topic.routingKey.pjb",context);
        }
    }

**（3）创建接收者**

在 rabbitmq-consumer（消息接收项目）中，创建接收者A。代码如下：

在 rabbitmq-consumer（消息接收项目）中，创建接收者B。代码如下：

分别运行 rabbitmq-provider（消息推送项目）中的发送方法，然后运行 rabbitmq-consumer（消息接收项目），将从控制台中看到执行结果。

**执行结果1：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3][]

**执行结果2：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 4][]

**执行结果3：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 5][]

## 2.3 实现广播模式 ##

Fanout 类型的交换器可以实现广播模式。在该模式下，绑定了交换器的所有队列都能接收到这个消息。

**（1）配置 Fanout 类型**

在 rabbitmq-provider（消息推送项目）中，配置广播模式的对象。代码如下：

**（2）创建发送者**

在 rabbitmq-provider（消息推送项目）中，创建发送者，代码如下：

package com.pjb;
    
    import com.pjb.config.FanoutRabbitMqConfig;
    import org.junit.jupiter.api.Test;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    /**
     * RabbitMQ测试类
     * @author pan_junbiao
     **/
    @SpringBootTest
    public class FanoutRabbitMqTest
    {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void sendFanoutMessage()
        {
            String context = "您好，欢迎访问 pan_junbiao的博客";
            System.out.println("Sender：" + context);
            this.rabbitTemplate.convertAndSend(FanoutRabbitMqConfig.FANOUT_EXCHANGE_NAME, "", context);
        }
    }

**（3）创建接收者**

在 rabbitmq-consumer（消息接收项目）中，创建接收者A。代码如下：

在 rabbitmq-consumer（消息接收项目）中，创建接收者B。代码如下：

在 rabbitmq-consumer（消息接收项目）中，创建接收者C。代码如下：

运行 rabbitmq-provider（消息推送项目）中的发送方法，然后运行 rabbitmq-consumer（消息接收项目），将从控制台中看到执行结果。

**执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 6][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70]: /images/20210724/484404e93c194059bbc2cf0ad2d52ab2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1]: /images/20210724/be2dff7db5444d94bface245e0a9e91b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2]: /images/20210724/08526d04fedb4f5f8b52d5224a036f4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3]: /images/20210724/eac386e0c05d4700a3e834f573c95c6a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 4]: /images/20210724/f8afcbbedbc445b7b350500ecdf986a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 5]: /images/20210724/849c13b01b544a4d91a8e49e95bba366.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 6]: /images/20210724/9c636a49bc5b4ec3bcf176262068d857.png