SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAmqp的官方地址：https://spring.io/projects/spring-amqp
AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制
SpringAMQP,Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接受消息。包括两部分，其中spring-amqp是基本抽向，spring-rabbit是底层的默认实现
SpringAMQP提供了三个功能：
自动声明队列、交换机及其绑定关系
基于注解的监听器模式，异步接收消息
封装了RabbitTemplate工具，用于发送消息
使用前记得导入依赖
<!--AMQP依赖，包含RabbitMQ--><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId></dependency>
配置MQ地址，在publisher服务的application.yml中添加配置
spring: rabbitmq: host: 192.168.50.129 # 主机号 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: root # 用户名 password: root # 密码
编写测试类，并利用RabbitTemplate实现消息发送
@RunWith(SpringRunner.class)@SpringBootTestpublic class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSentMessageSimpleQueue(){
String queueName = "simple.queue";
String message = "hello,spring amqp";
rabbitTemplate.convertAndSend(queueName,message);
}}
首先配置MQ地址，在consumer服务的application.yml中添加配置
spring: rabbitmq: host: 192.168.50.129 # 主机号 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: root # 用户名 password: root # 密码
然后在Consumer服务中新建一个类SpringRabbitListener
import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg){
System.out.println("spring 消费者接收到消息：【" + msg + "】");
}}
Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。
模拟WorkQuue，实现一个队列绑定多个消费者
基本思路：
在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue
在consumer服务中定义两个消息监听者，都监听simple.queue队列
消费者1每秒处理50条消息，消费者2每秒处理10条消息
循环发消息，模拟大量消息堆积现象
@Testpublic void testSendWorkQueue() throws InterruptedException {
String queueName = "simple.queue";
String message = "hello,message_";
for (int i = 0; i < 50; i++) {
rabbitTemplate.convertAndSend(queueName,message + i);
Thread.sleep(20);
}}
@RabbitListener(queues = "simple.queue")public void listenWorkQueue1(String msg) throws InterruptedException{
System.out.println("消费者1接收到消息：【" + msg + "】");
Thread.sleep(20);}@RabbitListener(queues = "simple.queue")public void listenWorkQueue2(String msg) throws InterruptedException{
System.out.println(".....消费者2接收到消息：【" + msg + "】");
Thread.sleep(200);}
经过测试发现，两个消费者接受的消息数量是相同的，这是因为RabbitMQ的消息预取机制，也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力
在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：
spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
Work模型的使用：
多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
通过设置prefetch来控制消费者预取的消息数量
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
Consumer：消费者，与以前一样，订阅队列，没有变化
Queue：消息队列也与以前一样，接收消息、缓存消息。
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
Fanout，在MQ中相当于广播
在广播模式下，消息发送流程是这样的：
可以有多个队列
每个队列都要绑定到Exchange（交换机）
生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
交换机把消息发送给绑定过的所有队列
订阅队列的消费者都能拿到消息
Spring提供了一个接口Exchange，来表示所有不同类型的交换机：
在consumer服务中创建一个类，声明队列和交换机：
@Configurationpublic class FanoutConfig {
/**
* 声明交换机
* @return
*/
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("luosaihuzi.fanout");
}
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
/**
* 绑定交换机和队列
* @param fanoutQueue1
* @param fanoutExchange
* @return
*/
@Bean
public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
@Bean
public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}}
在publisher服务中添加测试方法：
@Testpublic void testFanoutExchange(){
String exchangeName = "luosaihuzi.fanout";
String message = "hello,everyone!";
rabbitTemplate.convertAndSend(exchangeName,"",message);}
在consumer服务中添加两个接收方法
@RabbitListener(queues = "fanout.queue1")public void listenFanoutQueue1(String msg) {
System.out.println("消费者1接收到Fanout消息：【" + msg + "】");}@RabbitListener(queues = "fanout.queue2")public void listenFanoutQueue2(String msg) {
System.out.println("消费者2接收到Fanout消息：【" + msg + "】");}
交换机的作用是什么？
接收publisher发送的消息
将消息按照规则路由到与之绑定的队列
不能缓存消息，路由失败，消息丢失
FanoutExchange的会将消息路由到每个绑定的队列
声明队列、交换机、绑定关系的Bean是什么？
Queue
FanoutExchange
Binding
在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
这种模式下的交换机可以根据消息发送时所携带的key来找到符合要求的队列
在Direct模型下：
队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
@Testpublic void testDirectExchange(){
//交换机名称
String exchangeName = "luosaihuzi.direct";
String message = "hello,blue";
rabbitTemplate.convertAndSend(exchangeName,"yellow",message);}
利用注解绑定exchange交换机和queue队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "luosaihuzi.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 = "luosaihuzi.direct",type = ExchangeTypes.DIRECT),
key = {"red","yellow"}))public void listenDirectQueue2(String msg){
System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");}
描述下Direct交换机与Fanout交换机的差异？
Fanout交换机将消息路由给每一个与之绑定的队列
Direct交换机根据RoutingKey判断路由给哪个队列
如果多个队列具有相同的RoutingKey，则与Fanout功能类似
基于@RabbitListener注解声明队列和交换机有哪些常见注解？
@Queue
@Exchange
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
通配符规则：
#：匹配一个或多个词
*：匹配不多不少恰好1个词
举例：
item.#：能够匹配item.spu.insert 或者 item.spu
item.*：只能匹配item.spu
@Testpublic void testTopicExchange(){
//交换机名称
String exchangeName = "luosaihuzi.topic";
String message = "今天中国天气很好!";
rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "luosaihuzi.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 = "luosaihuzi.topic",type = ExchangeTypes.TOPIC),
key = "#.news"))public void listenTopicQueue2(String msg){
System.out.println("消费者接收到topic.queue2的消息：【"+msg+"】");}
描述下Direct交换机与Topic交换机的差异？
Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割
Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词
*：代表1个词
Spring会把发送的消息序列化作为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：
数据体积过大
有安全漏洞
可读性差
SpringAMQP中消息的序列化和反序列化是怎么实现的？
利用MessageConverter实现的，默认实JDK的序列化
注意发送方和接受方必须使用相同的MessageConverter
显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。
在publisher和consumer两个服务中都引入依赖：
<dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.9.10</version></dependency>
配置消息转换器。
在启动类中添加一个Bean即可：
@Beanpublic MessageConverter jsonMessageConverter(){
return new Jackson2JsonMessageConverter();}
