消息队列（一）：需求分析

为什么要做这样一个项目？

首先，我们在之前学习的时候，就认识了一下 生产者消费者模式，这样一个模式有两大好处：

• 解耦合
  •   本来有个分布式系统，A服务器 调⽤ B服务器（A给B发请求，B给A返回响应）===》 A 和 B 的耦合是⽐较⼤的！
  • 引⼊消息队列后，A把请求发送到消息队列，B再从消息队列获取到请求；此时就算 A 或者 B 崩溃了，都不会影响到对方
• 削峰填谷
  • ⽐如A是⼊⼝服务器，A 调⽤ B 完成⼀些具体业务，如果是 A 和 B 直接通信，如果突然A 收到 ⼀组⽤户的请求的峰值，此时 B 也会随着受到峰值
  • 引⼊消息队列后，A把请求发送到消息队列，B再从消息队列获取到请求。 （虽然A收到很多请 求，队列也收到了很多请求，但是B仍旧可以按照原来的节奏处理请求。不⾄于说⼀下就收到太 多的并发量。）
  • 举个例⼦：⾼铁⽕⻋站，进站⼝。 乘客好⽐A ，进站⼝好⽐B，是有限的，就需要⼀个队列来 排队，这样不管⼈多少，就不会影响到乘客进站以后的坐⻋。

什么是消息队列？

队列可以说是一个数据结构，可以存储数据，如下图，我们从右侧（队尾）插入元素（入队），从队头获取元素（出队）。

了解了队列之后，我们来看一下什么是消息队列，消息队列就是我们常说的MQ，英文叫Message Queue，是作为一个单独的中间件产品存在的，独立部署。

这个消息队列有啥好处呢？

我们先前已经提过了 解耦合 、 削峰填谷 ；除此之外还有一个好处 ： 异步，关于这个异步，等到正式介绍的相关代码我再来细说，这里就不多介绍了。

市⾯上⼀些知名的消息队列

• RabbitMQ
• Kafka
• RocketMQ
• ActiveMQ

本项目是 基于 RabbitMQ 实现的较为简易的（所谓建议的意思就是还可以继续完善） MQ；RabbitMQ 是一款比较 “ 年长 ” 的消息队列，比较经典，所以选择了这一款消息队列。

需求分析

核心概念1

1. ⽣产者（Producer）
2. 消费者（Consumer）
3. 中间⼈（Broker）
4. 发布（Push） ⽣产者向中间⼈这⾥投递消息的过程
5. 订阅（Subscribe） 哪些消费者要从中间⼈取数据，这个注册的过程，称为 "订阅"
6. 消费 （Consume） 消费者从中间⼈这⾥取数据的动作

一个生产者，一个消费者

多个生产者，多个消费者

核心概念2

Broker server 内部也涉及⼀些关键概念（是为了如何进出队列）

• 虚拟主机（Virtual Host），类似于 MySQL 中的 database，算是⼀个 "逻辑" 上的数据集合
  • ⼀个Broker server 上可以组织多种不同类别数据，可以使⽤ Virtual Host 做出逻辑上的区分
  • 实际开发中，⼀个 Broker server也可能同时⽤来管理多个 业务线上的数据，就可以使⽤ Virtual Host 做出逻辑上的区分。
• 交换机（Exchange）
  •  ⽣产者把消息投递给 Broker Server，实际上是把消息先交给了 （公司某⼀层楼）Broker Server 上的交换机，再由交换机把消息交给对应的队列。 （交换机类似于“前台⼩姐姐”）
• 队列（Queue）
  •  真正⽤来存储处理消息的实体，后续消费者也是从对应的队列中取数据
  • ⼀个⼤的消息队列中，可以有很多具体的⼩队列
• 绑定（Binding）
  •  把交换机和队列之间，建⽴关系。         
  • 可以把 交换机 和 队列 视为，数据库中 多对多的关系。可以想象，在 MQ 中，也是有⼀个这 样的中间表，所谓的 “绑定’其实就是中间表中的⼀项
• 消息（Message）
  • 具体来说，是 服务器A 发给 B 的请求（通过MQ转发）， 服务器B 给 服务器A返回的响应（通过MQ转发）
  • ⼀个消息，可以视为⼀个字符串（⼆进制数据），具体由程序员⾃定义

持久化

关于持久化，上述 虚拟机、交换机、队列、绑定、消息，需要存储起来，光是存内存显然是不够的（为啥不够呢？万一服务器宕机了，我们消息还没有处理，所以还需要将消息恢复一下），所以我们还需要存硬盘，这里以内存为主、硬盘为辅。

关于存硬盘，我们这里要进行一个区分 ：

• 存数据库：交换机、队列、绑定
• 存文件：主要是与消息相关的

在内存中存储的原因：

对于 MQ 来说，能够⾼效的转发处理数据，是⾮常关键的指标！ 因此对于使⽤内存来组织数据，得 到的效率，就⽐放硬盘要⾼很多

在硬盘中存储的原因：

为了防⽌内存中数据随着进程重启/主机重启⽽丢失

核心 API

消息队列服务器（Broker Server），要提供的核⼼API

• 创建队列（queueDeclare）
  • 此处不⽤ Create这样的术语，原因是Create仅仅是创建；⽽ Declare 起到的效果是，不存在则创建，存在就啥也不做
