模拟实现消息队列项目(系列4) -- 服务器模块(内存管理)

这篇具有很好参考价值的文章主要介绍了模拟实现消息队列项目(系列4) -- 服务器模块(内存管理)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

前言

1. 创建MemoryDataCenter

2. 封装Exchange 和 Queue方法

3. 封装Binding操作

4. 封装Message操作

4.1 封装消息中心集合messageMap

4.2 封装消息与队列的关系集合queueMessageMap的操作

5. 封装未确认消息集合waitMessage的操作

6. 从硬盘中恢复数据到内存中

7. MemoryDataCenter单元测试

结语


前言

        上一节我们总结了服务器模块的硬盘管理,将交换机,队列,绑定存书到Sqlite数据库中,将消息按照队进行创建文件存储在本地硬盘中.并且封装了对于数据库和文件的各种操作.实现了持久化的效果,但是实际的消息存储/转发,主要靠内存的结构.对于消息队列来说,内存部分是更关键的,内存速度更快,可以达到更高的并发.本节就对内存管理进行封装.本项目全部代码已上传Gitee,链接放在文章末尾,欢迎大家访问!


模拟实现消息队列项目(系列4) -- 服务器模块(内存管理),消息队列项目,服务器,spring boot,java,mq,rabbitmq

1. 创建MemoryDataCenter

路径:mqserver.datacenter.MemoryDataCenter

模拟实现消息队列项目(系列4) -- 服务器模块(内存管理),消息队列项目,服务器,spring boot,java,mq,rabbitmq

考虑到多线程的原因,我们将HashMap替换成ConcurrentHashMap (对每个哈希桶进行加锁,相对来说是线程安全的)

@Data
public class MemoryDataCenter {
    // 1. 交换机  多线程环境下使用,使用ConcurrentHashMap会相对线程安全
    //         key:ExchangeName,value:Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    // 2. 队列  key:QueueName,value:MSQueue对象
    private ConcurrentHashMap<String, MSQueue> queueMap = new ConcurrentHashMap<>();

    // 3. 绑定  key:ExchangeName,value:HashMap(key:QueueName,value:MSQueue对象)
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    // 4. 消息  key:MessageID,value:Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // 5. 消息和队列的映射关系 HashMap: key:QueueName,value:LinkedList(Message对象)
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    // 6. 未确认的消息  HashMap: key:QueueName,value:HashMap(key:MessageID,value:Message对象)
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();
}

2. 封装Exchange 和 Queue方法

主要就是插入和获取数据以及删除

   /**
     * 1. 针对内存中的交换机,队列设置操作
     */
    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 insertQueue(MSQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName=" + queue.getName());
    }

    public MSQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName=" + queueName);
    }

3. 封装Binding操作

这里呢之所以将绑定的操作单独列举出来,是因为存储绑定信息的数据结构是相对比较复杂的,是嵌套的HashMap.

对于插入绑定信息:

1, 首先按照交换机的名字进行查找,如果查找不到就进行创建一个HashMap的数据结构存储到含有绑定信息的HashMap中,如果存在的话在按照队列名字进行查找绑定信息,如果查找到了,说明改绑定信息已经插入过就不要进行插入了,如果没找到就进行插入操作.

2. 在上述查找和插入的操作比并不是原子的,所以我们要给是上述操作,按照bindingMap进行加锁.以保证我们的线程操作是安全的.

下述是相关对于绑定的操作的代码:

    /**
     * 2. 针对绑定进行操作
     */
    /**
     * 2.1插入绑定信息
     * @param binding
     * @throws MqException
     */
    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());
            }
            // 最后将绑定关系传入到bingMap中
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    /**
     * 2.2 获取绑定1: 根据exchangeName, queueName 获取唯一的绑定
     * @param exchangeName
     * @param queueName
     */
    public Binding getBinding(String exchangeName, String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            return null;
        }
        synchronized (bindingMap){
            // 防止当别的操作删除了这个队列的绑定信息,而导致的线程错误
            return bindingMap.get(queueName);
        }
    }

    /**
     * 2.3 获取绑定2: 根据exchangeName 查询所有绑定
     * @param exchangeName
     * @return
     */
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) throws MqException {
        if (bindingsMap.get(exchangeName) == null){
            return null;
        }
        return bindingsMap.get(exchangeName);
    }

    /**
     * 2.4 删除绑定关系(单个) 一个交换机对应的单个队列的绑定关系
     * @param binding
     * @throws MqException
     */
    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());
    }

    /**
     * 2.5 删除绑定关系(多个) 1个交换机对应的多个队列的绑定关系.
     */
    public void deleteBinding(String exchangeName){
        bindingsMap.remove(exchangeName);
    }

