Day427.RabbitMQ消息队列--1 -谷粒商城

Posted 2021-11-12 阿昌喜欢吃黄桃

RabbitMQ消息队列–1

一、MQ队列介绍

1、MQ

全称 Message Queue，被称为消息队列

---

2、回顾队列

队列，一种类似于 List 的数据结构，专门来存储数据的队列

如果要往队列中存数据的话

• 可以从队头中存，1先进，存到队列中就是这样：3 -> 2 -> 1，从队尾取的时候，1先出，结果就是这样：1 -> 2 -> 3，这是很典型的先进先出队列

• 如果存的方式与上面一样：3 -> 2 -> 1，而取的时候从队头取，3先出，结果就变成了这样：3 -> 2 -> 1，这就是先进后出

• 如果队头、队尾都可以放数据，这样的队列被称为双端队列

Java 中也提供 Queue 的相关操作，但是 Java 里面的 API 都是基于内存级别的，而我们的微服务使用它的 API 来保存数据，那最多只能在它的机器里面使用

分布式系统中，我们需要消息中间件 ，是安装在服务器里的，我们的消息全部保存到这个服务器里面，所有微服务都可以通过这个服务器取消息，而 RabbitMQ 就是我们要使用的消息中间件

---

3、RabbitMQ使用场景

①异步处理

假设我们的一个普通业务，以用户注册为例，用户通过浏览器提交了账号、密码等注册信息，注册信息可能分为以下这三步

这是一种同步模式，用户在每一个步骤都会花费 50ms 的时间，加起来就是 150ms 的时间， 我们发现这是没有必要的

尤其是第2步跟第3步，如果我们使用下面这种模式，可以给它弄一个异步

用户将注册信息写入到数据库之后，我们创建两个异步任务，一个发送注册邮件，一个发送注册短信，当然我们需要最终的完整返回，但是我们只需要等待时间最长的一个返回，就可以获取到两个结果

这样，我们就将时间缩短到了 100ms ，看起来时间更快了，但是实际上，我们连异步都不需要

因为，我们发现注册邮件、注册短信，这两个任务，让它在后台慢慢发就行，成功或是失败，我们无需知道，只要它做了这个事就行，而且我们经常会有收不到短信、收不到邮件的情况，所以，遇到这种情况，我们还可以使用下面这种方式

我们如果将注册信息写入数据库成功了，接下来，我们将注册成功的消息写入消息队列，保存在消息中间件这个服务器中，假设：保存了一个1号用户注册成功的消息，至此，我们就直接给用户返回

因为给消息中间件写消息这个耗时是非常短的，数据库插入数据库可能很慢，需要 50ms ，而写消息，类似于直接操作 redis 一样，可能只花费 5ms，很快，那么用户收到这个响应只需要 55ms，但是用户能不能收到邮件、短信呢？

也可以，我们的消息既然存到了消息队列里面，别的服务就可以从消息队列里面拿到这个消息，假设：这个服务拿到1号用户注册成功的消息了，那么它就在后台该发邮件发邮件，该发短信发短信，我们不关心它什么时候发短信、发邮件，只要它干了这个事就行，但是用户会一起响应成功。

• 总结

这就是异步处理，使用起来比异步任务更快，异步任务我们还必须等待消息返回，而这个异步处理，只需要给消息中间件的服务器发一个消息，让它在后台慢慢处理就行。

---

②应用解耦

我们以下订单为例：比如，我们下了一个订单，我们下完订单之后，需要做出库操作，像我们以前做的，一般是使用下面这种方式

假设订单系统有3个参数，库存系统有5个参数，直接调用就可以，如果这个库存系统不升级，API 也不变，一直是这几个参数还好，

假设我们库存系统经常会升级， 减库存的接口经常发生变化，这样我们以前的这种调用方式，一旦库存系统升级了，则订单系统必须修改它的源代码，重新部署，这样就感觉会非常麻烦，所以我们可以引入消息队列，

订单系统只要下好订单，我们给消息队列里面写上一个消息，说我们哪个用户下了哪个订单购买了哪个商品，把这个消息保存到队列里面，我们不关心库存系统的接口是什么样，不管它要几个参数，我们只需要把我们的消息写进去，接下来库存系统要实时订阅我们队列里面的内容，只要有内容，库存系统就会收到我们订单系统写的消息，然后它自己分析消息，然后对库存进行修改

• 总结

此时，我们发现，订单系统执行完任务之后，我们无需关心库存系统要调用什么接口，我们只需要写消息即可，所以我们就实现了应用解耦

以后，无论什么系统，想要知道我们订单成功之后要做什么，只需要订阅消息队列中订单成功的消息，

而订单系统不需要关心别的系统接口设计成什么样子，因为订单系统根本就不会调用它们

---

③流量控制（流量削峰）

针对一些秒杀业务来说，瞬间流量会非常大，比如：瞬间百万个请求都要进来秒杀一个商品，这个商品要真正去执行业务，就算我们前端服务器可以接受百万请求，我们要执行业务代码，因为我们秒杀完之后，要下订单，整个流程会非常慢， 后台会一直阻塞，可能就会导致资源耗尽，最终导致宕机

