Spring AMQP与RabbitMQ深度整合指南：从基础到高级应用（2）
三、SpringAMQP
将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了 AMQP 协议，因此它具备跨语言的特性。任何语言只要遵循 AMQP 协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ 官方提供的 Java 客户端编码相对复杂，一般生产环境下我们更多会结合 Spring 来使用。而 Spring 的官方刚好基于RabbitMQ 提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于 SpringBoot 对其实现了自动装配，使用起来非常方便。
SpringAmqp 的官方地址：
https://spring.io/projects/spring-amqp
SpringAMQP 提供了三个功能：
自动声明队列、交换机及其绑定关系。
基于注解的监听器模式，异步接收消息。
封装了 RabbitTemplate 工具，用于发送消息。
3.1 依赖和配置文件
要进行下面的操作，首先我们需要引入相关依赖：
<!--AMQP依赖，包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
spring: rabbitmq: host: xxxxx # 你的虚拟机IP port: 5672 # 端口 virtual-host: /hmall # 虚拟主机 username: hmall # 用户名 password: 123 # 密码
注意：不论是消息的发送者还是消息的接收者都需要进行上面的配置。
3.2 消息发送和接收：
3.2.1 消息发送：
利用RabbitTemplate实现消息发送，RabbitTemplate 在我们引入依赖和配置完文件启动项目就自动创建了，我们直接进行依赖注入即可。
下面的代码中，没有实现交换机，因为比较复杂，我们下面讲，这里其实就是感受 RabbitTemplate 的使用。队列需要我们在 MQ 控制台手动创建。
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
public void testSimpleQueue() {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
// 发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}3.2.2 消息接收：
@Component
public class SpringRabbitListener {
// 利用RabbitListener来声明要监听的队列信息
// 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring 消费者接收到消息：【" + msg + "】");
}
}3.3 WorkQueues 模型：
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息处理，同一条消息只会被一个消费者处理，这样，消息处理的速度就能大大提高了。
不过在 RabbitMQ 中该模型还存在一个问题：消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。
这显然是不合理的。
改进：
在spring中有一个简单的配置，可以解决这个问题，在 application.yml 文件中添加配置：
spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
这样就能实现能者多劳的效果，充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。
总结：
Work模型的使用：
多个消费者绑定到一个队列，同一条消息只会被一个消费者处理。
通过设置prefetch来控制消费者预取的消息数量。
3.4 交换机类型：
在之前的案例中，都没有交换机，生产者直接发送消息到队列。而一旦引入交换机，消息发送的模式会有很大变化：
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
Publisher：生产者，不再发送消息到队列中，而是发给交换机。
Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
Consumer：消费者，与以前一样，订阅队列，没有变化。
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的类型有四种：
Fanout：广播，将消息交给所有绑定到交换机的队列。
Direct：订阅，基于 RoutingKey（路由 key）发送给订阅了消息的队列。
Topic：通配符订阅，与 Direct 类似，只不过 RoutingKey 可以使用通配符。
Headers：头匹配，基于 MQ 的消息头匹配，用的较少。
我们这里讲解前面三种交换机。
3.4.1 Fanout 交换机：
在广播模式下，消息发送流程是这样的：
总结：
交换机的特点：
接收 publisher 发送的消息，将消息按照规则路由到与之绑定的队列。
不能缓存消息，路由失败，消息丢失。
FanoutExchange 的会将消息路由到每个绑定的队列。
3.4.2 Direct 交换机：
在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。
在 Direct 模型下：
队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）。
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey。
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息。
总结：
描述下 Direct 交换机与 Fanout 交换机的差异？
Fanout 交换机将消息路由给每一个与之绑定的队列。
Direct 交换机根据 RoutingKey 判断路由给哪个队列。
如果 Direct 交换机绑定的多个队列具有相同的 RoutingKey，则与 Fanout 功能类似。
3.4.3 Topic 交换机：
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！
**BindingKey 一般都是由一个或多个单词组成，多个单词之间以.分割。**例如： item.insert
通配符规则：
#：匹配 0 个或多个词。
*：匹配不多不少恰好 1 个词。
本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！
云掣基于多年在运维领域的丰富时间经验，编写了《云运维服务白皮书》，欢迎大家互相交流学习：
《云运维服务白皮书》下载地址：https://fs80.cn/v2kbbq
想了解更多大数据运维托管服务、数据库运维托管服务、应用系统运维托管服务的的客户，欢迎点击云掣官网沟通咨询：https://yunche.pro/?t=shequ
