一、前言
该技术博客是基于SpringAMQP模板进行讲解，不会使用原生的RabbitMQ代码，首先我们先了解一下什么是SpringAMQP
SpringAMQP：是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAMQP的官方地址：https://spring.io/projects/spring-amqp
那么官方提供的五大消息模型都有哪些？
二、准备工作
创建项目：
首先创建一个父工程，名字任意（mq-demo）
然后在父工程下创建两个模块（基于SpringBoot），分别为：publisher、consumer
该项目包括三部分：
- mq-demo：父工程，管理项目依赖
- publisher：消息的发送者
- consumer：消息的消费者
在父工程下导入依赖：
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
编写配置：
spring:
rabbitmq:
host: 192.168.150.101 # 主机名
port: 5672 # 消息队列端口
virtual-host: / # 虚拟主机
username: coderxu # 用户名
password: 123321 # 密码
三、五大消息模型
1、BasicQueue
官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：
- publisher：消息发布者，将消息发送到队列queue
- queue：消息队列，负责接受并缓存消息
- consumer：订阅队列，处理队列中的消息
代码实现：
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
//在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
//队列名称
String queueName = "simple.queue";
//消息内容
String message = "hello, spring amqp!";
//使用RabbitTemplate发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}
==============================================================================
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
//然后在consumer服务中新建一个类SpringRabbitListener，代码如下：
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue") //监听simple.queue队列
//String msg就是从队列中监听到的消息
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring 消费者接收到消息：【" + msg + "】");
}
}
2、WorkQueue
让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用该模型，多个消费者共同处理消息，速度就能大大提高。
代码展示：
/**
* 这次我们循环发送消息，模拟大量消息堆积现象。
* 在publisher服务中的SpringAmqpTest类中添加一个测试方法
* 向队列中不停发送消息，模拟消息堆积
*/
@Test
public void testWorkQueue() throws InterruptedException {
//队列名称
String queueName = "simple.queue";
//发送的消息
String message = "hello, message_";
//循环发送50条消息
for (int i = 0; i < 50; i++) {
// 发送消息
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}
==============================================================================
//要模拟多个消费者绑定同一个队列，我们在consumer服务中的SpringRabbitListener中添加2个新的方法：
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(200);
}
运行之后，可以看到消费者1很快完成了自己的25条消息。
消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的！
那么该如何处理这种情况呢？
# 在spring中有一个简单的配置，可以解决这个问题。
# 我们修改consumer服务的application.yml文件，添加配置：
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每个消费者每次只能获取一条消息，处理完成才能获取下一条消息
附加知识：发布/订阅
在讲解其他余下三个消息模型时，需要先学习了解发布、订阅
发布订阅的模型如图：
此时可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
- Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给
Exchange（交换机） - Consumer：消费者，与以前一样，订阅队列，没有变化
- Queue：消息队列也与以前一样，接收消息、缓存消息。
- Exchange：交换机，一方面，接收生产者发送的消息。另一方面，处理消息。
例如递交给某个特别队列、递交给所有队列、将消息丢弃。
到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：Fanout：广播，将消息交给所有绑定到交换机的队列Direct：定向，把消息交给符合指定routing key 的队列Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
3、Fanout
Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
代码展示：
//在publisher服务的SpringAmqpTest类中添加测试方法：
@Test
public void testFanoutExchange() {
//队列名称
String exchangeName = "itcast.fanout";
//消息
String message = "hello, everyone!";
//发送，第二个参数为RoutingKey，在Direct交换机中会讲解
rabbitTemplate.convertAndSend(exchangeName, "", message);
}
==============================================================================
//在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：
@RabbitListener(bindings = @QueueBinding(
//将队列和交换机进行绑定，然后监听队列fanout.queue1
value = @Queue("fanout.queue1"),
exchange = @Exchange(value = "itcast.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue1(String msg) {
System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
//将队列和交换机进行绑定，然后监听队列fanout.queue2
value = @Queue("fanout.queue2"),
exchange = @Exchange(value = "itcast.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue2(String msg) {
System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}
4、Direct
在Fanout模式中，一条消息，会被所有订阅队列进行消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下：
- 队列与交换机的绑定，不能是任意绑定，而是要指定一个
RoutingKey（路由key） - 消息的发送方在 向 Exchange 发送消息时，也必须指定消息的
RoutingKey。 - Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routingkey完全一致，才会接收到消息
代码展示：
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
}
==============================================================================
@Test
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "itcast.direct";
// 消息
String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
// 发送消息，RoutingKey指定为 red
rabbitTemplate.convertAndSend(exchangeName, "red", message);
}
描述下Direct交换机与Fanout交换机的差异？
- Fanout交换机将消息路由给每一个与之绑定的队列
- Direct交换机根据RoutingKey判断路由给哪个队列
5、Topic
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定Routingkey 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
通配符规则：
#：匹配一个或多个词*：匹配一个词
解释：
- Queue1：绑定的是
china.#，因此凡是以china.开头的routingkey都会被匹配到。包括china.news和china.weather - Queue2：绑定的是
#.news，因此凡是以.news结尾的routingkey都会被匹配。包括china.news和japan.news
代码展示：
@Test
public void testSendTopicExchange() {
// 交换机名称
String exchangeName = "itcast.topic";
// 消息
String message = "我想大声告诉你";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}
==============================================================================
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
}
描述下Direct交换机与Topic交换机的差异？
- Topic交换机接收的消息RoutingKey必须是多个单词，以
**.**分割 - Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表1个或多个词*：代表1个词
