RabbitMQ实战之三种交换机（Direct、Topic和Fanout）

我就是我 2022-12-11 09:30 200阅读 0赞

# 一、消息推送到接收的流程图 #

首先先介绍一个简单的一个消息推送到接收的流程，提供一个简单的图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center]  
黄色的圈圈就是我们的消息推送服务，将消息推送到 中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系（后面会详细讲）将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。

常用的交换机有以下三种，因为消费者是从队列获取信息的，队列是绑定交换机的（一般），所以对应的消息推送/接收模式也会有以下几种：

## 1.Direct Exchange ##

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

## 2.Fanout Exchange ##

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

## 3.Topic Exchange ##

主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。  
简单地介绍下规则：  
\*(星号) 用来表示一个单词 (必须出现的)  
\#(井号) 用来表示任意数量（零个或多个）单词  
通配的绑定键是跟队列进行绑定的，举例如下：  
队列Q1 绑定键为 *.TT.*  
队列Q2绑定键为 TT.\#  
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；  
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；

# 二、Direct Exchange #

直连交换机中，每对交换机和队列之间只能通过唯一一个路由键来绑定（当然，每个交换机可绑定多个队列，每个队列也可绑定多个交换机）。

## 1.DirectConfig.java ##

package com.springboot.rabbit.demo.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    @Configuration
    public class DirectConfig { 
        //用@Bean注解配置好需要注册到RabbitMQ服务器的消息队列。项目启动时，这些队列就会被注册。
    	//队列 起名：com.direct.queue.1
        @Bean
        public Queue directQueue1() { 
            // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
            // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
            // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
            // 比如return new Queue("TestDirectQueue",true,true,false);
            return new Queue("com.direct.queue.1");
        }
    
        @Bean
        public Queue directQueue2() { 
            return new Queue("com.direct.queue.2");
        }
    
        @Bean
        public Queue directQueue3() { 
            return new Queue("com.direct.queue.3");
        }
    
        //Direct交换机 起名：com.direct.exchange
        @Bean
        public DirectExchange directExchange() { 
            return new DirectExchange("com.direct.exchange");
        }
        
        //绑定 将队列和交换机绑定, 并设置用于匹配键：com.direct.routingKey.1
        @Bean
        public Binding bindQueue1() { 
            return BindingBuilder.bind(directQueue1()).to(directExchange()).with("com.direct.routingKey.1");
        }
    
        @Bean
        public Binding bindQueue2() { 
            return BindingBuilder.bind(directQueue2()).to(directExchange()).with("com.direct.routingKey.2");
        }
    
        @Bean
        public Binding bindQueue3() { 
            return BindingBuilder.bind(directQueue3()).to(directExchange()).with("com.direct.routingKey.3");
        }
    }

## 2.生产者DirectProducer.java ##

package com.springboot.rabbit.demo.producer;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class DirectProducer { 
    	
    	//使用RabbitTemplate,这提供了接收/发送等等方法
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void send1(String msg) { 
        	//将消息携带绑定键值：com.direct.routingKey.1 发送到交换机com.direct.exchange
            rabbitTemplate.convertAndSend("com.direct.exchange", "com.direct.routingKey.1", msg);
        }
    
        public void send2(String msg) { 
            rabbitTemplate.convertAndSend("com.direct.exchange", "com.direct.routingKey.2", msg);
        }
    
        public void send3(String msg) { 
            rabbitTemplate.convertAndSend("com.direct.exchange", "com.direct.routingKey.3", msg);
        }
    }

## 3.消费者DirectConsumer.java ##

package com.springboot.rabbit.demo.consumer;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = { "com.direct.queue.1", "com.direct.queue.2", "com.direct.queue.3"})
    //监听的队列名称 com.direct.queue.1\2\3
    public class DirectConsumer { 
        @RabbitHandler
        public void process(String msg) { 
           System.out.println(msg);
        }
    }

## 4.业务控制层DirectController.java ##

