目录
一、引言
二、MemoryDataCenter
1.设计数据结构
2.封装Exchange方法
3.封装MsgQueue方法
4.封装Binding方法
5.封装Message
6.实现待确定消息的管理
7.将数据从硬盘上恢复到内存中
三、测试MemoryDataCenter
1.准备工作
2.测试交换机
3.测试队列
4.测试绑定
5.测试消息
6.测试发送消息
7.测试待确认消息
8.测试从硬盘上读取消息到内存
四、总结
一、引言
上一篇文章介绍了统一硬盘处理的操作，这一篇文章我们就简单介绍一下数据在内存里面的管理。
二、MemoryDataCenter
1.设计数据结构
// 此处为了线程安全，我们使用ConcurrentHashMap这样的数据结构private ConcurrentHashMap<String,Exchange> exchangeMap = new ConcurrentHashMap<>();private ConcurrentHashMap<String,MsgQueue> queueMap = new ConcurrentHashMap<>();// 绑定：使用嵌套的HashMap，key是exchangeName，value也是一个HashMap（key是queueName，value是Binding对象）private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingMap = new ConcurrentHashMap<>();private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();// 表示“未被确认”的消息：使用嵌套的HashMap，key是queueName，value是HashMap（key是messageId，value是Message对象）// 存储当前队列中哪些消息被消费者取走，但是还没有应答private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();2.封装Exchange方法
/*封装交换机*/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);}3.封装MsgQueue方法
/*封装队列*/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);}4.封装Binding方法
/*封装绑定*/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());}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);}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());}5.封装Message
/*封装消息*/public void addMessage(Message message){messageMap.put(message.getMessageId(),message);System.out.println("[MemoryDataCenter] 新消息添加成功！messageId="+message.getMessageId());}public Message getMessage(String messageId){return messageMap.get(messageId);}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<>());synchronized (messages){messages.add(message);}addMessage(message);System.out.println("[MemoryDataCenter] 消息投递到队列成功！messageId="+message.getMessageId());}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;}}public int getMessageCount(String queueName){LinkedList<Message> messages = queueMessageMap.get(queueName);if(messages==null){return 0;}synchronized (messages){return messages.size();}}6.实现待确定消息的管理
/*实现待确定消息的管理*/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] 消息进入待确认队列！queueName="+queueName);}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);}7.将数据从硬盘上恢复到内存中
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);}}}三、测试MemoryDataCenter
1.准备工作
@SpringBootTest
public class MemoryDataCenterTests {private MemoryDataCenter memoryDataCenter = null;@BeforeEachpublic void setUp(){memoryDataCenter = new MemoryDataCenter();}@AfterEachpublic void tearDown(){memoryDataCenter = null;}private Exchange createTestExchange(String exchangeName){Exchange exchange = new Exchange();exchange.setName(exchangeName);exchange.setType(ExchangeType.DIRECT);exchange.setDurable(true);exchange.setAutoDelete(false);return exchange;}private MsgQueue createTestQueue(String queueName){MsgQueue queue = new MsgQueue();queue.setName(queueName);queue.setDurable(true);queue.setExclusive(false);queue.setAutoDelete(false);return queue;}private Message createTestMessage(String content){Message message = Message.createMessageWithId("testRoutingKey",null,content.getBytes());return message;}}2.测试交换机
@Testpublic void testExchange(){Exchange expectedExchange = createTestExchange("testExchange");memoryDataCenter.insertExchange(expectedExchange);Exchange actualExchange = memoryDataCenter.getExchange("testExchange");Assertions.assertEquals(expectedExchange,actualExchange);memoryDataCenter.deleteExchange("testExchange");actualExchange = memoryDataCenter.getExchange("testExchange");Assertions.assertNull(actualExchange);}3.测试队列
@Testpublic void testQueue(){MsgQueue expectedQueue = createTestQueue("testQueue");memoryDataCenter.insertQueue(expectedQueue);MsgQueue actualQueue = memoryDataCenter.getQueue("testQueue");Assertions.assertEquals(expectedQueue,actualQueue);memoryDataCenter.deleteQueue("testQueue");actualQueue = memoryDataCenter.getQueue("testQueue");Assertions.assertNull(actualQueue);}4.测试绑定
@Testpublic void testBinding() throws MqException {Binding expectedBinding = new Binding();expectedBinding.setExchangeName("testExchange");expectedBinding.setQueueName("testQueue");expectedBinding.setBindingKey("testBindingKey");memoryDataCenter.insertBinding(expectedBinding);Binding actualBinding = memoryDataCenter.getBinding("testExchange","testQueue");Assertions.assertEquals(expectedBinding,actualBinding);ConcurrentHashMap<String,Binding> bindingMap = memoryDataCenter.getBindings("testExchange");Assertions.assertEquals(1,bindingMap.size());Assertions.assertEquals(expectedBinding,bindingMap.get("testQueue"));memoryDataCenter.deleteBinding(expectedBinding);actualBinding = memoryDataCenter.getBinding("testExchange","testQueue");Assertions.assertNull(actualBinding);}5.测试消息
@Testpublic void testMessage(){Message expectedMessage = createTestMessage("testMessgae");memoryDataCenter.addMessage(expectedMessage);Message actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageId());Assertions.assertEquals(expectedMessage,actualMessage);memoryDataCenter.removeMessage(expectedMessage.getMessageId());actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageId());Assertions.assertNull(actualMessage);}6.测试发送消息
@Testpublic void testSendMessage(){MsgQueue queue = createTestQueue("testQueue");List<Message> expectedMessages = new ArrayList<>();for(int i=0;i<10;i++){Message message = createTestMessage("testMessage"+i);memoryDataCenter.sendMessage(queue,message);expectedMessages.add(message);}List<Message> actualMessage = new ArrayList<>();while(true){Message message = memoryDataCenter.pollMessage("testQueue");if(message==null){break;}actualMessage.add(message);}Assertions.assertEquals(expectedMessages.size(),actualMessage.size());for(int i=0;i<actualMessage.size();i++){Assertions.assertEquals(expectedMessages.get(i),actualMessage.get(i));}}7.测试待确认消息
@Testpublic void testMessageWaitAck(){Message expectedMessage = createTestMessage("expectedMessage");memoryDataCenter.addMessageWaitAck("testQueue",expectedMessage);Message actualMessage = memoryDataCenter.getMessageWaitAck("testQueue",expectedMessage.getMessageId());Assertions.assertEquals(expectedMessage,actualMessage);memoryDataCenter.removeMessageWaitAck("testQueue",expectedMessage.getMessageId());actualMessage = memoryDataCenter.getMessageWaitAck("testQueue",expectedMessage.getMessageId());Assertions.assertNull(actualMessage);}8.测试从硬盘上读取消息到内存
@Testpublic void testRecovery() throws IOException, MqException, ClassNotFoundException {SpringDemoMqApplication.context = SpringApplication.run(SpringDemoMqApplication.class);DiskDataCenter diskDataCenter = new DiskDataCenter();diskDataCenter.init();Exchange expectedExchange = createTestExchange("testExchange");diskDataCenter.insertExchange(expectedExchange);MsgQueue expectedQueue = createTestQueue("testQueue");diskDataCenter.insertQueue(expectedQueue);Binding expectedBinding = new Binding();expectedBinding.setExchangeName("testExchange");expectedBinding.setQueueName("testQueue");expectedBinding.setBindingKey("testBindingKey");diskDataCenter.insertBinding(expectedBinding);Message expectedMessage = createTestMessage("testContent");diskDataCenter.sendMessage(expectedQueue,expectedMessage);memoryDataCenter.recovery(diskDataCenter);Exchange actualExchange = memoryDataCenter.getExchange("testExchange");Assertions.assertEquals(expectedExchange.getName(),actualExchange.getName());Assertions.assertEquals(expectedExchange.getType(),actualExchange.getType());MsgQueue actualQueue = memoryDataCenter.getQueue("testQueue");Assertions.assertEquals(expectedQueue.getName(),actualQueue.getName());Assertions.assertEquals(expectedQueue.isExclusive(),actualQueue.isExclusive());Binding actualBinding = memoryDataCenter.getBinding("testExchange","testQueue");Assertions.assertEquals(expectedBinding.getExchangeName(),actualBinding.getExchangeName());Assertions.assertEquals(expectedBinding.getQueueName(),actualBinding.getQueueName());Message actualMessage = memoryDataCenter.pollMessage("testQueue");Assertions.assertEquals(expectedMessage.getMessageId(),actualMessage.getMessageId());Assertions.assertEquals(expectedMessage.getRoutingKey(),actualMessage.getRoutingKey());Assertions.assertEquals(expectedMessage.getDeliverMode(),actualMessage.getDeliverMode());Assertions.assertArrayEquals(expectedMessage.getBody(),actualMessage.getBody());SpringDemoMqApplication.context.close();File dataDir = new File("./data");FileUtils.deleteDirectory(dataDir);}四、总结
本篇文章主要介绍了一下如何在内存中管理数据，代码层面也是对相关概念进行操作以及测试，下一篇文章我们将学习虚拟主机设计内容，感谢观看！