【消息队列开发】实现MemoryDataCenter类——管理内存数据

た入场券 2024-04-17 10:53 35阅读 0赞

#### 文章目录 ####

*  ?前言
 *  ?数据格式的准备
 *  ?内存操作
 *   *  ?对于交换机
     *  ?对于队列
     *  ?对于绑定
     *  ?对于单个消息
     *  ?对于队列与消息链表
     *  ?对于未确认消息
     *  ?从硬盘上读取数据
 *  ⭕总结

## ?前言 ##

本次开发目标

*  创建一个类，用适当的数据结构来存储硬盘上的数据。

前面说过，我们的数据要在硬盘与内存中各存在一份，银盘中的数据通过前面的操作已经存储完毕，接下来我们在内存中进行存储。

![在这里插入图片描述][b1d51e9b996f450bbd2aa0a78ac0966d.png]

## ?数据格式的准备 ##

由于我们提供的这些方法后面可能在多线程的环境下使用，所以我们这里使用的是线程安全的 `ConcurrentHashMap`

我们所需要的数据格式有以下几种

key 是 exchangeName, value 是 Exchange 对象

*  `ConcurrentHashMap<String, Exchange>`

key 是 queueName, value 是 MSGQueue 对象

*  `ConcurrentHashMap<String, MSGQueue> queueMap`

第一个 key 是 exchangeName, 第二个 key 是 queueName

*  `ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>>`

key 是 messageId, value 是 Message 对象

*  `private ConcurrentHashMap<String, Message>`

key 是 queueName, value 是一个 Message 的链表

*  `ConcurrentHashMap<String, LinkedList<Message>>`

key 是 queueName, 第二个 key 是 messageId

*  `ConcurrentHashMap<String,ConcurrentHashMap<String, Message>>`

代码实现如下：

// key 是 exchangeName, value 是 Exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName, value 是 MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName, 第二个 key 是 queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId, value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName, value 是一个 Message 的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个 key 是 queueName, 第二个 key 是 messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

## ?内存操作 ##

对这些内存格式提供一些简单的操作

### ?对于交换机 ###

调用ConcurrentHashMap 相应方法即可

代码实现如下：

public void insertExchange(Exchange exchange) {
        
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());
    }
    
    public Exchange getExchange(String exchangeName) {
        
        return exchangeMap.get(exchangeName);
    }
    
    public void deleteExchange(String exchangeName) {
        
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }

### ?对于队列 ###

直接调用方法进行使用即可

代码实现如下：

public void deleteExchange(String exchangeName) {
        
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }
    
    public void insertQueue(MSGQueue queue) {
        
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName=" + queue.getName());
    }
    
    public MSGQueue getQueue(String queueName) {
        
        return queueMap.get(queueName);
    }

### ?对于绑定 ###

我们在添加绑定时，需要注意的是，由于我们使用的了一个嵌套的 ConcurrentHashMap ，所以在添加时我们需要进行判定，判定当前交换机对应的 value 是否位null，若为null，我们则需要进行创建。

而对于里面嵌套的 哈希表来说，我们也需要进行判断，根据 队列名字进行查询。如果存在，则说明该交换机与队列已经绑定了，就无需在绑定了，否则添加绑定

此外，该操作，为了保证线程安全，我们对他进行加锁操作

代码实现如下：

public void insertBinding(Binding binding) throws MqException {
        
    //        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
    //        if (bindingMap == null) {
        
    //            bindingMap = new ConcurrentHashMap<>();
    //            bindingsMap.put(binding.getExchangeName(), bindingMap);
    //        }
        // 先使用 exchangeName 查一下, 对应的哈希表是否存在. 不存在就创建一个.
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
    
        synchronized (bindingMap) {
        
            // 再根据 queueName 查一下. 如果已经存在, 就抛出异常. 不存在才能插入.
            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());
    }

获取绑定的时候，我们这里提供两种方法：

*  一种是根据所提供的交换机与队列名确定唯一的绑定
 *  一种是根据所提供的交换机名返回该交换机的所有绑定

代码实现如下：

public Binding getBinding(String exchangeName, String queueName) {
        
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
        
            return null;
        }
        return bindingMap.get(queueName);
    }
    
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        
        return bindingsMap.get(exchangeName);
    }

删除绑定时，我们首先要根据传入的 Binding对象进行查询绑定，如果发现为null，则就不用删除，抛出异常即可。

正常则删除即可。代码实现如下：

