RabbitMQ详解（三）：存储机制、流控

六、RabbitMQ高阶

1、存储机制

不管是持久化的消息还是非持久化的消息都可以被写入到磁盘。持久化的消息在到达队列时就被写入磁盘，并且如果可以，持久化的消息也会在内存中保存一份备份，这样可以提高一定的性能，当内存吃紧的时候会从内存中清除。非持久化的消息一般只保存在内存中，在内存吃紧的时候会被换入到磁盘中，以节省内存空间。这两种类型的落盘处理都在RabbitMQ的持久层中完成

持久层是一个逻辑上的概念，实际包含两个部分：队列索引（rabbit_queue_index）和消息存储（rabbit_msg_store）。rabbit_queue_index负责维护队列中落盘消息的信息，包括消息的存储地点、是否已被交付给消费者、是否已被消费者ack等。每个队列都有与之对应的一个rabbit_queue_index。rabbit_msg_store以键值对的形式存储消息，它被所有队列共享，在每个节点有且只有一个。rabbit_msg_store具体还可以分为msg_store_persistent和msg_store_transient，msg_store_persistent负责持久化消息的持久化，重启后消息不会丢失；msg_store_transient负责非持久化消息的持久化，重启后消息会丢失

消息（包括消息体、属性和headers）可以直接存储在rabbit_queue_index中，也可以被保存在rabbit_msg_store中。最佳的配备是较小的消息存储在rabbit_queue_index中而较大的消息存储在rabbit_msg_store中。这个消息大小的界定可以通过queue_index_embed_msgs_below来配置，默认大小为4096B。这里的大小是指消息体、属性及headers整体的大小

rabbit_queue_index中以顺序（文件名从0开始累加）的段文件来进行存储，后缀为.idx，每个段文件中包含固定的SEGMENT_ENTRY_COUNT条记录，SEGMENT_ENTRY_COUNT默认为16384。每个rabbit_queue_index从磁盘中读取消息的时候至少要在内存中维护一个段文件

经过rabbit_msg_store处理的所有消息都会以追加的方式写入到文件中，当一个文件的大小超过指定的限制（file_size_limit）后，关闭这个文件再创建一个新的文件以供新的消息写入。文件名（文件后缀是.rdq）从0开始进行累加，因此文件名最小的文件也是最老的文件。在进行消息的存储时，RabbitMQ会在ETS表中记录消息在文件中的位置映射和文件的相关信息

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

消息的删除只是从ETS表删除指定消息的相关信息，同时更新消息对应的存储文件的相关信息。执行消息删除操作时，并不立即对在文件中的消息进行删除，也就是说消息依然在文件中，仅仅是标记为垃圾数据而已。当一个文件中都是垃圾数据时可将这个文件删除。当检测到前后两个文件中的有效数据可以合并在一个文件中，并且所有的垃圾数据的大小和所有文件的数据大小的比值超过设置的阈值GARBAGE_FRACTION（默认值为0.5）时才会触发垃圾回收将两个文件合并

执行合并的两个文件一定是逻辑上相邻的两个文件。执行合并时首先锁定这两个文件，并对前面文件中的有效数据进行整理，再将后面文件的有效数据写入到前面的文件，同时更新ETS表中的记录，最后删除后面的文件

1）、队列的结构

通常队列由rabbit_amqqueue_process和backing_queue这两部分组成，rabbit_amqqueue_process负责协议相关的消息处理，即接收生产者发布的消息、向消费者交付消息、处理消息的确认等。backing_queue是消息存储的具体形式和引擎，并向rabbit_amqqueue_process提供相关的接口以供调用

如果消息投递的目的队列是空的，并且有消费者订阅了这个队列，那么该消息会直接发送给消费者，不会经过队列这一步。而当消息无法直接投递给消费者时，需要暂时将消息存入队列，以便重新投递。消息存入队列后，不是固定不变的，它会随着系统的负载在队列中不断地流动，消息的状态会不断发生变化，RabbitMQ中的队列消息可能会处于以下4中状态

• alpha：消息内容（包括消息体、属性和headers）和消息索引都存储在内存中
• beta：消息内容保存在磁盘中，消息索引保存在内存中
• gamma：消息内容保存在磁盘中，消息索引在磁盘中和内存中都有
• delta：消息内容和索引都在磁盘中

对于持久化的消息，消息内容和消息索引都必须先保存在磁盘上，才会处于上述状态中的一种。而gamma状态的消息是只有持久化的消息才会有的状态

RabbitMQ在运行时会根据统计的消息传送速度定期计算一个当前内存中能够保存的最大消息数量，如果alpha状态的消息数量大于此值时，就会引起消息的状态转换，多余的消息可能会转换到beta状态、gamma状态或者delta状态。区分这4种状态的主要作用是满足不同的内存和CPU需求。alpha状态最耗内存，但很少消耗CPU。delta状态基本不消耗内存，但是需要消耗更多的CPU和磁盘I/O操作。delta状态需要执行两次I/O操作才能读到消息，一次是读消息索引，一次是读消息内容；beta和gamma状态都只需要一次I/O操作就可以读取到消息

对于普通的没有设置优先级和镜像的队列来说，backing_queue的morning实现是rabbit_variable_queue，其内部通过5个子队列Q1、Q2、Delta、Q3和Q4来体现消息的各种状态。整个队列包括rabbit_amqqueue_process和backing_queue的各个子队列。其中Q1、Q4只包含alpha状态的消息，Q2和Q3包含beta和gamma状态的消息，Delta只包含delta状态的消息。一般情况下，消息按照Q1->Q2->Delta->Q3->Q4这样的顺序步骤进行流动，但并不是每一条消息都一定会经历所有的状态，这个取决于当前系统的负载状况。从Q1至Q4基本经历内存到磁盘，再由磁盘到内存这样的一个过程，如此可以在队列负载很高的情况下，能够通过将一部分消息由磁盘保存来节省内存空间，而在负载很低的时候，这部分消息又渐渐回到内幕才能被消费者获取，使得整个队列具有很好的弹性。

