消息队列 - RabbitMQ

1. 名词解释

Producer：生产者

Broker：接收和分发消息的应用

Connection：生产者和消费者与 Broker 之间的 TCP 连接

Channel：信道；在 Connection 内部建立的逻辑连接，每个 Channel 之间是相互隔离的。相当于数据库的连接池，不必每次访问都建立连接，减少系统的开销

Routing Key：路由规则；用于交换机识别将消息推送至哪个队列中

Exchange：交换机，负责消息的分发；消息到达 Broker 后，首先流转至交换机。然后根据制定的消息分发规则，去查询表中匹配消息中的 Routing Key，然后将消息推送至队列中

Queue：队列；在交换机中找到对应的 Routing Key 后，消息就流转到队列中，等待被消费

Binding：Exchange 和 Queue 之间的虚拟连接，连接中可以包含相应的 Routing Key，连接中用于匹配 Routing Key 的信息会被保存到 Exchange 的查询表中，用于消息的分发

Virtual host：虚拟主机；当不同的用户使用同一个 RabbitMQ 服务时，将不同的用户划分在不同的虚拟主机中，每一台主机之间互不干扰

Consumer：消费者

2. 基础架构

根据上面的描述，我们可以得到一个大概的 RabbitMQ 架构图

3. 常用交换机类型

3.1 Direct Exchange

直连交换机

一个队列会和一个交换机绑定，然后再绑定一个 Routing Key
消息在被发送时，要 Binding 一个 Routing Key
当消息送达到交换机之后，会在查询表中找到对应 Routing Key 的队列，然后将交换机推送至队列中

3.2 Fanout Exchange

扇形交换机

像广播一样，将消息发送到与之绑定的全部队列中

3.3 Topic Exchange

主题交换机

按本人的理解，主题交换机是直连交换机的升级版
主题交换机是让 Routing Key 按照一定的匹配规则去匹配交换机中的多个队列
也就是说，一个 Message 可以发送至多个队列中

但是，Routing Key 不能随意编写，要满足一定的要求
首先他必须是一个单词列表，以点号分开

规则：

• 【*】 用于匹配一个单词
• 【#】用于匹配多个单词

当交换机中的队列绑定键是【#】，那么交换机中这个队列将接收所有数据，类似于上述的扇形交换机
当交换机中的队列绑定键当中没有【#】和【*】出现，那么交换机中这个队列将只会接收和他唯一对应 Routing Key 的数据，类似于上述的直连交换机

3.4 Header Exchange

头交换机

不依赖于 Routing Key 的匹配规则
匹配机制是匹配消息头中的属性信息

在 Message、Queue 与 Exchange 进行 Binding 之前，会声明一个键值对对象，通过这个键值实现 Message、Queue 与 Exchange 的绑定

当 Message 发送到 MQ 时，会将 Message 中的 Headers 信息取出，与 Exchange 绑定时使用的键值对进行匹配

匹配规则有两种类型：

• x-match = all ：表示所有的键值对都匹配才能接受到消息
• x-match = any ：表示只要存在一个键值对匹配就能接受到消息

4. 发送数据

1. 在 RabbitMQ 中，Producer 不会知道 Message 是否已经被推送到 Queue 中，他们的工作只是将消息推送至 Exchange 中

2. Exchange 的工作也很简单，一边接收 Producer 的消息，一边查找查询表的路由规则，将消息推送至 Queue 中

5. 保存数据

5.1 消息分类

消息有两种类型：持久化消息和非持久化消息；两种消息都会被写入磁盘中
非持久化消息：一般只存于内存中，当内存不足且又还没被消费时，才会被写入磁盘，以节省内存空间
持久化消息：在被推送到队列时写入磁盘，同时会在内存中进行备份；当内存不足时，消息就会被清理掉

5.2 存储层

RabbitMQ 存储层包含两个部分：队列索引和消息存储

5.2.1 队列索引 rabbit_queue_index