4. 封装Message操作

4.1 封装消息中心集合messageMap

  • 1. 添加消息到消息中心
  • 2. 根据消息ID查询消息
  • 3. 根据消息ID删除消息
 /**
     * 3. 针对消息进行操作
     */
    /**
     * 3.1 添加消息
     * @param message
     */
    public void addMessage(Message message) {
        messageMap.put(message.getMessageID(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageID());
    }

    /**
     * 3.2 根据 id 查询消息
     * @param messageId
     * @return
     */
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    /**
     * 3.3 根据 id 删除消息
     * @param messageId
     */
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

4.2 封装消息与队列的关系集合queueMessageMap的操作

  • 1. 发送消息到指定队列名字的队列
  • 2. 从指定队列中获取消息集合
  • 3. 获取指定队列名字队列中消息的个数
 /**
     * 4 针对消息和队列的关系进行操作
     */
    /**
     * 4.1 发送消息到指定队列
     * @param queue
     * @param message
     */
    public void sendMessage(MSQueue queue, Message message) {
        // 先根据队列的名字, 找到该队列对应的消息链表.
        // 先根据队列的名字进行查询,查不到就进行创建该队列对应的链表  // computeIfAbsent线程安全的
        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());
    }

    /**
     * 4.2 从指定队列名字中进行提取信息
     * @param queueName
     * @return
     */
    public Message pollMessage(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 队列中没有信息
        if (messages == null){
            System.out.println("[MemoryDataCenter] 该队列中没有信息! queueName=" + queueName);
            return null;
        }
        // 将队列进行头删除(提取信息)
        synchronized (messages){
            if (messages.size() == 0){
                System.out.println("[MemoryDataCenter] 该队列中没有信息! queueName=" + queueName);
                return null;
            }
            Message currentMessage = messages.remove(0); System.out.println
                    ("[MemoryDataCenter] 消息已经从队列中取出! queueName=" + queueName + ", MessageID=" + currentMessage.getMessageID() );
            return currentMessage;
        }
    }

    /**
     * 4.3 获取指定队列名字中消息的个数
     * @param queueName
     * @return
     */
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        // 队列中没有信息
        if (messages == null){
            System.out.println("[MemoryDataCenter] 该队列中没有信息! queueName=" + queueName);
            return 0;
        }
        // 将队列进行头删除(提取信息)
        synchronized (messages){
            if (messages.size() == 0){
                System.out.println("[MemoryDataCenter] 该队列中没有信息! queueName=" + queueName);
                return 0;
            }
            return messages.size();
        }
    }

5. 封装未确认消息集合waitMessage的操作

  • 1. 添加消息到等待确认队列
  • 2. 从指定未确认队列中删除消息
  • 3. 根据指定的消息ID与未确认队列名字获取消息内容
/**
     * 5. 未确认消息Map的操作
     */

    /**
     * 5.1 添加消息到指定等待确认队列
     * @param queueName
     * @param message
     */
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String,Message> waitMessage = queueMessageWaitAckMap
                .computeIfAbsent(queueName, k-> new ConcurrentHashMap<>());
            waitMessage.put(message.getMessageID(),message);
            System.out.println("[MemoryDataCenter] 消息进入等待确认队列! messageID=" + message.getMessageID());

    }

    /**
     * 5.2 从指定的未确认消息队列中进行删除消息
     * @param queueName
     * @param messageId
     */
    public void removeMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String,Message> waitMessage = queueMessageWaitAckMap.get(queueName);
        if (waitMessage == null){
            System.out.println("[MemoryDataCenter] 该队列为空! queueName=" + queueName);
            return;
        }

        waitMessage.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息已经从等待确认队列中移除! messageId=" + messageId);

    }

    /**
     * 5.3 根据指定消息ID从队列中进行获取信息
     * @param queueName
     * @param messageId
     * @return
     */
    public Message geMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String,Message> waitMessage = queueMessageWaitAckMap.get(queueName);
        if (waitMessage == null){
            System.out.println("[MemoryDataCenter] 该队列为空! queueName=" + queueName);
            return null;
        }
        return waitMessage.get(messageId);
    }

