SpringBoot的RabbitMQ消息队列: 四、第三模式"Publish/Subscribe"

Publish/Subscribe 发布/订阅模式，官方定义是每次把消息发送给多个消费者（Sending messages to many consumers at once）。

它的结构式

一、相关概念

1、RabbitMQ结构

通常我们谈到队列服务, 会有三个概念： 发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

• 左侧 P 代表 生产者，也就是往 RabbitMQ 发消息的程序。
• 中间即是 RabbitMQ，其中包括了 交换机 和 队列。
• 右侧 C 代表 消费者，也就是往 RabbitMQ 拿消息的程序。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

• 虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
• 交换机：Exchange 用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。
  这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
• 绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

 2、交换机(Exchange)

RabbitMQ 的Messaging Model就是Producer并不会直接发送Message到queue。实际上，Producer并不知道它发送的Message是否已经到达queue。

Producer发送的Message实际上是发到了Exchange中。它的功能也很简单：从Producer接收Message，然后投递到queue中。Exchange需要知道如何处理Message，是把它放到那个queue中，还是放到多个queue中？这个rule是通过Exchange 的类型定义的。

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

• Direct：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
• Topic：按规则转发消息（最灵活）
• Headers：设置header attribute参数类型的交换机
• Fanout：转发消息到所有绑定队列

Direct Exchange

Direct Exchange是RabbitMQ默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

第一个 X - Q1 就有一个 binding key，名字为 orange； X - Q2 就有 2 个 binding key，名字为 black 和 green。当消息中的 路由键 和 这个 binding key 对应上的时候，那么就知道了该消息去到哪一个队列中。

Ps：为什么 X 到 Q2 要有 black，green，2个 binding key呢，一个不就行了吗？ - 这个主要是因为可能又有 Q3，而Q3只接受 black 的信息，而Q2不仅接受black 的信息，还接受 green 的信息。

Topic Exchange

Topic Exchange 转发消息主要是根据通配符。 在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

在这种交换机模式下：

• 路由键必须是一串字符，用句号（.） 隔开，比如说 agreements.us，或者 agreements.eu.stockholm 等。
• 路由模式必须包含一个 星号（*），主要用于匹配路由键指定位置的一个单词，比如说，一个路由模式是这样子：agreements..b.*，那么就只能匹配路由键是这样子的：第一个单词是 agreements，第四个单词是 b。 井号（#）就表示相当于一个或者多个单词，例如一个匹配模式是agreements.eu.berlin.#，那么，以agreements.eu.berlin开头的路由键都是可以的。

具体代码发送的时候还是一样，第一个参数表示交换机，第二个参数表示routing key，第三个参数即消息。如下：

[java] view plain copy

1. rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is  RabbitMQ!");  

topic 和 direct 类似, 只是匹配上支持了"模式", 在"点分"的 routing_key 形式中, 可以使用两个通配符:

• *表示一个词.
• #表示零个或多个词.

Headers Exchange

headers 也是根据规则匹配, 相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型.
在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

Fanout Exchange

Fanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing_key会被忽略。

二、Fanout模式实现发布/订阅模式

1、建立工程， 通过http://start.spring.io，建立Fanout工程

2、下载、解压，导入eclipse

3、修改pom.xml,以便于热部署

4、增加日志文件logback.xml

5、修改application.properties

[html] view plain copy

1.  #服务器配置    
2. spring.application.name=rabbitmq-fanout    
3. server.port=9080     
4. #rabbitmq连接参数    
5. spring.rabbitmq.host=localhost    
6. spring.rabbitmq.port=5672    
7. spring.rabbitmq.username=test    
8. spring.rabbitmq.password=123456    

6、配置文件。增加类FanoutRabbitConfig，编制代码为

[html] view plain copy

1. package com.example;  
2.   
3. import org.springframework.amqp.core.Binding;  
4. import org.springframework.amqp.core.BindingBuilder;  
5. import org.springframework.amqp.core.FanoutExchange;  
6. import org.springframework.amqp.core.Queue;  
7. import org.springframework.context.annotation.Bean;  
8. import org.springframework.context.annotation.Configuration;  
9.   
10.   
11. @Configuration  
12. public class FanoutRabbitConfig {  
13.   
14.     @Bean  
15.     public Queue AMessage() {  
16.         return new Queue("fanout.A");  
17.     }  
18.   
19.     @Bean  
20.     public Queue BMessage() {  
21.         return new Queue("fanout.B");  
22.     }  
23.   
24.     @Bean  
25.     public Queue CMessage() {  
26.         return new Queue("fanout.C");  
27.     }  
28.   
29.     @Bean  
30.     FanoutExchange fanoutExchange() {  
31.         return new FanoutExchange("fanoutExchange");  
32.     }  
33.   
34.     @Bean  
35.     Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {  
36.         return BindingBuilder.bind(AMessage).to(fanoutExchange);  
37.     }  
38.   
39.     @Bean  
40.     Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {  
41.         return BindingBuilder.bind(BMessage).to(fanoutExchange);  
42.     }  
43.   
44.     @Bean  
45.     Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {  
46.         return BindingBuilder.bind(CMessage).to(fanoutExchange);  
47.     }  
48.   
49. }  

