一、介绍

1.消息队列（MQ）

1.1 什么是消息队列

消息队列，即MQ，Message Queue。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-JvG6NAOg-1646754613488)(assets/1527063872737.png)]

消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

如果以后有其它系统也依赖商品服务的数据，同样监听消息即可，商品服务无需任何代码修改。

1.2 AMQP和JMS

MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FbkwQP6o-1646754613489)(assets/1527064480681.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-D5ChFvfC-1646754613489)(assets/1527064487042.png)]

两者间的区别和联系：

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模型；而AMQP的消息模型更加丰富

1.3 常见MQ产品

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tmJNpXfX-1646754613490)(assets/1527064606029.png)]

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
Kafka：分布式消息系统，高吞吐量

2. RabbitMQ

RabbitMQ是基于AMQP的一款消息管理系统

官网： http://www.rabbitmq.com/

官方教程：http://www.rabbitmq.com/getstarted.html

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TE9wmh52-1646754613491)(assets/1532758972119.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kclSyL3i-1646754613491)(assets/1527064762982.png)]

3. 五种消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lA6W0DJH-1646754613492)(assets/1527068544487.png)]

二、RabbitMQ安装

1 docker镜像下载

docker pull rabbitmq:management

注意：如果docker pull rabbitmq 后面不带management，启动rabbitmq后是无法打开管理界面的，所以我们要下载带management插件的rabbitmq.

2.查看docker镜像

docker images

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qUOq6yGb-1646754613492)(assets/image-20210303114254928.png)]

3.创建rabbitmq容器

docker run -di -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management

4.访问管理界面

访问管理界面的地址就是 http://192.168.188.128:15672，可以使用默认的账户登录，用户名和密码都guest[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Psj3vyQl-1646754613492)(assets/1610364513253.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Bz3dRkVv-1646754613493)(assets/1610364550379.png)]

三、SpringBoot整合AMQP

1. 简单队列模式（Queue）

RabbitMQ是一个消息代理：它接受和转发消息。你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。

RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LIzgjvbY-1646754613493)(assets/1532762975546.png)]

P（producer/ publisher）：生产者，一个发送消息的用户应用程序。

C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序

队列（红色区域）：RabbitMQ内部类似于邮箱的一个概念。虽然消息流经RabbitMQ和你的应用程序，但是它们只能****存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

总之：

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

我们将用SpringBoot编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。

我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。

1.1 导入依赖

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KDUMrqOb-1646754613493)(assets/image-20210323195331967.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zkp2v6Ao-1646754613493)(assets/image-20210323195356604.png)]

pom.xml 代码如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.9.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.offcn</groupId><artifactId>spring-amqp-demo01</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-amqp-demo01</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

1.2 修改配置文件

application.yml，代码如下

spring:rabbitmq:host: 192.168.188.128port: 5672username: guestpassword: guest

1.3 创建配置类

package com.offcn.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class SimpleQueueConfig {/*** 定义简单队列名*/private String simpleQueue = "spring.test.queue";@Beanpublic Queue simpleQueue() {return new Queue(simpleQueue);}
}

注意：Queue不要导错包

1.4 发送端

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class SendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/")public void send() {for(int i=0;i<5;i++) {rabbitTemplate.convertAndSend("spring.test.queue", "简单队列消息"+i);}}
}

1.5 接收端

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class ReceiveListener {@RabbitListener(queues= "spring.test.queue")public void recieve(String message) {System.out.println(message);}
}

1.6 输出

访问http://localhost:8080/

控制台输出

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-d0jYuez2-1646754613494)(assets/image-20210323195928332.png)]

2.work消息模型

工作队列或者竞争消费者模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Iog7VsA4-1646754613494)(assets/1532765197277.png)]

创建一个工作队列，在多个工作者之间分配耗时任务。

工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

接下来我们来模拟这个流程：

P：生产者：任务的发布者

C1：消费者，领取任务并且完成任务，假设完成速度较快

C2：消费者2：领取任务并完成任务，假设完成速度慢

面试题：避免消息堆积？

1）采用workqueue，多个消费者监听同一队列。

2）接收到消息以后，而是通过线程池，异步消费。

2.1 创建配置类

@Configuration
public class WorkQueueConfig {/*** 队列名.*/private final String work = "work_queue";@Beanpublic Queue workQueue() {return new Queue(work);}
}

2.2 发送端

