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