当前位置：首页 > news >正文

消息队列：内存与磁盘数据中心设计与实现

news 2026/5/11 22:25:32

在实现一个轻量级消息队列时，我们需要同时兼顾数据持久化和高效读写。这篇文章会结合代码和设计思路，详细拆解DiskDataCenter（磁盘数据中心）与MemoryDataCenter（内存数据中心）的实现逻辑，重点分析为什么要选择这些数据结构，以及它们如何协同工作。

一、整体设计思路

对于一个 MQ 来说，核心数据分为以下两部分：

元数据：交换机（Exchange）、队列（MSGQueue）、绑定（Binding）—— 这些数据需要持久化存储，重启后不能丢失，所以交给DiskDataCenter管理。
消息数据：消息本身（Message）—— 既要持久化到磁盘保证不丢失，也要在内存中维护一份副本保证消费速度，所以由两个类共同管理。

DiskDataCenter负责和硬盘交互，MemoryDataCenter负责在内存中缓存所有数据，提供高速读写能力。上层业务逻辑只需要和这两个类交互，不需要关心底层是数据库还是文件。

二、磁盘数据中心（DiskDataCenter）

1. 核心职责

管理数据库：存储交换机、队列、绑定的元数据。
管理数据文件：存储消息内容，实现消息的持久化与回收。
对外提供统一的硬盘操作接口，屏蔽底层实现细节。

2. 核心代码结构

public class DiskDataCenter { // 管理数据库元数据 private DataBaseManager dataBaseManager = new DataBaseManager(); // 管理消息数据文件 private MessageFileManager messageFileManager = new MessageFileManager(); public void init() { dataBaseManager.init(); messageFileManager.init(); } // 交换机操作 public void insertExchange(Exchange exchange) { ... } public void deleteExchange(String exchangeName) { ... } public List<Exchange> selectAllExchanges() { ... } // 队列操作 public void insertQueue(MSGQueue queue) throws IOException { ... } public void deleteQueue(String queueName) throws IOException { ... } public List<MSGQueue> selectAllQueues() { ... } // 绑定操作 public void insertBinding(Binding binding) { ... } public void deleteBinding(Binding binding) { ... } public List<Binding> selectAllBindings() { ... } // 消息操作 public void sendMessage(MSGQueue queue, Message message) throws IOException, MqException { ... } public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException, MqException { ... } public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException { ... } }

3. 设计要点

分层解耦：DiskDataCenter本身不做具体的 IO 操作，而是委托给DataBaseManager和MessageFileManager，方便后续替换底层实现（比如从文件存储换成 Kafka 的日志存储）。
原子操作：创建队列时，同时在数据库和文件系统创建对应资源；删除队列时，同时清理数据库和文件，保证数据一致性。
垃圾回收（GC）：删除消息后，会检查文件碎片比例，达到阈值后触发 GC，整理数据文件释放空间。

三、内存数据中心（MemoryDataCenter）

这部分是 MQ 运行时的核心，所有高频读写操作都在内存中完成。我们重点分析每个属性的数据结构选择原因。

1. 核心属性与数据结构选择

2. 关键代码解析

（1）绑定的插入（线程安全处理）

public void insertBinding(Binding binding) throws MqException { ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>()); synchronized (bindingMap) { if (bindingMap.get(binding.getQueueName()) != null) { throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName()); } bindingMap.put(binding.getQueueName(), binding); } System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName()); }

computeIfAbsent：如果交换机对应的绑定 Map 不存在，则自动创建，避免空指针。
synchronized (bindingMap)：在多线程环境下，防止两个线程同时插入相同的绑定，保证数据唯一性。

（2）消息投递与消费

public void sendMessage(MSGQueue queue, Message message) { LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>()); synchronized (messages) { messages.add(message); } addMessage(message); } public Message pollMessage(String queueName) { LinkedList<Message> messages = queueMessageMap.get(queueName); if (messages == null) { return null; } synchronized (messages) { if (messages.size() == 0) { return null; } return messages.remove(0); } }

链表结构：LinkedList实现 FIFO（First In First Out，即先进先出，是消息队列的核心语义，指先被生产投递到队列的消息，会被优先消费，保证消息的顺序性，这也是 MQ 作为消息中间件最基础的特性之一）
add()方法是向链表尾部插入消息（对应消息生产）、remove(0)方法是从链表头部删除消息（对应消息消费），这两个操作的时间复杂度都是 O (1)，无需移动其他元素，在高并发的消息生产消费场景下效率极高，非常贴合消息队列的核心模型。
线程安全：对链表操作加锁，防止多线程同时读写导致数据混乱（比如一个线程正在生产消息往链表尾部加数据，另一个线程同时消费消息从头部删数据，不加锁可能出现链表结构异常或消息丢失 / 重复的问题）。
全局索引：消息同时存入messageMap，其他模块可以通过消息 ID 快速查找（比如消费确认时，无需遍历队列链表，直接通过消息 ID 从messageMap中定位消息，大幅提升操作效率）。

（3）待确认消息管理

public void addMessageWaitAck(String queueName, Message message) { ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>()); messageHashMap.put(message.getMessageId(), message); } public void removeMessageWaitAck(String queueName, String messageId) { ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName); if (messageHashMap == null) { return; } messageHashMap.remove(messageId); }

消息被消费者取走后，会从queueMessageMap移动到queueMessageWaitAckMap，等待消费者确认（ACK）。
如果服务器重启，待确认消息不会从硬盘恢复，会重新进入queueMessageMap等待再次消费，保证消息不丢失。

（4）数据恢复（Recovery）

public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException { // 清空内存 exchangeMap.clear(); queueMap.clear(); bindingsMap.clear(); messageMap.clear(); queueMessageMap.clear(); // 从硬盘恢复元数据 List<Exchange> exchanges = diskDataCenter.selectAllExchanges(); for (Exchange exchange : exchanges) { exchangeMap.put(exchange.getName(), exchange); } // ... 恢复队列、绑定、消息 ... }