6. 从硬盘中恢复数据到内存中

使用之前封装过的diskDataCenter进行恢复数据.

1. 清空当前内存数据结构中的数据

2. 恢复所有的交换机,队列,绑定,消息数据,恢复消息数据的时候,要将消息中心和消息与队列的映射进行恢复.

/**
     * 6. 从硬盘中恢复数据到内存中 (使用之前封装好的管理硬盘的类进行实现)
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 1. 清空内存中各种数据信息
        queueMap.clear();
        exchangeMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 2. 恢复所有的交换机信息
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for (Exchange exchange :exchanges) {
             exchangeMap.put(exchange.getName(),exchange);
        }
        // 3. 恢复所有的队列信息
        List<MSQueue> queues = diskDataCenter.selectAllMSQueue();
        for (MSQueue msQueue :queues) {
            queueMap.put(msQueue.getName(),msQueue);
        }

        // 4. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for (Binding binding: bindings){
            ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.
                    computeIfAbsent(binding.getExchangeName(), k-> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }

        // 4. 恢复所有的消息数据

        // 4.1 遍历所有的队列
        // List<MSQueue> queues = diskDataCenter.selectAllMSQueue();
        for (MSQueue msQueue:queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(msQueue.getName());
            // 4.2 将获取的消息进行进行加入到队列
            queueMessageMap.put(msQueue.getName(),messages);
            // 4.3 将消息添加上到消息中心
            for (Message message : messages) {
                messageMap.put(message.getMessageID(),message);
            }
        }

7. MemoryDataCenter单元测试

package com.example.demo.mqserver.datacenter;

import com.example.demo.DemoApplication;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.core.*;
import org.apache.tomcat.util.http.fileupload.FileUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import static org.junit.jupiter.api.Assertions.*;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: YAO
 * Date: 2023-07-31
 * Time: 10:30
 */
@SpringBootTest
class MemoryDataCenterTest {
    MemoryDataCenter memoryDataCenter = null;

    @BeforeEach
    void setUp() {
        memoryDataCenter = new MemoryDataCenter();
    }

    @AfterEach
    void tearDown() {
        memoryDataCenter = null;
    }

    // 创建一个测试交换机
    private Exchange createTestExchange(String exchangeName) {
        Exchange exchange = new Exchange();
        exchange.setName(exchangeName);
        exchange.setType(ExchangeType.DIRECT);
        exchange.setAutoDelete(false);
        exchange.setDurable(true);
        return exchange;
    }

    // 创建一个测试队列
    private MSQueue createTestQueue(String queueName) {
        MSQueue queue = new MSQueue();
        queue.setName(queueName);
        queue.setDurable(true);
        queue.setExclusive(false);
        queue.setAutoDelete(false);
        return queue;
    }

    /**
     * 1. 针对交换机进行操作
     */
    @Test
    public void testExchange(){
        // 1. 创建交换机进行插入
        Exchange expectExchange = createTestExchange("testExchange");
        memoryDataCenter.insertExchange(expectExchange);
        // 2. 查询交换机
        Exchange actualExchange = memoryDataCenter.getExchange("testExchange");
        // 比较内存中的引用是否是同一个引用
        Assertions.assertEquals(expectExchange,actualExchange);
        // 3. 删除交换机
        memoryDataCenter.deleteExchange("testExchange");
        // 4. 查询交换机,比较结果
        actualExchange = memoryDataCenter.getExchange("testExchange");
        Assertions.assertNull(actualExchange);
    }