package com.springboot.rabbit.demo.controller;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import com.springboot.rabbit.demo.producer.DirectProducer;
    @RestController
    @RequestMapping("direct")
    public class DirectController { 
        @Autowired
        private DirectProducer directProducer;
    
        @RequestMapping("1")
        public void produce1(String msg) { 
            directProducer.send1(msg);
        }
    
        @RequestMapping("2")
        public void produce2(String msg) { 
            directProducer.send2(msg);
        }
    
        @RequestMapping("3")
        public void produce3(String msg) { 
            directProducer.send3(msg);
        }
    }

## 5.测试结果 ##

启动RabbitService，然后运行RabbitDemoApplication.java主类，在浏览器中访问如下链接，进行测试：  
![在这里插入图片描述][2020093009493387.png_pic_center]  
![在这里插入图片描述][20200930094949934.png_pic_center]  
详细信息在http://localhost:15672/\#/中即可查看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center 1]

# 三、Fanout Exchange #

扇出交换机无需绑定路由，只要是生产者发送到扇出交换机上的消息全部都会被消费者监听到并消费。所以，随意发送任意路由都可以。

## 1.FanoutConfig.java ##

## 2.生产者FanoutProducer.java ##

## 3.消费者FanoutConsumer.java ##

package com.springboot.rabbit.demo.consumer;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = { "com.fanout.queue.1", "com.fanout.queue.2", "com.fanout.queue.3"})
    public class FanoutConsumer { 
        @RabbitHandler
        public void process(String msg) { 
        	System.out.println(msg);
        }
    }

## 4.业务控制层FanoutController.java ##

## 5.测试结果 ##

测试结果如下：  
![在这里插入图片描述][2020093010404578.png_pic_center]  
![在这里插入图片描述][20200930104112363.png_pic_center]

# 四、Topic Exchange #

Topic Exchange转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。  
在这种交换机模式下：  
（1）路由键（Routing Key）命名必须为一串字符，用句号（.） 隔开，比如 jake.topic.queue。  
（2）队列和交换机通过路由键绑定。

## 1.TopicConfig.java ##

package com.springboot.rabbit.demo.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;
    @Configuration
    public class TopicConfig { 
    
        @Bean
        public Queue topicQueueAccurate() { 
            return new Queue("com.topic.queue.accurate");
        }
    
        @Bean
        public Queue topicQueueSingle() { 
            return new Queue("com.topic.queue.single");
        }
    
        @Bean
        public Queue topicQueueAny() { 
            return new Queue("com.topic.queue.any");
        }
    
        @Bean
        public TopicExchange topicExchange() { 
            return new TopicExchange("com.topic.exchange");
        }
    
        @Bean
        public Binding bindWithAccurateMatcher() { 
            return BindingBuilder.bind(topicQueueAccurate()).to(topicExchange()).
                    with("com.topic.routingKey.accurate");
        }
    
        @Bean
        public Binding bindWithSingleWordMatcher() { 
        	// * (星号) 用来表示一个单词 (必须出现的)
            return BindingBuilder.bind(topicQueueSingle()).to(topicExchange()).
                    with("com.topic.routingKey.*");
        }
    
        @Bean
        public Binding bindWithAnyWordMatcher() { 
        	// # (井号) 用来表示任意数量（零个或多个）单词
            return BindingBuilder.bind(topicQueueAny()).to(topicExchange()).
                    with("com.topic.#");
        }
    
    }

## 2.生产者TopicProducer.java ##

package com.springboot.rabbit.demo.producer;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class TopicProducer { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendAccurate(String msg) { 
            rabbitTemplate.convertAndSend("com.topic.exchange", "com.topic.routingKey.accurate", msg);
        }
    
        public void sendSingle(String msg) { 
            rabbitTemplate.convertAndSend("com.topic.exchange", "com.topic.routingKey.only-one-word", msg);
        }
    
        public void sendAny(String msg) { 
            rabbitTemplate.convertAndSend("com.topic.exchange", "com.topic.routingKey.as.much.as.you.want", msg);
        }
    }

## 3.消费者 ##

精确匹配TopicAccurateMatchConsumer.java

package com.springboot.rabbit.demo.consumer;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    @RabbitListener(queues = { "com.topic.queue.accurate"})
    public class TopicAccurateMatchConsumer { 
        @RabbitHandler
        public void process(String msg) { 
        	System.out.println(msg);
        }
    }

