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

在实现一个轻量级消息队列时我们需要同时兼顾数据持久化和高效读写。这篇文章会结合代码和设计思路详细拆解DiskDataCenter磁盘数据中心与MemoryDataCenter内存数据中心的实现逻辑重点分析为什么要选择这些数据结构以及它们如何协同工作。一、整体设计思路对于一个 MQ 来说核心数据分为以下两部分元数据交换机Exchange、队列MSGQueue、绑定Binding—— 这些数据需要持久化存储重启后不能丢失所以交给DiskDataCenter管理。消息数据消息本身Message—— 既要持久化到磁盘保证不丢失也要在内存中维护一份副本保证消费速度所以由两个类共同管理。DiskDataCenter负责和硬盘交互MemoryDataCenter负责在内存中缓存所有数据提供高速读写能力。上层业务逻辑只需要和这两个类交互不需要关心底层是数据库还是文件。二、磁盘数据中心DiskDataCenter1. 核心职责管理数据库存储交换机、队列、绑定的元数据。管理数据文件存储消息内容实现消息的持久化与回收。对外提供统一的硬盘操作接口屏蔽底层实现细节。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 ListExchange selectAllExchanges() { ... } // 队列操作 public void insertQueue(MSGQueue queue) throws IOException { ... } public void deleteQueue(String queueName) throws IOException { ... } public ListMSGQueue selectAllQueues() { ... } // 绑定操作 public void insertBinding(Binding binding) { ... } public void deleteBinding(Binding binding) { ... } public ListBinding selectAllBindings() { ... } // 消息操作 public void sendMessage(MSGQueue queue, Message message) throws IOException, MqException { ... } public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException, MqException { ... } public LinkedListMessage 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 { ConcurrentHashMapString, 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) { LinkedListMessage messages queueMessageMap.computeIfAbsent(queue.getName(), k - new LinkedList()); synchronized (messages) { messages.add(message); } addMessage(message); } public Message pollMessage(String queueName) { LinkedListMessage messages queueMessageMap.get(queueName); if (messages null) { return null; } synchronized (messages) { if (messages.size() 0) { return null; } return messages.remove(0); } }链表结构LinkedList实现 FIFOFirst In First Out即先进先出是消息队列的核心语义指先被生产投递到队列的消息会被优先消费保证消息的顺序性这也是 MQ 作为消息中间件最基础的特性之一add()方法是向链表尾部插入消息对应消息生产、remove(0)方法是从链表头部删除消息对应消息消费这两个操作的时间复杂度都是 O (1)无需移动其他元素在高并发的消息生产消费场景下效率极高非常贴合消息队列的核心模型。线程安全对链表操作加锁防止多线程同时读写导致数据混乱比如一个线程正在生产消息往链表尾部加数据另一个线程同时消费消息从头部删数据不加锁可能出现链表结构异常或消息丢失 / 重复的问题。全局索引消息同时存入messageMap其他模块可以通过消息 ID 快速查找比如消费确认时无需遍历队列链表直接通过消息 ID 从messageMap中定位消息大幅提升操作效率。3待确认消息管理public void addMessageWaitAck(String queueName, Message message) { ConcurrentHashMapString, Message messageHashMap queueMessageWaitAckMap.computeIfAbsent(queueName, k - new ConcurrentHashMap()); messageHashMap.put(message.getMessageId(), message); } public void removeMessageWaitAck(String queueName, String messageId) { ConcurrentHashMapString, Message messageHashMap queueMessageWaitAckMap.get(queueName); if (messageHashMap null) { return; } messageHashMap.remove(messageId); }消息被消费者取走后会从queueMessageMap移动到queueMessageWaitAckMap等待消费者确认ACK。如果服务器重启待确认消息不会从硬盘恢复会重新进入queueMessageMap等待再次消费保证消息不丢失。4数据恢复Recoverypublic void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException { // 清空内存 exchangeMap.clear(); queueMap.clear(); bindingsMap.clear(); messageMap.clear(); queueMessageMap.clear(); // 从硬盘恢复元数据 ListExchange exchanges diskDataCenter.selectAllExchanges(); for (Exchange exchange : exchanges) { exchangeMap.put(exchange.getName(), exchange); } // ... 恢复队列、绑定、消息 ... }服务器启动时调用recovery方法将硬盘上的元数据和消息加载到内存恢复运行状态。待确认消息不恢复重启后自动重新入队符合 MQ 的消息可靠性语义。四、总结DiskDataCenter和MemoryDataCenter是整个 MQ 的数据底座它们的设计直接决定了 MQ 的性能、可靠性和可维护性。通过合理选择数据结构、分离内存与磁盘操作我们实现了一个高效、可靠且易于扩展的消息队列存储层。