    /**
     * 2. 针对队列进行操作
     */
    @Test
    public void testQueue(){
        // 1. 创建交换机进行插入
        MSQueue expectQueue = createTestQueue("testQueue");
        memoryDataCenter.insertQueue(expectQueue);
        // 2. 查询交换机
        MSQueue actualQueue = memoryDataCenter.getQueue("testQueue");
        // 比较内存中的引用是否是同一个引用
        Assertions.assertEquals(expectQueue,actualQueue);
        // 3. 删除交换机
        memoryDataCenter.deleteQueue("testQueue");
        // 4. 查询交换机,比较结果
        actualQueue = memoryDataCenter.getQueue("testQueue");
        Assertions.assertNull(actualQueue);
    }

    /**
     * 3. 针对绑定进行测试
     */
    @Test
    public void testBinding() throws MqException {
        // 1.创建绑定并加入到集合中
        Binding expectedBinding = new Binding();
        expectedBinding.setExchangeName("testExchange");
        expectedBinding.setQueueName("testQueue");
        memoryDataCenter.insertBinding(expectedBinding);
        // 2. 查询绑定(单个)
        Binding actualBinding = memoryDataCenter.getBinding("testExchange","testQueue");
        Assertions.assertEquals(expectedBinding,actualBinding);
        // 2.1 查询所有的绑定
        ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings("testExchange");
        Assertions.assertEquals(1, bindingMap.size());
        Assertions.assertEquals(expectedBinding, bindingMap.get("testQueue"));

        // 3. 删除绑定
        memoryDataCenter.deleteBinding("testExchange");
        actualBinding = memoryDataCenter.getBinding("testExchange","testQueue");
        Assertions.assertNull(actualBinding);
        bindingMap = memoryDataCenter.getBindings("testExchange");
        Assertions.assertNull(bindingMap);
    }

    private Message createTestMessage(String content) {
        Message message = Message.createMessageWithId("testRoutingKey", null, content.getBytes());
        return message;
    }
    /**
     * 4. 针对消息进行测试
     */
    @Test
    public void testMessage(){
        // 1. 创建消息并插入
        Message expectedMessage = createTestMessage("testMessage");
        memoryDataCenter.addMessage(expectedMessage);

        // 2. 查询消息并比较
        Message actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageID());
        Assertions.assertEquals(expectedMessage, actualMessage);

        // 4. 删除消息
        memoryDataCenter.removeMessage(expectedMessage.getMessageID());

