RabbitMQ的基本架构与实现原理

1.RabbitMQ Exchange类型

2.RabbitMQ的数据存储

RabbitMQ整体逻辑架构

1.RabbitMQ Exchange类型

常用的交换器类型分为：Direct、Topic、 Fanout、Header 四种。

Fanout

fanout类型的交换器会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中，如图：

Direct

direct类型的交换器会把发送到该交换器的消息路由到BindingKey 与 RoutingKey完全匹配的队列中，如图：

Topic

topic类型的交换器在direct的匹配规则上进行了拓展，也是将消息路由到BindingKey 和 RoutingKey相匹配的队列中，但是匹配规则有些不同：

BindingKey 和 RoutingKey都是由 "." 分隔的字符串，BindingKey中存在两种特殊字符" # "和" * "，用户模糊匹配，其中" * "用于匹配一个单词，” # “用于匹配多个单词。

Headers

headers类型的交换器不依赖于路由键的匹配规则来路由信息，而是根据发送的消息内容中的headers属性进行匹配。在绑定队列和交换器时指定一组键值对，当发送的消息到交换器时，RabbitMQ会获取到该消息的headers, 对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果匹配，消息就会路由到该队列。headers类型的交换器性能很差，不实用。

2.RabbitMQ的数据存储

存储机制

RabbitMQ的消息分为两种类型：持久化消息和非持久化消息。

这两种消息都会被写入磁盘

持久化消息在到达队列时写入磁盘，同事内存中会保存一份备份，当内存吃紧时，消息从内存中剔除，会提高一定的性能。

非持久化消息一般只存在与内存中，当内存压力大时数据刷盘处理，以节省内存空间。

RabbitMQ存储层包含两个部分：

队列索引（rabbit_queue_index）和消息存储(rabbit_msg_store)

队列索引：rabbit_queue_index

索引维护队列的落盘消息的信息，如存储地点，是否已经被消费者接收，是否已经被消费者ack等，每个队列都有对应的索引。

索引使用顺序的段文件来存储，后缀为.idx, 文件名从0开始累加，每个段文件中包含固定的segment_entry_count条记录，默认值是16384。每个index从磁盘中读取信息的时候，至少要在内存中维护一个段文件，所以设置 queue_index_embed_msgs_below 值的时候要格外谨慎，一点点增大也可能导致内存爆炸式增长。

消息存储：rabbit_msg_store

消息以键值对的形式存储到文件中，一个虚拟主机上的所有队列共用一块存储，每个节点只有一个。存储分为持久化存储（msg_store_persistent）和短暂存储（msg_store_transient）。持久化存储的内容在broker重启后不会丢失，短暂存储的内容在broker重启后会丢失。

store 使用文件来存储，后缀.rdq，经过store处理过的所有消息都会以追加的方式写入到该文件中，当文件大小超过指定限制（file_size_limit）后，将会关闭该文件并创建一个新文件以供新消息写入。文件名从0开始进行累加。在进行消息存储时，RabbitMQ会在ETS(Erlang Term Storage)表中记录消息在文件中的位置映射和文件的相关信息。

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

消息（包括消息头，消息体，属性）可以直接存储在index中，也可以存储在store中。最佳方式是较小的消息存储在index中，较大的消息存储在store中。消息大小的界定可以通过queue_index_embed_msgs_below来配置，默认是4096B。当一个完整消息小于设定的大小阈值时，就可以存储在index中，这样性能上可以得到优化（这也是修改消息大小界定可能导致内存爆炸式增长的原因）。

msg 的大小 > queue_index_embed_msgs_below ?

存储到 msg_store_persistent的 <num>.rdq文件中：

存储到<num>.idx索引文件中

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