匹配所有TopicAnyMatchConsumer.java

package com.springboot.rabbit.demo.consumer;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    @RabbitListener(queues = { "com.topic.queue.any"})
    public class TopicAnyMatchConsumer { 
        @RabbitHandler
        public void process(String msg) { 
        	System.out.println(msg);
        }
    }

匹配单一队列TopicSingleMatchConsumer.java

package com.springboot.rabbit.demo.consumer;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    @RabbitListener(queues = { "com.topic.queue.single"})
    public class TopicSingleMatchConsumer { 
        @RabbitHandler
        public void process(String msg) { 
        	System.out.println(msg);
        }
    }

## 4.业务控制层FanoutController.java ##

package com.springboot.rabbit.demo.controller;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import com.springboot.rabbit.demo.producer.TopicProducer;
    @RestController
    @RequestMapping("topic")
    public class TopicController { 
        @Autowired
        private TopicProducer topicProducer;
    
        @RequestMapping("accurate")
        public void accurate(String msg) { 
            topicProducer.sendAccurate(msg);
        }
    
        @RequestMapping("single")
        public void single(String msg) { 
            topicProducer.sendSingle(msg);
        }
    
        @RequestMapping("any")
        public void any(String msg) { 
            topicProducer.sendAny(msg);
        }
    }

## 5.测试结果 ##

精确匹配测试结果：  
![在这里插入图片描述][20200930105107580.png_pic_center]  
![在这里插入图片描述][20200930105359466.png_pic_center]  
任意匹配测试结果：  
![在这里插入图片描述][20200930105443614.png_pic_center]  
![在这里插入图片描述][20200930105252537.png_pic_center]  
单一匹配测试结果：  
![在这里插入图片描述][20200930105532274.png_pic_center]  
![在这里插入图片描述][2020093010552464.png_pic_center]

# 五、项目结构 #

## 1.application.yml配置 ##

目的是自由切换不同的yml配置文件，切换项目环境（dev、prod、test）等。  
application.yml

spring:
      profiles:
        active: dev

application-dev.yml

server:
      port: 8086

## 2.项目目录结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center 2]  
以上整体项目代码链接如下  
[RabbitMQ][]

代码思路参考了[SpringBoot + RabbitMQ实战之通过代码熟悉三种交换机（Direct、Topic和Fanout）][SpringBoot _ RabbitMQ_Direct_Topic_Fanout]  
特此对这位博主表示感谢！

**参考文章**  
https://blog.csdn.net/qq\_15329947/article/details/86528854  
https://blog.csdn.net/qq\_35387940/article/details/100514134

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/91500e93d81e4a11b791ad05f09b85d7.png
[2020093009493387.png_pic_center]: /images/20221123/854077727df04bc8930d74ba6b907d2a.png
[20200930094949934.png_pic_center]: /images/20221123/a762411e9253465e9af2806681702f84.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/6ebc609171164cda8b3571dad4f0b373.png
[2020093010404578.png_pic_center]: /images/20221123/62f37c2b7c64464da3abc9bd6b72e617.png
[20200930104112363.png_pic_center]: /images/20221123/147e3ed1106b46d6bec4c4b333b3295c.png
[20200930105107580.png_pic_center]: /images/20221123/827f26c8da5e4b29aedac55437b66c66.png
[20200930105359466.png_pic_center]: /images/20221123/5b912fc3afcd4022b62aaffac4d3c3b6.png
[20200930105443614.png_pic_center]: /images/20221123/a7b4360d112648e597e72a54ba3fda81.png
[20200930105252537.png_pic_center]: /images/20221123/0035bc77bc64424d91ed50b209f66587.png
[20200930105532274.png_pic_center]: /images/20221123/5335832c884c435bb4d123e96cf812c6.png
[2020093010552464.png_pic_center]: /images/20221123/15d52559022744f6a69f230b858454a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppa2UxMTIzMQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/15be37ebef5742a3920b7fd0848e0c98.png
[RabbitMQ]: https://gitee.com/codefarmer001/rabbit-mq.git
[SpringBoot _ RabbitMQ_Direct_Topic_Fanout]: https://blog.csdn.net/qq_15329947/article/details/86528854