        // 5. 查询消息并比较
        actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageID());
        Assertions.assertNull(actualMessage);
    }

    /**
     * 5. 测试将消息发送到对列中
     */
    @Test
    public void sendMessage(){
        // 1. 创建一个队列. 创建10条消息,进行插入到队列
        MSQueue expectQueue = createTestQueue("testQueue");
        List<Message> expectMessage = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message message = createTestMessage("testMessage" + i);
            memoryDataCenter.sendMessage(expectQueue,message);
            expectMessage.add(message);
        }
        // 2.从队列进行取出消息
        List<Message> actualMessage = new ArrayList<>();
        while (true){
            Message message = memoryDataCenter.pollMessage("testQueue");
            if (message == null){
                break;
            }
            actualMessage.add(message);
        }
        // 3. 比较消息前后是否一致
        Assertions.assertEquals(expectMessage.size(),actualMessage.size());
        for (int i = 0; i < expectMessage.size(); i++) {
            Assertions.assertEquals(expectMessage.get(i),actualMessage.get(i));
        }
    }

    /**
     * 6. 测试未被确认的消息
     */
    @Test
    public void testMessageWaitAck(){
        // 1. 创建消息,插入到未被确认的队列中
        Message expectedMessage = createTestMessage("expectedMessage");
        memoryDataCenter.addMessageWaitAck("testQueue", expectedMessage);

        // 2. 获取消息从未被确认的队列中
        Message actualMessage = memoryDataCenter.geMessageWaitAck("testQueue", expectedMessage.getMessageID());
        Assertions.assertEquals(expectedMessage, actualMessage);

        // 3. 从未被确认的队列中进行删除消息
        memoryDataCenter.removeMessageWaitAck("testQueue", expectedMessage.getMessageID());
        // 4. 比较删除之后的队列是否还有消息
        actualMessage = memoryDataCenter.geMessageWaitAck("testQueue", expectedMessage.getMessageID());
        Assertions.assertNull(actualMessage);
    }

    /**
     * 7. 测试从硬盘中恢复数据到内存
     */
    @Test
    public void testRecovery() throws IOException, MqException, ClassNotFoundException {
        // 由于后续需要进行数据库操作, 依赖 MyBatis. 就需要先启动 SpringApplication, 这样才能进行后续的数据库操作.
        DemoApplication.context = SpringApplication.run(DemoApplication.class);

        // 1. 在硬盘上构造好数据
        DiskDataCenter diskDataCenter = new DiskDataCenter();
        diskDataCenter.init();

        // 构造交换机
        Exchange expectedExchange = createTestExchange("testExchange");
        diskDataCenter.insertExchange(expectedExchange);

        // 构造队列
        MSQueue 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);

        // 2. 执行恢复操作
        memoryDataCenter.recovery(diskDataCenter);

        // 3. 对比结果
        Exchange actualExchange = memoryDataCenter.getExchange("testExchange");
        Assertions.assertEquals(expectedExchange.getName(), actualExchange.getName());
        Assertions.assertEquals(expectedExchange.getType(), actualExchange.getType());
        Assertions.assertEquals(expectedExchange.isDurable(), actualExchange.isDurable());
        Assertions.assertEquals(expectedExchange.isAutoDelete(), actualExchange.isAutoDelete());

        MSQueue actualQueue = memoryDataCenter.getQueue("testQueue");
        Assertions.assertEquals(expectedQueue.getName(), actualQueue.getName());
        Assertions.assertEquals(expectedQueue.isDurable(), actualQueue.isDurable());
        Assertions.assertEquals(expectedQueue.isAutoDelete(), actualQueue.isAutoDelete());
        Assertions.assertEquals(expectedQueue.isExclusive(), actualQueue.isExclusive());

        Binding actualBinding = memoryDataCenter.getBinding("testExchange", "testQueue");
        Assertions.assertEquals(expectedBinding.getExchangeName(), actualBinding.getExchangeName());
        Assertions.assertEquals(expectedBinding.getQueueName(), actualBinding.getQueueName());
        Assertions.assertEquals(expectedBinding.getBindingKey(), actualBinding.getBindingKey());

        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());

        // 4. 清理硬盘的数据, 把整个 data 目录里的内容都删掉(包含了 meta.db 和 队列的目录).
        DemoApplication.context.close();
        File dataDir = new File("./data");
        FileUtils.deleteDirectory(dataDir);
    }
}

模拟实现消息队列项目(系列4) -- 服务器模块(内存管理),消息队列项目,服务器,spring boot,java,mq,rabbitmq


结语

        以上内容就是针对内存管理的封装,主要是设计了6中数据机构进行存储交换机 队列 绑定 消息 消息和队列的映射 未确认信息.后续对数据进行操作的时候会更加具有效率.这样我们虚拟主机中两大核心部分:硬盘管理和内存管理都总结完成,下一节会对上述两种操作进一步封装到(VirtualHost)中,然后正式的提出消息队列服务器BrokerServer这个概念,对其进行完善和功能封装.请持续关注,谢谢!!!

完整的项目代码已上传Gitee,欢迎大家访问.👇👇👇

模拟实现消息队列https://gitee.com/yao-fa/advanced-java-ee/tree/master/My-mq模拟实现消息队列项目(系列4) -- 服务器模块(内存管理),消息队列项目,服务器,spring boot,java,mq,rabbitmq文章来源地址https://www.toymoban.com/news/detail-634032.html