@RestController
public class SendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/")public void send() {for (int i = 0; i < 50; i++) {String message = "工作队列消息" + i;rabbitTemplate.convertAndSend("work_queue", message);}}
}

这里我们循环发送50条消息

2.3 接收端

@Component
public class ReceiveListener {@RabbitListener(queues = "work_queue")public void recieve1(String message) {System.out.println("listener1:" + message);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}@RabbitListener(queues = "work_queue")public void recieve2(String message) {System.out.println("listener2:" + message);}
}

我们在消费者1里设置延时时间，模式消费耗时

2.4 输出

访问http://localhost:8080/

控制台输出

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9Fdeu6eB-1646754613494)(assets/image-20210323214604754.png)][外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rtPWwZR5-1646754613495)(assets/image-20210323214620833.png)]

可以发现，两个消费者各自消费了25条消息，而且各不相同，这就实现了任务的分发，并且平均分配。

2.5 能者多劳

刚才的实现有问题吗？

消费者1比消费者2的效率要低，一次任务的耗时较长
然而两人最终消费的消息数量是一样的
消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

我们可以设置每次只处理一个信息，告诉RabbitMQ一次不要向消费端发送多于一条消息。或者换句话说，不要向消费端发送新消息，直到它处理并确认了前一个消息。相反，它会将其分派给不是处于忙碌状态的下一个消费端。

修改配置文件，如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-IEWGSLvR-1646754613495)(assets/image-20210323223608692.png)]

再次运行程序，效果如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2VOiBLtf-1646754613495)(assets/image-20210323223640602.png)] [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eUApXbdo-1646754613495)(assets/image-20210323223656484.png)]

3. 发布者订阅者模式（Publish/Subscribe）

在之前的模式中，我们创建了一个工作队列。工作队列背后的假设是：每个任务只被传递给一个工作人员。在这一部分，我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者。这种模式被称为“发布/订阅”。

解读：

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange类型有以下几种：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

我们这里先学习

Fanout：即广播模式

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

3.1 创建配置类

package com.offcn.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutQueueConfig  {/*** 声明队列名*/private final String fanout1 = "fanout_queue_1";private final String fanout2 = "fanout_queue_2";/*** 声明交换机的名字.*/private final String fanoutExchange = "fanoutExchange";/*** 声明队列.** @return*/@Beanpublic Queue fanoutQueue1() {return new Queue(fanout1);}@Beanpublic Queue fanoutQueue2() {return new Queue(fanout2);}/*** 声明交换机.*/@Beanpublic FanoutExchange exchange() {return new FanoutExchange(fanoutExchange);}/*** 队列绑定交换机，也可在可视化工具中进行绑定.** @return*/@Beanpublic Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange exchange) {return BindingBuilder.bind(fanoutQueue1).to(exchange);}@Beanpublic Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange exchange) {return BindingBuilder.bind(fanoutQueue2).to(exchange);}
}

3.2 发送端

package com.offcn.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class SendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/")public void send() {for (int i = 0; i < 5; i++) {String message = "订阅模式消息" + i;rabbitTemplate.convertAndSend("fanoutExchange", "", message);}}
}

注意：发送消息有三个参数，中间参数为路由key。

3.3 接收端

package com.offcn.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class ReceiveListener {@RabbitListener(queues = "fanout_queue_1")public void recieve1(String message) {System.out.println("listener1:" + message);}@RabbitListener(queues = "fanout_queue_2")public void recieve2(String message) {System.out.println("listener2:" + message);}
}

3.3 输出

访问 http://localhost:8080/

输出

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sgsGwwf7-1646754613496)(assets/image-20210323200335459.png)]

控制台打印，发现两个绑定了不同队列的消费者都接受到了同一条消息
查看RabbitMq 服务器：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HyKZZLKK-1646754613496)(assets/image-20210323200543986.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2isSlsRn-1646754613496)(assets/image-20210323200608143.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wOVNok9T-1646754613497)(assets/image-20210323200641505.png)]

4.路由模式（Direct）

有选择性的接收消息

在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。

在路由模式中，我们将添加一个功能 - 只能订阅一部分消息。例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-E66GoPP4-1646754613497)(assets/1532766437787.png)]

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

4.1 创建配置类

