建设门户网站的意义,专业的网站开发公司,爱空间网站模板,软件开发模型的作用目录 一、内存数据管理
1.1、需求分析
1.2、实现 MemoryDataCenter 类
1.2.1、ConcurrentHashMap 数据管理
1.2.2、封装交换机操作
1.2.3、封装队列操作
1.2.4、封装绑定操作
1.2.5、封装消息操作
1.2.6、封装未确认消息操作
1.2.7、封装恢复数据操作 一、内存数据管理…目录 一、内存数据管理
1.1、需求分析
1.2、实现 MemoryDataCenter 类
1.2.1、ConcurrentHashMap 数据管理
1.2.2、封装交换机操作
1.2.3、封装队列操作
1.2.4、封装绑定操作
1.2.5、封装消息操作
1.2.6、封装未确认消息操作
1.2.7、封装恢复数据操作 一、内存数据管理 1.1、需求分析
当前已经使用 数据库 管理了 交换机、绑定、队列又使用 数据文件 管理了 消息.
最后还使用一个类将上述两部分整合在了一起对上层提供统一的一套接口.
但对于 MQ 来说是以内存存储数据为主硬盘存储数据为辅硬盘数据主要是为了持久化保存重启之后数据不丢失. 接下来就需要使用 内存 来管理上述数据~~ 这里我们主要使用 ConcurrentHashMap 来进行数据管理主要是因为线程安全问题.
交换机使用 ConcurrentHashMapkey 是 namevalue 是 Exchange 对象。
队列使用 ConcurrentHashMapkey 是 namevalue 是 MSGQueue 对象。
绑定使用嵌套的 ConcurrentHashMapkey 是 exchangeNamevalue 是一个 ConcurrentHashMapkey 是 queueNamevalue 是 Binding 对象。
消息使用 ConcurrentHashMapkey 是 messageId value 是 Message 对象。
队列和消息的关联关系使用嵌套的 ConcurrentHashMapkey 是 queueNamevalue 是一个 LinkedList每个元素是一个 Message 对象。
表示 “未被确认” 的消息使用嵌套的 ConcurrentHashMapkey 是 queueNamevalue 是 ConcurrentHashMapkey 是 messageIdvalue 是 Message 对象后续实现消息确认的逻辑需要根据 ack 响应的内容会提供一个确认的 messageId根据这个 messageId 来把上述结构中的 Message 对象找到并移除。 Ps此处实现的 MQ支持两种应答模式的 ACK 自动应答消费者取了元素整个消息就算是被应答了此时整个消息就可以被干掉了。手动应答消费者取了元素这个消息不算被应答需要消费者主动再调用一个 basicAck 方法此时才认为是真正应答了才能删除这个消息。 1.2、实现 MemoryDataCenter 类
1.2.1、ConcurrentHashMap 数据管理
这里就是用 ConcurrentHashMap 来对上述数据进行统一内存管理. //key 是 exchangeName, value 是 Exchange 对象private ConcurrentHashMapString, Exchange exchangeMap new ConcurrentHashMap();//key 是 queueName, value 是 MSGQueue 对象private ConcurrentHashMapString, MSGQueue queueMap new ConcurrentHashMap();//第一个 key 是 exchangeName第二个 key 是 queueNameprivate ConcurrentHashMapString, ConcurrentHashMapString, Binding bindingsMap new ConcurrentHashMap();//key 是 messageId value 是 Message 对象private ConcurrentHashMapString, Message messageMap new ConcurrentHashMap();//key 是 queueName value 是 Message 的链表private ConcurrentHashMapString, LinkedListMessage queueMessageMap new ConcurrentHashMap();// 第一个 key 是 queueName 第二个 key 是 messageIdprivate ConcurrentHashMapString, ConcurrentHashMapString, Message queueMessageWaitAckMap new ConcurrentHashMap();1.2.2、封装交换机操作
主要就是对 exchangeMap 插入、获取、删除交换机.
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);}1.2.3、封装队列操作
主要就是对 queueMap 插入、获取、删除队列. 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);}public void deleteQueue(String queueName) {queueMap.remove(queueName);System.out.println([MemoryDataCenter] 队列删除成功! queueName queueName);}1.2.4、封装绑定操作
这里值得注意的是加锁逻辑并不是加了锁就一定安全也不是说不加锁就一定不安全如果这段代码前后逻辑性很强需要打包成一个原子性的操作那就可以进行加锁如果不是那么强的因果就没必要因为加锁也是需要开销的加锁之后的锁竞争更是一个时间消耗。 public void insertBinding(Binding binding) throws MqException {
// ConcurrentHashMapString, Binding bindingMap bindingsMap.get(binding.getExchangeName());
// if(bindingMap null) {
// bindingMap new ConcurrentHashMap();
// bindingsMap.put(binding.getExchangeName(), bindingMap);
// }//上面这段逻辑可以用以下代码来替换ConcurrentHashMapString, 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());}/*** 获取绑定有两个版本* 1.根据 exchangeName 和 queueName 确定唯一一个 Binding* 2.根据 exchangeName 获取到所有的 Binding* param exchangeName* param queueName* return*/public Binding getBinding(String exchangeName, String queueName) throws MqException {ConcurrentHashMapString, Binding bindingMap bindingsMap.get(exchangeName);if(bindingMap null) {throw new MqException([MemoryDataCenter] 绑定不存在exchangeName exchangeName , queueName queueName);}return bindingMap.get(queueName);}public ConcurrentHashMapString, Binding getBindings(String exchangName) {return bindingsMap.get(exchangName);}public void deleteBinding(Binding binding) throws MqException {ConcurrentHashMapString, 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());}1.2.5、封装消息操作
这里值得注意的是 LinkedList 是线程不安全的要特殊处理. /*** 添加消息* param message*/public void addMessage(Message message) {messageMap.put(message.getMessageId(), message);System.out.println([MemoryDataCenter] 新消息添加成功messageId message.getMessageId());}/*** 根据 id 查询消息* param messageId*/public Message selectMessage(String messageId) {return messageMap.get(messageId);}/*** 根据 id 删除消息* param messageId*/public void removeMessage(String messageId) {messageMap.remove(messageId);System.out.println([MemoryDataCenter] 消息被移除messageId messageId);}/*** 发送消息到指定队列* param message*/public void sendMessage(MSGQueue queue, Message message) {//先根据队列名字找到指定的链表LinkedListMessage messages queueMessageMap.computeIfAbsent(queue.getName(), k - new LinkedList());//LinkedList 是线程不安全的synchronized (messages) {messages.add(message);}//这里把消息在消息中心也插入一下。即使 message 在消息中心存在也没关系//因为相同的 messageId 对应的 message 的内容一定是一样的(服务器不会修改 Message 的内容)addMessage(message);System.out.println([MemoryDataCenter] 消息被投递到队列当中messageId message.getMessageId());}/*** 从队列中取消息* param queueName* return*/public Message pollMessage(String queueName) {LinkedListMessage 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;}}/*** 获取指定队列的消息个数* param queueName* return*/public int getMessageCount(String queueName) {LinkedListMessage messages queueMessageMap.get(queueName);if(messages null) {return 0;}synchronized (messages) {return messages.size();}}1.2.6、封装未确认消息操作
“未被确认” 的消息使用嵌套的 ConcurrentHashMapkey 是 queueNamevalue 是 ConcurrentHashMapkey 是 messageIdvalue 是 Message 对象后续实现消息确认的逻辑需要根据 ack 响应的内容会提供一个确认的 messageId根据这个 messageId 来把上述结构中的 Message 对象找到并移除。 /*** 添加未确认的消息* param queueName* param message*/public void addMessageWaitAck(String queueName, Message message) {ConcurrentHashMapString, Message messageHashMap queueMessageWaitAckMap.computeIfAbsent(queueName,k - new ConcurrentHashMap());messageHashMap.put(message.getMessageId(), message);System.out.println([MemoryDataCenter] 消息进入待确认队列messageId message.getMessageId());}/*** 删除未确认的消息* param messageId*/public void removeMessageWaitAck(String queueName, String messageId) {ConcurrentHashMapString, 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) {ConcurrentHashMapString, Message messageHashMap queueMessageWaitAckMap.get(queueName);if(messageHashMap null) {return null;}return messageHashMap.get(messageId);}1.2.7、封装恢复数据操作
从硬盘上读取数据把硬盘中之前持久化存储的各个维度的数据都恢复到内存中.
public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {//1.先清空之前所有的数据exchangeMap.clear();queueMap.clear();bindingsMap.clear();messageMap.clear();queueMessageMap.clear();//2.恢复所有的交换机数据ListExchange exchanges diskDataCenter.selectAllExchanges();for(Exchange exchange : exchanges) {exchangeMap.put(exchange.getName(), exchange);}//3.恢复所有的队列数据ListMSGQueue queues diskDataCenter.selectAllQueue();for(MSGQueue queue : queues) {queueMap.put(queue.getName(), queue);}//4.恢复所有绑定数据ListBinding bindings diskDataCenter.selectAllBindings();for(Binding binding : bindings) {ConcurrentHashMapString, Binding bindingMap bindingsMap.computeIfAbsent(binding.getExchangeName(),k - new ConcurrentHashMap());bindingMap.put(binding.getQueueName(), binding);}//5.恢复所有的消息数据for(MSGQueue queue : queues) {LinkedListMessage messages diskDataCenter.loadAllMessagesFromQueue(queue.getName());queueMessageMap.put(queue.getName(), messages);//遍历所有的队列根据每个队列名字。来恢复所有消息for(Message message : messages) {messageMap.put(message.getMessageId(), message);}}}Ps“未确认的消息” 这部分数据不需要从硬盘中恢复之前硬盘存储也没有考虑过这里~ 一旦在等待 ack 的过程中服务器重启了这些 “未被确认的消息” 就恢复成了 “未被取走的消息”这个消息在硬盘上存储的时候就是当作 “未被取走”。