到了这里,关于模拟实现消息队列项目(系列4) -- 服务器模块(内存管理)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Windows Server 2012 R2服务器Microsoft 消息队列远程代码执行漏洞CVE-2023-21554补丁KB5025288的安装及问题解决

    近日,系统安全扫描中发现Windows Server 2012 R2服务器存在Microsoft 消息队列远程代码执行漏洞。本文记录补丁安装中遇到的“此更新不适用于你的计算机”问题及解决办法。 一、问题描述: 1、系统安全扫描中发现Windows Server 2012 R2服务器存在Microsoft 消息队列远程代码执行漏洞,

    2024年02月10日
    浏览(53)
  • Web服务器实现|基于阻塞队列线程池的Http服务器|线程控制|Http协议

    代码地址:WebServer_GitHub_Addr 摘要 本实验通过C++语言,实现了一个基于阻塞队列线程池的多线程Web服务器。该服务器支持通过http协议发送报文,跨主机抓取服务器上特定资源。与此同时,该Web服务器后台通过C++语言,通过原生系统线程调用 pthread.h ,实现了一个 基于阻塞队列

    2024年02月07日
    浏览(65)
  • 极光Java 版本服务器端实现别名消息推送

    REST API 文档:

    2024年02月15日
    浏览(48)
  • Qt实现客户端与服务器消息发送

    里用Qt来简单设计实现一个场景,即: (1)两端:服务器QtServer和客户端QtClient (2)功能:服务端连接客户端,两者能够互相发送消息,传送文件,并且显示文件传送进度。 环境:VS20013 + Qt5.11.2 + Qt设计师 先看效果: 客户端与服务器的基本概念不说了,关于TCP通信的三次握

    2024年02月11日
    浏览(51)
  • 【网络】UDP网络服务器简单模拟实现

    【网络】UDP网络服务器简单模拟实现 UDP的封装 : UDP网络服务器模拟实现:主要分为makefile文件进行编译 UDP客户端 :udpClient.cc(客户端的调用),udpClient.hpp(客户端的实现) UDP服务端 :udpServer.cc(服务端的调用),udpServer.hpp(服务端的实现) 创建makefile文件: makefile里可以定义变

    2024年02月08日
    浏览(49)
  • 【Python】OPC UA模拟服务器实现

     在工业自动化和物联网(IoT)领域,OPC UA(开放平台通信统一架构)已经成为一种广泛采用的数据交换标准。它提供了一种安全、可靠且独立于平台的方式来访问实时数据。在本文中,我们将探讨如何使用Python和OPC UA库来创建一个高效的数据服务器,该服务器能够从CSV文件

    2024年04月29日
    浏览(68)
  • 微信小程序实现订阅消息功能(Node服务器篇)

            * 源码已经上传到资源处,需要的话点击跳转下载 |  源码下载         在上一篇内容当中在微信小程序中实现订阅消息功能,都在客户端(小程序)中来实现的,在客户端中模拟了服务器端来进行发送订阅消息的功能,那么本篇就将上一篇内容中仅在客户端中实现

    2024年02月03日
    浏览(64)
  • 基于RabbitMQ的模拟消息队列之二---创建项目及核心类

    创建一个SpringBoot项目,环境:JDK8,添加依赖:Spring Web、MyBatis FrameWork(最主要) 2.核心类 在mqserver包中添加一个包,名字为core,表示核心类。 Exchange ExchangeType MSGQueue (为了区分Queue) Binding Message BasicProperties

    2024年02月11日
    浏览(47)
  • 使用Java服务器实现UDP消息的发送和接收(多线程)

    在本篇博客中,我们将介绍如何使用Java服务器来实现UDP消息的发送和接收,并通过多线程的方式来处理并发请求。UDP(User Datagram Protocol)是一种无连接、不可靠的传输协议,适合于实时性要求高的应用场景,如实时游戏、语音通信等。 步骤: 首先,我们需要导入Java提供的

    2024年02月12日
    浏览(47)
  • SSE与WebSocket分别实现服务器发送消息通知(Golang、Gin)

    服务端推送,也称为消息推送或通知推送,是一种允许应用服务器主动将信息发送到客户端的能力,为客户端提供了实时的信息更新和通知,增强了用户体验。 服务端推送的背景与需求主要基于以下几个诉求: 实时通知:在很多情况下,用户期望实时接收到应用的通知,如

    2024年02月03日
    浏览(52)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包