@Configuration
public class DirectQueueConfig {/*** 声明队列名.*/private final String direct1 = "direct_queue_1";private final String direct2 = "direct_queue_2";/*** 声明交换机的名字.*/private final String directExchange = "directExchange";/*** 声明队列.** @return*/@Beanpublic Queue directQueue1() {return new Queue(direct1);}@Beanpublic Queue directQueue2() {return new Queue(direct2);}/*** 声明路由交换机.** @return*/@Beanpublic DirectExchange directExchange() {return new DirectExchange(directExchange);}/*** 队列绑定交换机,指定routingKey,也可在可视化工具中进行绑定.** @return*/@BeanBinding bindingDirectExchange1(Queue directQueue1, DirectExchange exchange) {return BindingBuilder.bind(directQueue1).to(exchange).with("update");}/*** 队列绑定交换机,指定routingKey,也可在可视化工具中进行绑定.** @return*/@BeanBinding bindingDirectExchange2(Queue directQueue2, DirectExchange exchange) {return BindingBuilder.bind(directQueue2).to(exchange).with("add");}}

4.2 发送端

@RestController
public class SendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send01")public void sendMessageA() {for (int i = 0; i < 5; i++) {String message = "路由模式--routingKey=update消息" + i;System.out.println("我是生产信息的：" + message);rabbitTemplate.convertAndSend("directExchange", "update", message);}}@RequestMapping("/send02")public void sendMessageB() {for (int i = 0; i < 5; i++) {String message = "路由模式--routingKey=add消息" + i;System.out.println("我是生产信息的：" + message);rabbitTemplate.convertAndSend("directExchange", "add", message);}}
}

4.3 接收端

@Component
public class ReceiveListener {@RabbitListener(queues = "direct_queue_1")public void recieve1(String message) {System.out.println("listener1:" + message);}@RabbitListener(queues = "direct_queue_2")public void recieve2(String message) {System.out.println("listener2:" + message);}
}

4.4 输出

访问http://localhost:8080/send01

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-IE4AIygq-1646754613498)(assets/image-20210323225013034.png)]

访问http://localhost:8080/send02

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SWLSWPAR-1646754613498)(assets/image-20210323225043108.png)]

5. 主题模式（Topic）

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

举例：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs

audit.*：只能匹配audit.irs

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fO2fPS4E-1646754613499)(assets/1532766712166.png)]

5.1 创建配置类

package com.offcn.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class TopicQueueConfig {/*** 声明队列名.*/private final String topic1 = "topic_queue_1";private final String topic2 = "topic_queue_2";/*** 声明交换机的名字.*/private final String topicExchange = "topicExchange";/*** 声明队列.** @return*/@Beanpublic Queue topicQueue1() {return new Queue(topic1);}@Beanpublic Queue topicQueue2() {return new Queue(topic2);}/*** 声明路由交换机.** @return*/@Beanpublic TopicExchange topicExchange() {return new TopicExchange(topicExchange);}/*** 队列绑定交换机,指定routingKey,也可在可视化工具中进行绑定.** @return*/@BeanBinding bindingTopicExchange1(Queue topicQueue1, TopicExchange exchange) {return BindingBuilder.bind(topicQueue1).to(exchange).with("topic.keyA");}/*** 队列绑定交换机,指定routingKey,也可在可视化工具中进行绑定.* 绑定的routing key 也可以使用通配符：* *：匹配不多不少一个词* #：匹配一个或多个词** @return*/@BeanBinding bindingTopicExchange2(Queue topicQueue2, TopicExchange exchange) {return BindingBuilder.bind(topicQueue2).to(exchange).with("topic.#");}
}

5.2 发送端

package com.offcn.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class SendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/send01")public void send01() {for (int i = 0; i < 5; i++) {String message = "通配符模式--routingKey=topic.keyA消息" + i;System.out.println("我是生产信息的：" + message);rabbitTemplate.convertAndSend("topicExchange", "topic.keyA", message);}}@RequestMapping("/send02")public void send02(){for (int i = 0; i < 5; i++) {String message = "通配符模式--routingKey=topic.#消息" + i;System.out.println("我是生产信息的：" + message);rabbitTemplate.convertAndSend("topicExchange", "topic.keyD.keyE", message);}}
}

5.3 接收端

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class ReceiveListener {@RabbitListener(queues = "topic_queue_1")public void recieve1(String message) {System.out.println("listener1:" + message);}@RabbitListener(queues = "topic_queue_2")public void recieve2(String message) {System.out.println("listener2:" + message);}
}

5.4 输出