本文讲解RabbitMQ的存储,主要有以下内容:
首先确认一个点,持久化和非持久化的消息都会落地磁盘,区别在于持久化的消息一定会写入磁盘(并且如果可以在内存中也会有一份),而非持久化的消息只有在内存吃紧的时候落地磁盘。两种类型消息的落盘都是在RabbitMQ的持久层中完成的。
RabbitMQ的持久层只是一个逻辑上的概念,实际包含两个部分:
消息(包括消息体、属性和headers)可以直接存储在rabbit_queue_index中,也可以被保存在rabbit_msg_store中。
最佳的配备方式是较小的消息存储在rabbit_queue_index中而较大的信息则存储在rabbit_msg_store中。消息大小的参数可以通过queue_index_embed_mgs_below来配置,默认大小4096,单位B。
rabbit_queue_index中以顺序的段文件来开始存储,后缀为".idx",每个段文件中包含固定的SEGMENT_ENTRY_COUNT条记录,SEGMENT_ENTRY_COUNT默认值是16384。
经过rabbit_msg_store处理的所有消息都会以追加的方式写入到文件中,当一个文件的大小超过指定的限制(filesizelimit)后,关闭这个文件再创建一个新的文件以供新的消息写入。文件名(文件后缀是".rdq")从0开始进行累加,因此文件名最小的文件也是最老的文件。在进行消息的存储时,RabbitMQ会在ETS(Erlang Term Storage)表中记录消息在文件中的位置映射(Index)和文件的相关信息(FileSummary)。
在读取消息的时候,先根据消息的ID(msg id)找到对应存储的文件,如果文件存在并且未被锁住,则直接打开文件,从指定位置读取消息的内容。如果文件不存在或者被锁住了,则发送请求由rabbit_msg_store进行处理。
消息删除是只是删除ETS表中该消息的相关信息,同时更新消息对应的存储文件的相关信息。执行消息删除操作时,并不立即对文件中的消息进行删除,也就是说消息依然在文件中,仅仅是被标识为垃圾数据而已。一个文件中都是垃圾数据时可以将这个文件删除。当检测到前后两个文件中的有效数据可以合并在一个文件中,并且所有的垃圾数据的大小和所有文件(至少有3个文件存在的情况下)的数据大小的比值超过设置的阀值GARBAGE FRACTION(默认值为0.5)时才会触发垃圾回收将两个文件合并。
通常队列由rabbit_amqpqueue_process和backing_queue两部分组成:
如果消息发送的队列是空的且队列有消费者,该消息不会经过该队列直接发往消费者,如果无法直接被消费,则需要将消息暂存入队列,以便重新投递。消息在存入队列后,主要有以下几种状态:
持久化的消息,消息内容和消息索引必须都保存在磁盘中,才会处于上面状态中的一种,gamma状态只有持久化的消息才有这种状态。
对于没有设置优先级和镜像的队列来说,backing_queue的默认实现是rabbit_variable_queue,其内部通过5个子队列来体现消息的各个状态:
消息的状态一般变更方向是Q1->Q2->Delta->Q3->Q4,大体是从内存到磁盘然后再到内存中。
消费者消费消息也会引起消息状态的转换。
通常在负载正常时,如果消息被消费的速度不小于接收新消息的速度,对于不需要保证可靠不丢失的消息来说,极有可能只会处于alpha状态。对于durable属性设置为true的消息,它一定会进入gamma状态,并且在开启publisher confirm机制时,只有到了gamma状态时才会确认该消息己被接收,若消息消费速度足够快、内存也充足,这些消息也不会继续走到下一个状态。
惰性队列会将接收到的消息直接存入文件系统中,而不管是持久化的或者是非持久化的,这样可以减少了内存的消耗,但是会增加I/0的使用,如果消息是持久化的,那么这样的I/0操作不可避免,惰性队列和持久化的消息可谓是"最佳拍档"。
队列具备两种模式:default和lazy。在队列声明的时候可以通过x-queue-mode参数来设置队列的模式,取值为default和lazy。对应的 Policy设置方式为:
rabbitmqctl set_policy lazy "^myQueue$" '{"queue-mode":"lazy"}' --apply-to queue