消费者获取消息也会引起消息的状态转换。当消费者获取消息时，首先会从Q4中获取消息，如果获取成功则返回。如果Q4为空，则尝试从Q3中获取消息，如果Q3为空则返回队列为空，即此时队列中无消息，如果Q3不为空，则取出Q3中的消息，进而在判断此时Q3和Delta中的长度，如果都为空，则可以认为Q2、Delta、Q3、Q4全部为空，此时将Q1中的消息直接转移至Q4，下次直接从Q4中获取消息。如果Q3为空，Delta不为空，则将Delta的消息转移至Q3，下次可以直接从Q3中获取消息。在将消息从Delta转移到Q3的过程中，是按照索引分段读取的，首先读取某一段，然后判断读取的消息的个数与Delta中消息的个数是否相等，如果相等，则可以判定此时Delta中已无消息，则直接将Q3和刚读取到的消息一并放入到Q3中；如果不相等，仅将此次读取到的消息转移到Q3

通常在负载正常时，如果消息被消费的速度不小于接收新消息的速度，对于不需要保证可靠不丢失的消息来说，极有可能只会处于alpha状态。对于durable属性设置为true的消息，它一定会进入gamma状态，并且在开启publisher confirm机制时，只有到了gamma状态时才会确认该消息已被接收，若消息消费速度足够快、内存也充足，这些消息也不会继续走到下一个状态

2）、惰性队列

惰性队列会尽可能地将消息存入到磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标时能够支持更长的队列，即支持更多的消息存储，当消费者由于各种各样的原因致使长时间内不能消费消息而造成堆积时，惰性队列就很有必要了

默认情况下，当生产者将消息发送到RabbitMQ的时候，队列中的消息会尽可能地存储在内存之中，这样可以更快地将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当RabbitMQ需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息

惰性队列会将接收到的消息直接存入文件系统中，而不管是持久化或者非持久化的，这样可以减少了内存的消耗，但是会增加I/O的使用，如果消息是持久化的，那么这样的I/O操作是不可避免。如果惰性队列中存储的是非持久化的消息，内存的使用率会一直很稳定，但是重启之后消息一样会丢失

2、流控

流控机制是用来避免消息的发送速率过快而导致服务器难以支撑的情形。内存和磁盘警告相当于全局的流控，一旦触发会阻塞集群中所有的Connection，而本节的流控是针对单个Connection的

Erlang进程之间并不共享内存，而是通过消息传递来通信，每个进程都有自己的进程邮箱。默认情况下，Erlang并没有对进程邮箱的大小进行限制，所以当有大量消息持续发往某个进程时，会导致该进程邮箱过大，最终内存溢出并崩溃。在RabbitMQ中，如果生产者持续高速发送，而消费者速度较低时，如果没有流控，很快就会使内部进程邮箱的大小达到内存阈值

RabbitMQ使用了一种基于信用证算法的流控机制来限制发送消息的速率以解决前面所提出的问题。它通过监控各个进程的进程邮箱，当某个进程负载过高而来不及处理消息时，这个进程的进程邮箱就会开始堆积消息。当堆积到一定量时，就会阻塞而不接收上游的新消息。从而慢慢地，上游进程的进程邮箱也会开始堆积消息。当堆积到一定量时也会阻塞而停止接收上游的消息，最后就会是负责网络数据包接收的进程阻塞而暂停接收新的数据

进程A接收消息并转发至进程B，进程B接收消息并转发至进程C。每个进程中都有一堆关于收发消息的credit值。以进程B为例，{{credit_from,C},value}表示能发送多少条消息给C，每发送一条消息该值减1，当为0时，进程B不再往进程C发送消息也不在接收进程A的消息。{{credit_to,A},value}表示再接收多少条消息就向进程A发送增加credit值的通知，进程A接收到该通知后就增加{{credit_from,B},value}所对应的值，这样进程A就能持续发送消息。当上游发送速率高于下游接收速率时，credit值就会被逐渐耗光，这时进程就会被阻塞，阻塞的情况一直会传递到最上游。当上游进程收到来自下游进程的增加credit值的通知时，若此时上游进程处于阻塞状态则解除阻塞，开始接收更上游进程的消息，一个一个传到最终能够解除最上游的阻塞状态

一个连接触发流控时会处于flow的状态，也就意味着这个Connection的状态每秒在blocked和unblocked之间来回切换数次，这样可以将消息发送的速率控制在服务器能够支撑的范围之内。可以通过rabbitmqctl list_connections命令或者Web管理界面来查看Connection的状态

处于flow状态的Connection和处于running状态的Connection并没有什么不同，这个状态只是告诉系统管理员相应的发送速率受限了。而对于客户端而言，它看到的知识服务器的带宽要比正常情况下要小一些

流控机制不只是作用于Connection，同样作用于信道和队列。从Connection到Channel，再到队列，最后是消息持久化存储形成一个完整的流控链，对于处于整个流控链中的任意进程，只要该进程阻塞，上游的进程必定全部被阻塞。也就是说，如果某个进程达到性能瓶颈，必然会导致上游所有的进程被阻塞。处理消息的几个关键进程及其对应的顺序关系如下图：

• rabbit_reader：Connection的处理进程，负责接收、解析AMQP协议数据包等
• rabbit_channel：Channel的处理进程，负责处理AMQP协议的各种方法、进行路由解析等
• rabbit_amqqueue_process：队列的处理进程，负责实现队列的所有逻辑
• rabbit_msg_store：负责实现消息的持久化