用于记录队列中所有 Message 的信息，信息包括：存储的地点，是否已经被推送至消费者等

每个队列都有相对应的 index

使用顺序的段文件来存储
段文件的意思就是，系统把一个逻辑空间分为若干个段，每个段都有他自己的一组逻辑意义
段的长度由相应的逻辑信息的长度来决定，也就是说，各个段的长度是不同的；在这一点上，分段是比较灵活的。因为分页存储，页的大小是固定的，只要空间不够，就会将数据切割开来，不会考虑逻辑信息是否完整。在编程和使用的方面来说，不够灵活

5.2.2 消息存储 rabbit_msg_store

键值的形式存储消息
所有队列共享同一个 rabbit_msg_store
持久化消息的持久化和非持久化消息的持久化由 rabbit_msg_store 来进行

使用文件来进行存储 ，经过 rabbit_msg_store 处理的所有消息，都是以文件追加的方式写入文件中；当超过文件的限制大小之后，就会关闭该文件，然后新建一个文件继续进行写入

要注意的是，写入文件的操作也不是立刻执行的
在写入文件之前，存在一个缓冲区。数据在写入文件时，首先会写到这个缓冲区里，如果缓冲区已经满了就会将缓冲区里的数据写入到文件中；或者是达到固定的刷盘时间时，此时无论缓冲区满不满都会写入到文件中

在进行消息的存储时，RabbitMQ 会在 ETS（Erlang Term Storage）表中记录消息在文件中的位置映射和文件的相关信息

ETS：Erlang 所独有的，是一个基于内存的 KV( Key Value) Table，支持大数据量存储以及高效查询

5.3 存储机制

Message 可以直接存储在 rabbit_queue_index 中，也可以存储在 rabbit_msg_store 中

一般的处理方式是：将较小的消息存在 rabbit_queue_index 中，而较大的消息存在 rabbit_msg_store 中

读取 Message 时，先根据 Message 的编号找到对应存储的文件
如果文件存在并且未被锁住，则直接打开文件，从指定位置读取消息内容
如果文件不存在或者被锁住了，则发送请求由 rabbit_msg_store 进行处理

6. 消费数据

有两种消费方式：推模式和拉模式

6.1 推模式

此时 Channel 被设置成投递模式
只要 Consumer 订阅了 Queue，当 Message 到达 RabbitMQ 时，RabbitMQ 会自动且不断的投递 Message 给匹配的 Consumer，而不需要 Consumer 手动来拉取

在该模式下，Message 会提前推送给 Consumer，那么此时就会出现 Consumer 来不及消费的情况；所以在 Consumer 里，设有一个缓冲区，来不及消费的 Message 会暂时存储在缓存区里

使用该模式，优点是 Message 可以及时的被消费，因为此时内存里存在较多的 Message，而不用去文件里，以 IO 的方式读取；缺点就是缓冲区可能会溢出

6.2 拉模式

Consumer 在需要时才会去 Queue 中拉取

在该模式下，实时性较差；且因为是在需要时才会重新建立连接进行拉取，会增加网络上的开销

7. 删除数据

只是从 ETS 表删除指定消息的相关信息，同时更新消息对应的存储文件和相关信息

并不会立即对文件中的消息进行删除，仅仅是标记为垃圾数据而已
有点像项目中的软删除

进行删除操作的前提是：一个文件中的数据全部都是垃圾数据；当满足这个前提之后，会将这个文件一次性全部删除

在 RabbitMQ 中设有检测机制：当检测到逻辑上相邻的两个文件中，所有的垃圾数据大小和所有文件的数据大小的比值超过设置的阈值时，才会触发垃圾回收
合并逻辑：

1. 锁住这两个文件，禁止读取
2. 按逻辑的顺序，先整理逻辑排在前面的文件，再整理后面的文件
3. 将后面文件的数据追加到前面的文件中
4. 更新 ETS
5. 删除后面的文件