RabbitMQ的交换器Exchange之fanout交换器(广播)
- 2019 年 11 月 5 日
- 笔记
1、Fanout交换器(广播)，以广播的模式进行消息的传递。广播模式一定没有路由键的存在，将消息从路由器发送到所有绑定的队列中去（即消息会发送到所有和指定路由器绑定的队列中去）。
项目的结构如下所示：
2、由于使用的是SpringBoot项目结合Maven项目构建的，pom.xml的配置文件，如下所示，生产者和消费者的配置文件一致，这里只贴一份了。
1 <?xml version="1.0" encoding="UTF-8"?> 2 <project xmlns="http://maven.apache.org/POM/4.0.0" 3 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 4 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 5 https://maven.apache.org/xsd/maven-4.0.0.xsd"> 6 <modelVersion>4.0.0</modelVersion> 7 <parent> 8 <groupId>org.springframework.boot</groupId> 9 <artifactId>spring-boot-starter-parent</artifactId> 10 <version>2.1.1.RELEASE</version> 11 <relativePath /> <!-- lookup parent from repository --> 12 </parent> 13 <groupId>com.bie</groupId> 14 <artifactId>rabbitmq-fanout-provider</artifactId> 15 <version>0.0.1-SNAPSHOT</version> 16 <name>rabbitmq-fanout-provider</name> 17 <description>Demo project for Spring Boot</description> 18 19 <properties> 20 <java.version>1.8</java.version> 21 </properties> 22 23 <dependencies> 24 <dependency> 25 <groupId>org.springframework.boot</groupId> 26 <artifactId>spring-boot-starter</artifactId> 27 </dependency> 28 <dependency> 29 <groupId>org.springframework.boot</groupId> 30 <artifactId>spring-boot-starter-web</artifactId> 31 </dependency> 32 <dependency> 33 <groupId>org.springframework.boot</groupId> 34 <artifactId>spring-boot-starter-test</artifactId> 35 <scope>test</scope> 36 </dependency> 37 <dependency> 38 <groupId>org.springframework.boot</groupId> 39 <artifactId>spring-boot-starter-amqp</artifactId> 40 </dependency> 41 </dependencies> 42 43 <build> 44 <plugins> 45 <plugin> 46 <groupId>org.springframework.boot</groupId> 47 <artifactId>spring-boot-maven-plugin</artifactId> 48 </plugin> 49 </plugins> 50 </build> 51 52 </project>
3、配置好pom.xml配置文件，就可以进行开发了，这里先约束一下配置文件，体现一下SpringBoot的魔力，约定大于配置。
1 # 给当前项目起名称. 2 spring.application.name=rabbitmq-fanout-provider 3 4 # 配置端口号 5 server.port=8081 6 7 # 配置rabbitmq的参数. 8 # rabbitmq服务器的ip地址. 9 spring.rabbitmq.host=192.168.110.133 10 # rabbitmq的端口号5672,区别于浏览器访问界面的15672端口号. 11 spring.rabbitmq.port=5672 12 # rabbitmq的账号. 13 spring.rabbitmq.username=guest 14 # rabbitmq的密码. 15 spring.rabbitmq.password=guest 16 17 # 设置交换器的名称,方便修改. 18 # 生产者和消费者的交换器的名称是一致的,这样生产者生产的消息发送到交换器,消费者可以从这个交换器中消费. 19 rabbitmq.config.exchange=order.exchange.fanout
生产者模拟一个服务，订单服务。使用广播模式，将相同的消息发送到sms短信服务、push服务中去，注意点，生产者和消费者使用相同的路由器。消费者的路由器和队列进行绑定，就可以将订单服务产生的消息发送到sms短信服务、push服务中去。
1 package com.example.bie.provider; 2 3 import org.springframework.amqp.core.AmqpTemplate; 4 import org.springframework.beans.factory.annotation.Autowired; 5 import org.springframework.beans.factory.annotation.Value; 6 import org.springframework.stereotype.Component; 7 8 /** 9 * 10 * @author biehl 11 * 12 * 生产者,生产消息同样需要知道向那个交换器Exchange发送消息的. 13 * 14 * 这里使用的交换器类型使用的是topic主题模式,根据规则匹配。 15 * 16 */ 17 @Component 18 public class RabbitMqUserLogProduce { 19 20 @Autowired 21 private AmqpTemplate rabbitmqAmqpTemplate; 22 23 // 交换器的名称Exchange 24 @Value(value = "${rabbitmq.config.exchange}") 25 private String exchange; 26 27 /** 28 * 发送消息的方法 29 * 30 * @param msg 31 */ 32 public void producer(String msg) { 33 // 向消息队列发送消息 34 // 参数1,交换器的名称 35 // 参数2,路由键,广播模式没有路由键,给定空串即可. 36 // 参数3,消息 37 this.rabbitmqAmqpTemplate.convertAndSend(this.exchange, "", "广播模式: " + msg); 38 } 39 40 }
这里使用web工程，浏览器访问调用，方便测试。你也可以使用单元测试的方法。
1 package com.example.bie.controller; 2 3 import org.springframework.beans.factory.annotation.Autowired; 4 import org.springframework.stereotype.Controller; 5 import org.springframework.web.bind.annotation.RequestMapping; 6 import org.springframework.web.bind.annotation.ResponseBody; 7 8 import com.example.bie.provider.RabbitMqUserLogProduce; 9 10 /** 11 * 12 * @author biehl 13 * 14 */ 15 @Controller 16 public class RabbitmqController { 17 18 @Autowired 19 private RabbitMqUserLogProduce rabbitMqUserLogProduce; 20 21 @RequestMapping(value = "/smsPush") 22 @ResponseBody 23 public String rabbitmqSendUserLogInfoMessage() { 24 String msg = "生产者===>生者的消息message: "; 25 for (int i = 0; i < 50000; i++) { 26 rabbitMqUserLogProduce.producer(msg + i); 27 } 28 return "生产===> 消息message ===> success!!!"; 29 } 30 31 }
生产者的启动类如下所示：
1 package com.example; 2 3 import org.springframework.boot.SpringApplication; 4 import org.springframework.boot.autoconfigure.SpringBootApplication; 5 6 @SpringBootApplication 7 public class RabbitmqProducerApplication { 8 9 public static void main(String[] args) { 10 SpringApplication.run(RabbitmqProducerApplication.class, args); 11 } 12 13 }
4、生产者开发完毕就可以进行消费者的开发，也是先约束一下配置文件application.properties。
1 # 给当前项目起名称. 2 spring.application.name=rabbitmq-fanout-consumer 3 4 # 配置端口号 5 server.port=8080 6 7 # 配置rabbitmq的参数. 8 # rabbitmq服务器的ip地址. 9 spring.rabbitmq.host=192.168.110.133 10 # rabbitmq的端口号5672,区别于浏览器访问界面的15672端口号. 11 spring.rabbitmq.port=5672 12 # rabbitmq的账号. 13 spring.rabbitmq.username=guest 14 # rabbitmq的密码. 15 spring.rabbitmq.password=guest 16 17 # 设置交换器的名称,方便修改. 18 # 路由键是将交换器和队列进行绑定的,队列通过路由键绑定到交换器. 19 rabbitmq.config.exchange=order.exchange.fanout 20 21 # sms短信服务的队列名称. 22 rabbitmq.config.queue.sms=order.sms.queue 23 24 # push服务的队列名称. 25 rabbitmq.config.queue.push=order.push.queue
约束好配置文件就可以进行消费者的开发了，这里是将订单服务产生的消息，以广播模式发送给sms短信服务、push服务中去。sms短信服务、push服务接收到的消息一致。广播模式一定不用配置routing-key路由键。
1 package com.example.bie.consumer; 2 3 import org.springframework.amqp.core.ExchangeTypes; 4 import org.springframework.amqp.rabbit.annotation.Exchange; 5 import org.springframework.amqp.rabbit.annotation.Queue; 6 import org.springframework.amqp.rabbit.annotation.QueueBinding; 7 import org.springframework.amqp.rabbit.annotation.RabbitHandler; 8 import org.springframework.amqp.rabbit.annotation.RabbitListener; 9 import org.springframework.stereotype.Component; 10 11 /** 12 * 13 * @author biehl 14 * 15 * 消息接收者 16 * 17 * 1、@RabbitListener bindings:绑定队列 18 * 19 * 2、@QueueBinding 20 * value:绑定队列的名称、exchange:配置交换器、key:路由键routing-key绑定队列和交换器 21 * 22 * 3、@Queue value:配置队列名称、autoDelete:是否是一个可删除的临时队列 23 * 24 * 4、@Exchange value:为交换器起个名称、type:指定具体的交换器类型 25 * 26 * 27 */ 28 @Component 29 @RabbitListener(bindings = @QueueBinding( 30 31 value = @Queue(value = "${rabbitmq.config.queue.sms}", autoDelete = "true"), 32 // 广播模式Fanout不需要路由键 33 exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.FANOUT) 34 35 )) 36 public class SmsConsumer { 37 38 /** 39 * 接收消息的方法,采用消息队列监听机制. 40 * 41 * @RabbitHandler意思是将注解@RabbitListener配置到类上面 42 * 43 * @RabbitHandler是指定这个方法可以进行消息的接收并且消费. 44 * 45 * @param msg 46 */ 47 @RabbitHandler 48 public void consumer(String msg) { 49 // 打印消息 50 System.out.println("sms短信服务消费者===>消费: " + msg); 51 } 52 53 }
1 package com.example.bie.consumer; 2 3 import org.springframework.amqp.core.ExchangeTypes; 4 import org.springframework.amqp.rabbit.annotation.Exchange; 5 import org.springframework.amqp.rabbit.annotation.Queue; 6 import org.springframework.amqp.rabbit.annotation.QueueBinding; 7 import org.springframework.amqp.rabbit.annotation.RabbitHandler; 8 import org.springframework.amqp.rabbit.annotation.RabbitListener; 9 import org.springframework.stereotype.Component; 10 11 /** 12 * 13 * @author biehl 14 * 15 * 消息接收者 16 * 17 * 1、@RabbitListener bindings:绑定队列 18 * 19 * 2、@QueueBinding 20 * value:绑定队列的名称、exchange:配置交换器、key:路由键routing-key绑定队列和交换器 21 * 22 * 3、@Queue value:配置队列名称、autoDelete:是否是一个可删除的临时队列 23 * 24 * 4、@Exchange value:为交换器起个名称、type:指定具体的交换器类型 25 * 26 * 27 */ 28 @Component 29 @RabbitListener(bindings = @QueueBinding( 30 31 value = @Queue(value = "${rabbitmq.config.queue.push}", autoDelete = "true"), 32 // 广播模式Fanout不需要路由键 33 exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.FANOUT) 34 35 )) 36 public class PushConsumer { 37 38 /** 39 * 接收消息的方法,采用消息队列监听机制. 40 * 41 * @RabbitHandler意思是将注解@RabbitListener配置到类上面 42 * 43 * @RabbitHandler是指定这个方法可以进行消息的接收并且消费. 44 * 45 * @param msg 46 */ 47 @RabbitHandler 48 public void consumer(String msg) { 49 // 打印消息 50 System.out.println("push短信服务消费者===>消费: " + msg); 51 } 52 53 }
消费者的启动类，如下所示：
1 package com.example; 2 3 import org.springframework.boot.SpringApplication; 4 import org.springframework.boot.autoconfigure.SpringBootApplication; 5 6 @SpringBootApplication 7 public class RabbitmqConsumerApplication { 8 9 public static void main(String[] args) { 10 SpringApplication.run(RabbitmqConsumerApplication.class, args); 11 } 12 13 }
5、运行效果如下所示：