7、消息发送。增加类FanoutSender，编制代码为

[java] view plain copy

1. package com.example;  
2.   
3. import org.slf4j.Logger;  
4. import org.slf4j.LoggerFactory;  
5. import org.springframework.amqp.core.AmqpTemplate;  
6. import org.springframework.beans.factory.annotation.Autowired;  
7. import org.springframework.stereotype.Component;  
8.   
9. @Component  
10. public class FanoutSender {  
11.   
12.     protected static Logger logger=LoggerFactory.getLogger(FanoutSender.class);   
13.       
14.     @Autowired  
15.     private AmqpTemplate rabbitTemplate;  
16.   
17.     public void send() {  
18.         String context = "hi, fanout msg ";  
19.         logger.debug("Sender : " + context);  
20.         this.rabbitTemplate.convertAndSend("fanoutExchange","", context);  
21.     }  
22.   
23. }  

8、消息接收A.增加类FanoutReceiverA，编制代码为

[java] view plain copy

1. package com.example;  
2.   
3. import org.slf4j.Logger;  
4. import org.slf4j.LoggerFactory;  
5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
7. import org.springframework.stereotype.Component;  
8.   
9. @Component  
10. @RabbitListener(queues = "fanout.A")  
11. public class FanoutReceiverA {  
12.   
13.     protected static Logger logger=LoggerFactory.getLogger(FanoutReceiverA.class);   
14.       
15.     @RabbitHandler  
16.     public void process(String message) {  
17.         logger.debug("fanout Receiver A  : " + message);  
18.     }  
19.   
20. }  

9、消息接收B.增加类FanoutReceiverB，编制代码为

[java] view plain copy

1. package com.example;  
2.   
3. import org.slf4j.Logger;  
4. import org.slf4j.LoggerFactory;  
5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
7. import org.springframework.stereotype.Component;  
8.   
9. @Component  
10. @RabbitListener(queues = "fanout.B")  
11. public class FanoutReceiverB {  
12.   
13.     protected static Logger logger=LoggerFactory.getLogger(FanoutReceiverB.class);   
14.       
15.     @RabbitHandler  
16.     public void process(String message) {  
17.         logger.debug("fanout Receiver B  : " + message);  
18.     }  
19.   
20. }  

10、消息接收C.增加类FanoutReceiverC，编制代码为

[java] view plain copy

1. package com.example;  
2.   
3. import org.slf4j.Logger;  
4. import org.slf4j.LoggerFactory;  
5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
7. import org.springframework.stereotype.Component;  
8.   
9. @Component  
10. @RabbitListener(queues = "fanout.C")  
11. public class FanoutReceiverC {  
12.   
13.     protected static Logger logger=LoggerFactory.getLogger(FanoutReceiverC.class);   
14.       
15.     @RabbitHandler  
16.     public void process(String message) {  
17.         logger.debug("fanout Receiver C  : " + message);  
18.     }  
19.   
20. }  

11、RestController。增加类FanoutController，编制代码为

[java] view plain copy

1. package com.example;  
2.   
3. import org.springframework.beans.factory.annotation.Autowired;  
4. import org.springframework.web.bind.annotation.RequestMapping;  
5. import org.springframework.web.bind.annotation.RestController;  
6.   
7. @RestController  
8. public class FanoutController {  
9.   
10.     @Autowired  
11.     private FanoutSender fanoutSender;  
12.       
13.     @RequestMapping("/send")  
14.     public String send() {  
15.         fanoutSender.send();  
16.         return "send ok";  
17.     }  
18. }  

11、运行工程。在工程所在文件夹打开cmd,输入mvn spring-boot:run

12、在浏览器中分别输入http://localhost:9080/send查看控制台输出 

三个消息接收器都收到消息

13、RebbitMQ Management

14、小结

A、配置文件中，定义一个FanoutExchange类型的交换机，再让三个队列都绑定到这个交换机。

B、消息发送，使用void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;但不指定routingKey。因为FanoutExchange类型的交换机，routingKey不起作用，它向所有的队列发送广播，只要队列绑定到该交换机即接受消息。

C、消息接收，就普通的rabbitmq消息监听类。

所有的秘密完全在配置文件里面。