消息存储机制

RabbitMQ 消息存储基于磁盘与内存双层架构，确保高吞吐写入与持久化安全。

存储架构

消息生命周期

生产者发送 → 网络层接收 → 路由到队列 → 写入存储 → 索引更新 → 等待消费

消息存储包含两个核心组件：

• 消息正文存储：消息内容落盘，持久化到文件系统
• 消息索引存储：消息偏移、状态、路由信息索引

消息持久化流程

写入条件

消息满足以下条件时触发持久化：

1. 队列声明为 durable=true
2. 消息投递模式 delivery_mode=2（持久化消息）
3. 队列未设置 auto-delete

落盘策略

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

public class PersistentMessageExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            // 声明持久化队列
            ch.queueDeclare("persistent.queue", 
                true,   // durable = true
                false, 
                false, 
                null);
            
            // 发送持久化消息
            String message = "This is a persistent message";
            ch.basicPublish("", 
                "persistent.queue",
                MessageProperties.PERSISTENT_TEXT_PLAIN,  // delivery_mode = 2
                message.getBytes());
            
            System.out.println("持久化消息已发送并落盘");
        }
    }
}

写入流程

1. 消息到达 Broker，写入内存缓冲区
2. 消息内容序列化写入 .idx 索引文件
3. 消息正文写入 .msg_store 目录的段文件
4. 调用 fsync 刷盘（持久化消息）
5. 更新内存索引，标记消息状态为 ready

非持久化消息（delivery_mode=1）仅存储在内存，Broker 重启后丢失。

消息存储格式

文件结构

/var/lib/rabbitmq/mnesia/rabbit@node/
├── msg_store_persistent/    # 持久化消息存储
│   ├── 0.rdq               # 段文件 0
│   ├── 1.rdq               # 段文件 1
│   └── ...
├── msg_store_transient/     # 非持久化消息存储
└── queues/                  # 队列目录
    └── <queue-id>/
        ├── journal          # 预写日志
        └── index            # 队列索引

段文件机制

消息正文存储在固定大小的段文件中（默认 1GB）：

• 消息按到达顺序追加写入
• 段文件满后创建新段
• 段文件编号单调递增

索引文件结构

消息读取流程

消费读取

1. 消费者发起 basic.consume 请求
2. 队列从索引中查找 ready 状态消息
3. 按偏移量从段文件读取消息正文
4. 消息状态标记为 unacked
5. 消息投递到消费者
6. 收到 basic.ack 后标记为 acked
7. 段文件中的所有消息都被 ack 后，段文件可被清理

import com.rabbitmq.client.*;
import java.io.IOException;

public class ConsumePersistentExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            ch.queueDeclare("persistent.queue", true, false, false, null);
            
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("收到消息: " + message);
                // 手动确认，标记消息为 acked
                ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            };
            
            // autoAck=false，需手动确认
            ch.basicConsume("persistent.queue", false, deliverCallback, consumerTag -> {});
            
            System.out.println("等待消费...");
            Thread.sleep(30000);
        }
    }
}

消息清理机制

段文件回收

段文件清理条件：

1. 段内所有消息已被确认（acked）
2. 段内消息已过期（TTL 场景）
3. 队列长度超过上限（x-max-length）

清理流程

检查段文件状态 → 所有消息已 ack → 标记段文件可回收 → 删除段文件 → 更新索引

如果段文件中存在未确认消息，该段文件不会被清理，即使其他消息已被消费。

性能优化

批量刷盘

RabbitMQ 使用批量刷盘策略减少 I/O 开销：

• 非持久化消息不刷盘
• 持久化消息累积到阈值后批量 fsync
• 默认刷盘间隔 ≤ 200ms

Lazy Queue

Lazy Queue 模式将消息直接写入磁盘，减少内存占用：

import com.rabbitmq.client.*;
import java.util.HashMap;
import java.util.Map;

public class LazyQueueExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            Map<String, Object> args_map = new HashMap<>();
            args_map.put("x-queue-mode", "lazy");  // 启用 Lazy 模式
            
            ch.queueDeclare("lazy.queue", true, false, false, args_map);
            
            // 消息直接写入磁盘，内存占用极小
            for (int i = 0; i < 1000; i++) {
                ch.basicPublish("", "lazy.queue", null, 
                    ("Message " + i).getBytes());
            }
            
            System.out.println("1000 条消息已写入 Lazy Queue");
        }
    }
}

注意事项

持久化消息写入性能低于非持久化消息，因为每次写入都需要调用 fsync 刷盘。

段文件清理需要等待所有消息被确认，未确认消息会阻塞段文件回收。

Lazy Queue 适合大消息存储场景，但会增加磁盘 I/O 压力，高吞吐场景需权衡使用。

要点总结

• 消息存储分为正文存储（段文件）和索引存储（偏移量、状态）
• 持久化消息需满足 durable=true 且 delivery_mode=2
• 消息写入流程：内存缓冲 → 写索引 → 写段文件 → fsync 刷盘
• 段文件按固定大小分割，消息按到达顺序追加写入
• 段文件回收条件：所有消息已 ack 或已过期
• Lazy Queue 将消息直接写入磁盘，适合大消息存储场景

文章存放路径：D:\git2\jwdev\articles\RABBITMQ\专家\底层原理与架构\消息存储机制.md