删除消息时，只是从ETS表删除指定消息的相关信息，同时更新消息对应的存储文件和相关信息。在执行消息删除操作时，不立即对文件进行删除，消息仍存在文件中，但是会被标记为垃圾数据。当一个文件中的所有消息都是垃圾数据时可以将这个文件删除。当检测到前后两个文件中的有效数据可以合并成一个文件，并且所有的垃圾数据的大小和所有文件的数量（至少3个文件存在的情况下）大小的比值超过设置的阈值 garbage_fraction（默认0.5）时，才会触发垃圾回收，将这两个文件合并，执行合并的文件一定是逻辑上相连的两个文件，合并逻辑：

1：先锁定两个文件

2：先整理前面文件的有效数据，再整理后面的文件的有效数据

3：将后面的文件的有效数据全部写入到前面的文件中

4：更新消息再ETS表中的记录

5：删除后面的文件

队列结构

通常队列由 rabbit_amqqueue_process 和 backing_queue 这两部分组成。

rabbit_amqqueue_process 负责协议相关的消息处理，即接收生产者发布的消息，向消费者交付消息，处理消息的确认（包括生产端的confirm 和消费端的 ack）等。 backing_queue 是消息存储的具体形式和引擎，并向 rabbit_amqqueue_process 提供相关的接口以供调用。

如果消息投递的目的队列当前为空，且有消费者订阅该队列，则该消息直接发送给消费者，不经过队列这一步。当消息无法直接投递给消费者时，需要将消息暂存入队列，以便重新投递。

rabbit_variable_queue.erl源码中定义了RabbitMQ队列的4种状态：

1.alpha: 消息索引和消息内容都存内存，最耗内存，很少消耗CPU

2.beta: 消息索引存内存，消息内容存磁盘

3.gama:消息索引内存和磁盘都有，消息内容存磁盘

4. delta:消息索引和内容都存磁盘，基本不消耗内存，消耗更多CPU和I/O操作

消息存如队列后，不是固定不变的，会随着系统的负载在队列中不断流动，消息的状态会不断发生变化。

gama状态只有持久化消息才会有的状态。

在运行时，RabbitMQ会根据消息传递的速度定期计算一个内存中能保存的最大消息数量（target_ram_count），如果alpha状态的消息大于此值，则会引起消息状态的转换，多余的消息可能会转换到beta、gama或者delta状态。区分这四种状态的主要作用是满足不同的内存和CPU需求。

对于普通没有设置优先级和镜像的队列来说，backing_queue的默认实现是 rabbit_variable_queue,其内部通过 5 个子队列Q1， Q2，delta、Q3、Q4体现消息的各个状态。

消费者获取消息也会引起消息的状态转换。

当消费者获取消息时

1.首先会从Q4中获取消息，如果获取成功则返回。

2.如果Q4为空，则尝试从Q3中获取，系统首先会判断Q3是否为空，如果为空则返回队里欸为空，即此时队列中无消息。

3.如果Q3不为空，则取出Q3中的消息，再判断此时Q3和Delta中的长度，如果都为空，则可以认为Q2、Delta、Q3、Q4全部为空，此时将Q1中的消息直接转移至Q4,下次直接从Q4中获取消息。

4.如果Q3为空，Delta不为空，则将Delta的消息转移至Q3中，下次可以直接从Q3中获取消息。在将消息从Delta转移到Q3的过程中，是按照索引分段读取的，首先读取某一段，然后判断读取的消息个数与Delta中的消息个数是否相等，如果相等，则判断Delta中已无消息，则直接将Q2和刚读取到的消息放入到Q3中，如果不相等，仅将此次读取的消息转移到Q3。

这里有两处疑问：

1.为什么Q3为空则可以认定整个队列为空？

试想，如果Q3为空，Delta不为空，那个Q3取出最后一条消息的时候，Delta上的消息就会被转移到Q3这样与Q3为空矛盾；

如果Delta为空且Q2不为空，则在Q3取出最后一条消息时会将Q2的消息并入到Q3中，这样也与Q3为空矛盾；

在Q3被取出最后一条消息之后，如果Q2、Delta、Q3为空，且Q1不为空时，Q1的消息会直接转移到Q4，这与Q4为空矛盾。

以上的论证也解释了另一个问题：为什么Q3和Delta都为空时，则可以认为整个队列为空？