此时，我们可以这样做，我们让大并发的请求全部进来，进来以后，先将它们存储到消息队列里面，存到消息队列以后，我们就可以不用管这个请求该怎么做了，直接给它响应：秒杀成功了或者其他

然后，消息队列中，后台真正的业务处理要下订单、减库存等等这些业务处理，我们不着急立即调用，只要存到消息队列里面，这些业务去订阅消息队列里面进来的这些秒杀请求，接下来，挨个处理：下订单…，即使后台每秒只能处理1个，那100W请求，也就花费100W秒，但永远都不会导致机器的资源耗尽，导致宕机所以我们可以达到前端的流量控制。

• 总结

我们把所有的流量存到队列里面，后台根据它的能力，去来进行消费处理，不会导致机器宕机，这就是流量控制，也被称为流量削峰，将峰值削下来，全部存到队列里面

---

二、RabbitMQ概述

1、简介

RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。

①大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦的能力

②消息服务中两个重要概念

消息代理（message broker）和目的地（destination）

• 消息代理：指的是一个代理代替我们发送、接收消息，

  • 简单理解就是，安装了消息中间件的服务器，发消息得用它发，接收消息也得连上它，才能拿到

• 目的地：指消息的目标位置

发消息的整体流程：

当消息发送者要发送消息时，这个消息会先发给消息代理（也就是消息中间件服务器broker），消息代理会发到我们指定的目的地

③消息队列主要有两种形式的目的地

1. 队列（queue）：

   点对点消息通信（point-to-point）

2. 主题（topic）：

   发布（publish）/订阅（subscribe）消息通信

只要是消息中间件，一定会有这两种模式

④点对点式通信（队列式）

消息发送者发送消息，首先发送给消息代理， 消息代理收到消息之后，如果消息发送者说要发给一个队列，消息代理就会将其放入一个队列中，队列都是先进先出，消息先进来就会先取到，

别人如果要获取队列中的消息，怎么办？

别人可以监听队列里的消息内容，一旦队列里面有消息，这个人就可以拿到消息 ，

• 总结

  • 消息有唯一的发送者、接收者， 也就是说谁发送消息这是肯定的，谁最终拿到消息这也是肯定的，但是并不是说只能有一个接收者 ，可以很多人都来接收队列里面的消息，队列可以允许很多人同时监听消息

  • 但是如果是点对点式（队列式），消息放到队列之后，最终只会交给一个人， 谁先抢到，就是谁的

  • 消息一旦被别人抢到，就会从队列中移除，队列里面就没有这个消息了

⑤发布订阅式

发送者（发布者）先将消息发给消息代理， 消息代理要将消息发送到主题，这个主题可以有多个接收者（订阅者）同时监听（订阅），跟队列一样，

• 如果是队列，那么多个人监听，最终只会有一个人收到消息，

• 但如果是一个主题，主题是一种发布订阅模式，只要消息一到达，那么所有订阅消息的人都能收到消息

⑥消息队列规范

JMS（Java Message Service）JAVA消息服务

基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现

⑦消息队列协议

AMQP（Advanced Message Queuing Protocol）

• 高级消息队列协议，也是一个消息代理的规范，兼容JMS
• RabbitMQ是AMQP的实现

⑧规范与协议的对比

	JMS（Java Message Service）	AMQP（Advanced Message Queuing Protocol）
定义	Java api	网络线级协议
跨语言	否	是
跨平台	否	是
Model	提供两种消息模型： （1）、Peer-2-Peer （2）、Pub/sub	提供了五种消息模型： （1）、direct exchange （2）、fanout exchange （3）、topic change （4）、headers exchange （5）、system exchange 本质来讲，后四种和JMS的pub/sub模型没有太大差别， 仅是在路由机制上做了更详细的划分；
支持消息类型	多种消息类型： TextMessage MapMessage BytesMessage StreamMessage ObjectMessage Message （只有消息头和属性）	byte[] 当实际应用时，有复杂的消息，可以将消息序列化后发 送。
综合评价	JMS 定义了JAVA API层面的标准；在java体系中， 多个client均可以通过JMS进行交互，不需要应用修 改代码，但是其对跨平台的支持较差；	AMQP定义了wire-level层的协议标准；天然具有跨平 台、跨语言特性。

⑨Spring支持

• spring-jms提供了对JMS的支持

• spring-rabbit提供了对AMQP的支持

• 需要ConnectionFactory的实现来连接消息代理

• 提供JmsTemplate、RabbitTemplate来发送消息

• @JmsListener(JMS)、@RabbitListener(AMQP)注解在方法上监听消息代理发布的消息

• @EnableJms、@EnableRabbit开启支持

⑩Spring Boot自动配置

• JmsAutoConfiguration
• RabbitAutoConfiguration

---

2、RabbitMQ工作流程

①流程

1. 无论是生产者（Publisher）想要发消息（Message），还是消费者（Consumer）要接消息（Message），它们都必须跟 RabbitMQ 建立一条连接（Connection）
2. 所有的收发数据都需要在连接（Connection）里面开辟信道（Channel）进行收发，想要收发的都是消息（Message），所以我们要构造一个消息（Message），消息有头有体，头相当于是对参数的一些设置、命令，体就是消息的真正内容，而消息里面最重要的一个就是路由键（routing-key）
3. 我们将消息指定好**路由键（routing-key）**要发给谁以后，消息（Message）先来到消息代理（Broker）指定的一个虚拟主机（Virtual Host）里边， 由虚拟主机（Virtual Host）里边指定交换机（Exchange）
4. 这就相当于我们要发消息（Message）的时候，我们还要指定好要发给哪个交换机（Exchange）
5. 由指定的交换机（Exchange）收到消息以后，它根据我们指定的路由键（routing-key），通过交换机跟其它队列（Queue）的绑定关系，将这个消息放到哪个队列（Queue）
6. 然后消费者（Consumer）就会监听这个队列，队列里面的内容就会被消费者（Consumer）实时拿到，当然也是通过信道（Channel）拿到的

②Message

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。

③Publisher

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

④Exchange

交换机 ，用来接收生产者发送的消息，并将这些消息路由给服务器中的队列。
Exchange有4种类型：

direct(默认)、fanout、topic、headers，不同类型的Exchange转发消息的策略有所区别

⑤Queue

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

⑥Binding

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Exchange 和 Queue 的绑定可以是多对多的关系。

⑦Connection

网络连接，比如一个TCP连接。

⑧Channel

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

⑨Consumer

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

⑩Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。

虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。

vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

11.Broker

表示消息队列服务器实体

---

3、RabbitMQ运行机制

• 消息生成者发送一个消息，消息先发给消息代理（Broker），由代理先将消息交给（Exchange）交换机，然后这个交换机下面可能会绑定（Bingdings）了很多队列（Queues），所以一个交换机跟很多种队列都有绑定关系，
• 一个交换机可以绑定很多队列，一个队列也可以被多个交换机绑定，所以它们之间有非常复杂的绑定关系，

接下来就由交换机决定，消息要按照什么绑定关系路由给哪个消息队列，这个关系就是消息路由

• 这个路由是根据一开始发的 路由键（routing-key）指定的，由于这个消息是先发送给交换机，所以交换机不一样，它绑定关系不一样，最终路由到的地方也不一样，消费者收到的消息就不一样

---

三、Docker安装RabbitMQ

# 运行，第一次没安装会自动安装
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management

# 开机自启
docker update rabbitmq --restart=always

• 4369, 25672 (Erlang发现&集群端口)

• 5672, 5671 (AMQP端口)

• 15672 (web管理后台端口)

• 61613, 61614 (STOMP协议端口)

• 1883, 8883 (MQTT协议端口)

---

访问web管理测试是否成功:http://192.168.109.101:15672/

默认账号密码为：guest/guest

---

四、Exchange（交换机）

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：

• direct：直接（精确匹配）

• fanout：广播类型（订阅的都收到）

• topic：主题，就是发布订阅那种模式（部分匹配广播）

• headers ：headers 匹配 AMQP 消息的 header 而不是路由键，headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类

direct、header 是一致的，它们都是JMS中说的点对点通信方式实现

fanout、topic则是发布订阅的一些实现

交换机的类型不同，最终路由到的地方就不一样

1、Web操作

①创建一个交换机

---

②创建一个队列

③交换机绑定队列

一创建之后，点进交换机，交换机的绑定关系里面，就会发现已经与队列 indi 绑定上了

---

2、交换机类型

①Direct Exchange（直接交换机）

精确匹配，指定给谁就发给谁

比如，现在有一个直接交换机，它绑定了3个队列，第一个叫 dog，第二个叫 dog.gurad，第三个叫 dog.puppy，如果说消息发送过来，我们用的路由键叫 dog，那它就会精确的只发送给 dog 队列，实现消息最终只能到达一个队列，这就叫直接类型交换机，也称为单播模式、点对点通信

路由键是跟交换机和队列的绑定关系进行匹配的，我们将这种匹配称之为路由键的完全匹配

发消息是发给交换机，监听消息是监听队列，交换机将消息交给队列了，那么监听这个队列的人就会拿到消息

②Fanout Exchange（广播类型交换机）

广播发送，发给所有绑定这个交换机的所有队列

如果交换机下绑定了3个队列，消息一到达交换机，这3个队列都会收到， 这个消息会广播出去，根本就不关心路由键是什么，把所有消息都通过交换机广播给它绑定的所有队列，被称为广播模式

③Topic Exchange（主题类型交换机）

部分广播，根据条件模糊匹配绑定的队列

虽然它也是广播模式，比如它绑定了几个交换机，但是它可以指定某些交换机来发送消息，其余没指定的，则不会收到消息，所以它是部分广播，主要是根据路由键匹配将消息发个队列，这就是主题-发布订阅模式

• 它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。

• 它同样也会识别两个通配符：

  **符号 “#” 和符号 “”。
  

• #匹配 0 个或多个单词

• * 匹配一个单词

以上面的 usa.# 为例，所有 usa 开头的路由键会进入这个队列，包括只有usa的

而 #.news ，则是所有以 news 结尾的路由键会进入这个队列，包括只有news的

---