public void deleteBinding(Binding binding) throws MqException {
        
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
        
            // 该交换机没有绑定任何队列. 报错.
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

### ?对于单个消息 ###

直接调用相应方法即可，实现如下：

// 添加消息
    public void addMessage(Message message) {
        
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageId());
    }
    
    // 根据 id 查询消息
    public Message getMessage(String messageId) {
        
        return messageMap.get(messageId);
    }
    
    // 根据 id 删除消息
    public void removeMessage(String messageId) {
        
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

### ?对于队列与消息链表 ###

添加该对象时，我们需要根据传入的队列名查询相应的链表是否存在，若不存在，进行创建

若存在，则将该消息插入链表，对于此操作，为了线程安全，进行了加锁操作。

插入链表后，我们希望在消息中心也进行添加一下

代码实现如下：

// 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        
        // 把消息放到对应的队列数据结构中.
        // 先根据队列的名字, 找到该队列对应的消息链表.
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // 再把数据加到 messages 里面
        synchronized (messages) {
        
            messages.add(message);
        }
        // 在这里把该消息也往消息中心中插入一下. 假设如果 message 已经在消息中心存在, 重复插入也没关系.
        // 主要就是相同 messageId, 对应的 message 的内容一定是一样的. (服务器代码不会对 Message 内容做修改 basicProperties 和 body)
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中! messageId=" + message.getMessageId());
    }

从队列中取消息，首先进行判断，若对应队列链表为 null，则就不用取了，抛出异常

若不为null，我们还需要进行判断，链表内是否有元素，若无，则直接返回 null 即可，若不为，则进行头删操作(这就是为什么选择存储结构为链表的原因)并返回。为了线程安全，此操作我们也进行加锁。

public Message pollMessage(String queueName) {
        
        // 根据队列名, 查找一下, 对应的队列的消息链表.
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
        
            return null;
        }
        synchronized (messages) {
        
            // 如果没找到, 说明队列中没有任何消息.
            if (messages.size() == 0) {
        
                return null;
            }
            // 链表中有元素, 就进行头删.
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

获取指定队列中消息的个数，首先进行判断，若查询结果为null，则直接返回0即可

不为0时，我们直接查询并返回长度即可，注意此过程也为了线程安全，我们需要对其进行加锁操作。

代码实现如下：

// 获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
        
            // 队列中没有消息
            return 0;
        }
        synchronized (messages) {
        
            return messages.size();
        }
    }

### ?对于未确认消息 ###

依旧与上面逻辑差不多，进行判断并返回即可

代码实现如下：

// 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId=" + message.getMessageId());
    }
    
    // 删除未确认的消息(消息已经确认了)
    public void removeMessageWaitAck(String queueName, String messageId) {
        
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
        
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId=" + messageId);
    }
    
    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
        
            return null;
        }
        return messageHashMap.get(messageId);
    }

### ?从硬盘上读取数据 ###

这个方法就是从硬盘上读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中.

分为5步：

1.  清空之前的所有数据(内存上)
2.  恢复所有的队列数据
3.  恢复所有的绑定数据
4.  恢复所有的消息数据

注意!!

*  针对 “未确认的消息” 这部分内存中的数据, 不需要从硬盘恢复. 之前考虑硬盘存储的时候, 也没设定这一块.
 *  因为一旦在等待 ack 的过程中, 服务器重启了, 此时这些 “未被确认的消息”, 就恢复成 “未被取走的消息” .
 *  而这个消息在硬盘上存储的时候, 就是当做 “未被取走”

这五步的执行过程，只用调用前面对内存与硬盘封装的方法即可

代码执行如下：

// 这个方法就是从硬盘上读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中.
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        
        // 1. 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 2. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
        
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 3. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
        
            queueMap.put(queue.getName(), queue);
        }
        // 4. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
        
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 5. 恢复所有的消息数据
        //    遍历所有的队列, 根据每个队列的名字, 获取到所有的消息.
        for (MSGQueue queue : queues) {
        
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
        
                messageMap.put(message.getMessageId(), message);
            }
        }
        // 注意!! 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复. 之前考虑硬盘存储的时候, 也没设定这一块.
        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
        // 这个消息在硬盘上存储的时候, 就是当做 "未被取走"
    }

## ⭕总结 ##

关于《【消息队列开发】 实现MemoryDataCenter类——管理内存数据》就讲解到这儿，感谢大家的支持，欢迎各位留言交流以及批评指正，如果文章对您有帮助或者觉得作者写的还不错可以点一下关注，点赞，收藏支持一下

[b1d51e9b996f450bbd2aa0a78ac0966d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/15/56414b2e95f44e63928789e1ab0ec2f5.png