• 销毁队列（queueDelete）
• 创建交换机（exchangeDeclare）
• 销毁交换机（exchageDelete）
• 创建绑定（queueBind）
• 解除绑定（queueUnbind）
• 发布消息（basicPublish）
• 订阅消息（basicConsume）
• 确认消息（basicAck）
  •  这个API起到的效果，是可以让消费者显式的告诉 broker server,这个消息我处理完毕了，提⾼ 整个系统的可靠性~保证消息处理没有遗漏
  • RabbitMQ 提供了 肯定 和 否定的 确认，此处我们项⽬就只有 肯定确认

交换机类型

交换机在转发消息的时候，有一套转发规则的

消息队列提供了几种不同的 交换机类型（ExchangeType） 来描述这里不同的转发规则

RabbitMQ主要实现了如下四种交换机（也是由 AMQP协议定义的）

• Direct 直接交换机
• Fanout 扇出交换机
• Topic 主题交换机
• Header 消息头交换机

而在我们手写的交换机只实现了前三种：

1. Direct 直接交换机

a. ⽣产者发送消息时，会指定⼀个"⽬标队列"的名字（此时的 routingKey就是 队列的名字）

b. 交换机收到后，就看看绑定的队列⾥⾯，有没有匹配的队列

c. 如果有，就转发过去（把消息塞进对应的队列中）

d. 如果没有，消息直接丢弃

2. Fanout 扇出交换机

a. 会把消息放到交换机绑定的每个队列

b. 只要和这个交换机绑定任何队列都会转发消息

3. Topic 主题交换机

a. bindingKey：把队列和交换机绑定的时候，指定⼀个单词（像是⼀个暗号⼀样）

b. routingKey：⽣产者发送消息的时候，也指定⼀个单词

c. 如果当前 bindingKey 和 routingKey 对上了，就可以把消息转发到对应的队列

4. 上述三种交换机类型，就像QQ群发红包

• 专属红包 ======== 直接交换机
• 发个10块钱红包，⼤家都能领 10块钱红包 ======== 扇出交换机
• 我发个⼝令红包，只有输⼊对应⼝令才能领导红包 ======== 主题交换机

至于为啥没有实现 header主题头 交换机 ，那是因为 header 的实现规则复杂，并且实用场景少。

网络通信

其他的服务器（生产者/消费者）通过网络，和咱们的 BrokerServer 进行交互的。

此处设定，使用 TCP + 自定义应用层协议 实现 生产者/消费者 和 BrokerServer 之间进行交互。

应用层协议主要工作：就是让客户端可以通过网络，调用 BrokerServer 提供的编程接口（从而达到看似是生产者远程调用服务器的API）

消息队列服务器（BrokerServer），要提供的核心 API

• 创建队列（queueDeclare）
  • 此处不⽤ Create这样的术语，原因是Create仅仅是创建；⽽ Declare 起到的效果是，不存在则创建，存在就啥也不做
• 销毁队列（queueDelete）
• 创建交换机（exchangeDeclare）
• 销毁交换机（exchageDelete）
• 创建绑定（queueBind）
• 解除绑定（queueUnbind）
• 发布消息（basicPublish）
• 订阅消息（basicConsume）
• 确认消息（basicAck）
  •  这个API起到的效果，是可以让消费者显式的告诉 broker server,这个消息我处理完毕了，提⾼ 整个系统的可靠性~保证消息处理没有遗漏
  • RabbitMQ 提供了 肯定 和 否定的 确认，此处我们项⽬就只有 肯定确认

因此，客户端这边也需要提供上述 API，只有服务器是真正干实事的；客户端只是发送请求、接收响应：

客户端除了提供了上述 9 个方法之外，还需要提供 4个 额外的方法，支撑其他工作

1. 创建 Connection 
2. 关闭 Connection
   1. 此处⽤的 TCP 连接，⼀个 Connection 对象，就代表⼀个 TCP连接
3. 创建 Channel
   1. ⼀个Connection ⾥⾯包含多个 Channel，每个 Channel 上传输的数据都是互不相⼲的
   2. TCP中，建⽴/断开⼀个连接，成本挺⾼的，因此很多时候不希望频繁建⽴断开 TCP 连接
   3. 所以定义⼀个 Channel ，不⽤的时候，销毁 Channel，此处 Channel 是逻辑概念，⽐ TCP 轻量很多
4. 关闭 Channel

消息应答模式

1. 自动应答，消费者把这个消息取走了（调用了回调函数），就算应答了
2. 手动应答，调用 basicAck 这个方法属于手动应答（消费者需要主动调用这个 API 来进行应答）

总结

需要做哪些⼯作？

1.  需要实现 ⽣产者，消费者，brokerserver 三个部分
2.  针对⽣产者消费者来说，主要编写的是 客户端和服务器的通信部分，给客户端提供⼀组 api，让客 户端的业务代码来调⽤，从⽽通过⽹络通信的⽅式远程调⽤ brokerserver 上的⽅法
   1. ⽐如创建交换机，客户端这边只需要提供相关参数即可，然后通过 socket 将 request 传⼊到 ⽹卡中，然后服务器从 ⽹卡中读取 request 解析。然后计算请求得到 response，再通过 socket 写回去⽹卡。
3.  实现 brokerserver 【重点】
   1. 
4. 持久化
   1. ​​​​​​​上述这些关键数据，在硬盘中怎么存储，啥格式存储，存储在哪？​​​​​​​
   2. 后续服务器宕机或是重启了，如何读取这些数据？

如果还是不太理解的话，可以结合下列思维导图一起理解：

思维导图