RabbitMQ的交换器Exchange之fanout交换器(广播)

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、运行效果如下所示: