四. 项⽬创建

创建 SpringBoot 项⽬.
使⽤ SpringBoot 2 系列版本, Java 8.
依赖引⼊ Spring Web 和 MyBatis

五. 创建核⼼类

创建包 mqserver.core

创建 Exchange（名字、类型、持久化）

package com.example.mq.mqserver.core;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.HashMap;
import java.util.Map;

/*
 * 这个类表示一个交换机
 */
public class Exchange {
    // 此处使用 name 来作为交换机的身份标识. (唯一的)
    private String name;
    // 交换机类型, DIRECT, FANOUT, TOPIC
    private ExchangeType type = ExchangeType.DIRECT;
    // 该交换机是否要持久化存储. true 表示需要持久化; false 表示不必持久化.
    private boolean durable = false;
    // 如果当前交换机, 没人使用了, 就会自动被删除.
    // 这个属性暂时先列在这里, 后续的代码中并没有真的实现这个自动删除功能~~ (RabbitMQ 是有的)
    private boolean autoDelete = false;
    // arguments 表示的是创建交换机时指定的一些额外的参数选项. 后续代码中并没有真的实现对应的功能, 先列出来. (RabbitMQ 也是有的)
    // 为了把这个 arguments 存到数据库中, 就需要把 Map 转成 json 格式的字符串.
    private Map<String, Object> arguments = new HashMap<>();

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public ExchangeType getType() {
        return type;
    }

    public void setType(ExchangeType type) {
        this.type = type;
    }

    public boolean isDurable() {
        return durable;
    }

    public void setDurable(boolean durable) {
        this.durable = durable;
    }

    public boolean isAutoDelete() {
        return autoDelete;
    }

    public void setAutoDelete(boolean autoDelete) {
        this.autoDelete = autoDelete;
    }

    // 这里的 get set 用于和数据库交互使用.
    public String getArguments() {
        // 是把当前的 arguments 参数, 从 Map 转成 String (JSON)
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.writeValueAsString(arguments);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        // 如果代码真异常了, 返回一个空的 json 字符串就 ok
        return "{}";
    }

    // 这个方法, 是从数据库读数据之后, 构造 Exchange 对象, 会自动调用到
    public void setArguments(String argumentsJson) {
        // 把参数中的 argumentsJson 按照 JSON 格式解析, 转成
        // 上述的 Map 对象
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            this.arguments = objectMapper.readValue(argumentsJson, new TypeReference<HashMap<String, Object>>() {});
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    // 在这里针对 arguments, 再提供一组 getter setter , 用来去更方便的获取/设置这里的键值对.
    // 这一组在 java 代码内部使用 (比如测试的时候)
    public Object getArguments(String key) {
        return arguments.get(key);
    }

    public void setArguments(String key, Object value) {
        arguments.put(key, value);
    }

    public void setArguments(Map<String, Object> arguments) {
        this.arguments = arguments;
    }
}

创建 MSGQueue（名字、持久化、独占标识）

package com.example.mq.mqserver.core;

import com.example.mq.common.ConsumerEnv;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.boot.autoconfigure.aop.AopAutoConfiguration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/*
 * 这个类表示一个存储消息的队列
 * MSG => Message
 */
public class MSGQueue {
    // 表示队列的身份标识.
    private String name;
    // 表示队列是否持久化, true 表示持久化保存, false 表示不持久化.
    private boolean durable = false;
    // 这个属性为 true, 表示这个队列只能被一个消费者使用(别人用不了). 如果为 false 则是大家都能使用
    // 这个 独占 功能, 也是先把字段列在这里, 具体的独占功能暂时先不实现.
    private boolean exclusive = false;
    // 为 true 表示没有人使用之后, 就自动删除. false 则是不会自动删除.
    // 这个 自动删除 功能, 也是先把字段列在这里, 具体的独占功能暂时先不实现.
    private boolean autoDelete = false;
    // 也是表示扩展参数. 当前也是先列在这里, 先暂时不实现
    private Map<String, Object> arguments = new HashMap<>();
    // 当前队列都有哪些消费者订阅了.
    private List<ConsumerEnv> consumerEnvList = new ArrayList<>();
    // 记录当前取到了第几个消费者. 方便实现轮询策略.
    private AtomicInteger consumerSeq = new AtomicInteger(0);

    // 添加一个新的订阅者
    public void addConsumerEnv(ConsumerEnv consumerEnv) {
        consumerEnvList.add(consumerEnv);
    }

    // 订阅者的删除暂时先不考虑.
    // 挑选一个订阅者, 用来处理当前的消息. (按照轮询的方式)
    public ConsumerEnv chooseConsumer() {
        if (consumerEnvList.size() == 0) {
            // 该队列没有人订阅的
            return null;
        }
        // 计算一下当前要取的元素的下标.
        int index = consumerSeq.get() % consumerEnvList.size();
        consumerSeq.getAndIncrement();
        return consumerEnvList.get(index);
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public boolean isDurable() {
        return durable;
    }

    public void setDurable(boolean durable) {
        this.durable = durable;
    }

    public boolean isExclusive() {
        return exclusive;
    }

    public void setExclusive(boolean exclusive) {
        this.exclusive = exclusive;
    }

    public boolean isAutoDelete() {
        return autoDelete;
    }

    public void setAutoDelete(boolean autoDelete) {
        this.autoDelete = autoDelete;
    }

    public String getArguments() {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.writeValueAsString(arguments);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "{}";
    }

    public void setArguments(String argumentsJson) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            this.arguments = objectMapper.readValue(argumentsJson, new TypeReference<HashMap<String, Object>>() {});
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    public Object getArguments(String key) {
        return arguments.get(key);
    }

    public void setArguments(String key, Object value) {
        arguments.put(key, value);
    }

    public void setArguments(Map<String, Object> arguments) {
        this.arguments = arguments;
    }
}

创建 Binding（交换机名字、队列名字、bindingKey用于与routingKey匹配）

package com.example.mq.mqserver.core;

/*
 * 表示队列和交换机之间的关联关系
 */
public class Binding {
    private String exchangeName;
    private String queueName;
    // bindingKey 就是在出题, 要求领红包的人要画个 "桌子" 出来~~
    private String bindingKey;

    // Binding 这个东西, 依附于 Exchange 和 Queue 的!!!
    // 比如, 对于持久化来说, 如果 Exchange 和 Queue 任何一个都没有持久化,
    // 此时你针对 Binding 持久化是没有任何意义的


    public String getExchangeName() {
        return exchangeName;
    }

    public void setExchangeName(String exchangeName) {
        this.exchangeName = exchangeName;
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public String getBindingKey() {
        return bindingKey;
    }

    public void setBindingKey(String bindingKey) {
        this.bindingKey = bindingKey;
    }
}

创建 Message（序列化、消息属性、消息体、起始位置和结束位置、有效、工厂方法）

package com.example.mq.mqserver.core;

import java.io.Serializable;
import java.util.Arrays;
import java.util.UUID;

/*
 * 表示一个要传递的消息
 * 注意!!! 此处的 Message 对象, 是需要能够在网络上传输, 并且也需要能写入到文件中.
 * 此时就需要针对 Message 进行序列化和反序列化.
 * 此处使用 标准库 自带的 序列化/反序列化 操作.
 */
public class Message implements Serializable {
    // 这两个属性是 Message 最核心的部分.
    private BasicProperties basicProperties = new BasicProperties();
    private byte[] body;

    // 下面的属性则是辅助用的属性.
    // Message 后续会存储到文件中(如果持久化的话).
    // 一个文件中会存储很多的消息. 如何找到某个消息, 在文件中的具体位置呢?
    // 使用下列的两个偏移量来进行表示. [offsetBeg, offsetEnd)
    // 这俩属性并不需要被序列化保存到文件中~~ 此时消息一旦被写入文件之后, 所在的位置就固定了. 并不需要单独存储.
    // 这俩属性存在的目的, 主要就是为了让内存中的 Message 对象, 能够快速找到对应的硬盘上的 Message 的位置.
    private transient long offsetBeg = 0;  // 消息数据的开头距离文件开头的位置偏移(字节)
    private transient long offsetEnd = 0;  // 消息数据的结尾距离文件开头的位置偏移(字节)
    // 使用这个属性表示该消息在文件中是否是有效消息. (针对文件中的消息, 如果删除, 使用逻辑删除的方式)
    // 0x1 表示有效. 0x0 表示无效.
    private byte isValid = 0x1;

    // 创建一个工厂方法, 让工厂方法帮我们封装一下创建 Message 对象的过程.
    // 这个方法中创建的 Message 对象, 会自动生成唯一的 MessageId
    // 万一 routingKey 和 basicProperties 里的 routingKey 冲突, 以外面的为主.
    public static Message createMessageWithId(String routingKey, BasicProperties basicProperties, byte[] body) {
        Message message = new Message();
        if (basicProperties != null) {
            message.setBasicProperties(basicProperties);
        }
        // 此处生成的 MessageId 以 M- 作为前缀.
        message.setMessageId("M-" + UUID.randomUUID());
        message.setRoutingKey(routingKey);
        message.body = body;
        // 此处是把 body 和 basicProperties 先设置出来. 他俩是 Message 的核心内容.
        // 而 offsetBeg, offsetEnd, isValid, 则是消息持久化的时候才会用到. 在把消息写入文件之前再进行设定.
        // 此处只是在内存中创建一个 Message 对象.
        return message;
    }

    public String getMessageId() {
        return basicProperties.getMessageId();
    }

    public void setMessageId(String messageId) {
        basicProperties.setMessageId(messageId);
    }

    public String getRoutingKey() {
        return basicProperties.getRoutingKey();
    }

    public void setRoutingKey(String routingKey) {
        basicProperties.setRoutingKey(routingKey);
    }

    public int getDeliverMode() {
        return basicProperties.getDeliverMode();
    }

    public void setDeliverMode(int mode) {
        basicProperties.setDeliverMode(mode);
    }

    public BasicProperties getBasicProperties() {
        return basicProperties;
    }

    public void setBasicProperties(BasicProperties basicProperties) {
        this.basicProperties = basicProperties;
    }

    public byte[] getBody() {
        return body;
    }

    public void setBody(byte[] body) {
        this.body = body;
    }

    public long getOffsetBeg() {
        return offsetBeg;
    }

    public void setOffsetBeg(long offsetBeg) {
        this.offsetBeg = offsetBeg;
    }

    public long getOffsetEnd() {
        return offsetEnd;
    }

    public void setOffsetEnd(long offsetEnd) {
        this.offsetEnd = offsetEnd;
    }

    public byte getIsValid() {
        return isValid;
    }

    public void setIsValid(byte isValid) {
        this.isValid = isValid;
    }

    @Override
    public String toString() {
        return "Message{" +
                "basicProperties=" + basicProperties +
                ", body=" + Arrays.toString(body) +
                ", offsetBeg=" + offsetBeg +
                ", offsetEnd=" + offsetEnd +
                ", isValid=" + isValid +
                '}';
    }
}

六. 数据库设计

对于 Exchange, MSGQueue, Binding, 我们使⽤数据库进⾏持久化保存.
此处我们使⽤的数据库是 SQLite, 是⼀个更轻量的数据库.
SQLite 只是⼀个动态库(当然, 官⽅也提供了可执⾏程序 exe), 我们在 Java 中直接引⼊ SQLite 依赖, 即
可直接使⽤, 不必安装其他的软件

配置 sqlite

引⼊ pom.xml 依赖

配置数据源 application.yml

Username 和 password 空着即可.
此处我们约定, 把数据库⽂件放到 ./data/meta.db 中.
SQLite 只是把数据单纯的存储到⼀个⽂件中. ⾮常简单⽅便.

实现创建表

@Mapper
public interface MetaMapper {
    // 提供三个核心建表方法
    void createExchangeTable();
    void createQueueTable();
    void createBindingTable();
}

本⾝ MyBatis 针对 MySQL / Oracle ⽀持执⾏多个 SQL 语句的, 但是针对 SQLite 是不⽀持的, 只能写
成多个⽅法

    <update id="createExchangeTable">
        create table if not exists exchange (
            name varchar(50) primary key,
            type int,
            durable boolean,
            autoDelete boolean,
            arguments varchar(1024)
        );
    </update>
    
    <update id="createQueueTable">
        create table if not exists queue (
            name varchar(50) primary key,
            durable boolean,
            exclusive boolean,
            autoDelete boolean,
            arguments varchar(1024)
        );
    </update>

    <update id="createBindingTable">
        create table if not exists binding (
            exchangeName varchar(50),
            queueName varchar(50),
            bindingKey varchar(256)
        );
    </update>

实现数据库基本操作

    // 针对上述三个基本概念, 进行 插入 和 删除
    void insertExchange(Exchange exchange);
    List<Exchange> selectAllExchanges();
    void deleteExchange(String exchangeName);
    void insertQueue(MSGQueue queue);
    List<MSGQueue> selectAllQueues();
    void deleteQueue(String queueName);
    void insertBinding(Binding binding);
    List<Binding> selectAllBindings();
    void deleteBinding(Binding binding);

    <insert id="insertExchange" parameterType="com.example.mq.mqserver.core.Exchange">
        insert into exchange values(#{name}, #{type}, #{durable}, #{autoDelete}, #{arguments});
    </insert>
    
    <select id="selectAllExchanges" resultType="com.example.mq.mqserver.core.Exchange">
        select * from exchange;
    </select>
    
    <delete id="deleteExchange" parameterType="java.lang.String">
        delete from exchange where name = #{exchangeName};
    </delete>

    <insert id="insertQueue" parameterType="com.example.mq.mqserver.core.MSGQueue">
        insert into queue values(#{name}, #{durable}, #{exclusive}, #{autoDelete}, #{arguments});
    </insert>
    
    <select id="selectAllQueues" resultType="com.example.mq.mqserver.core.MSGQueue">
        select * from queue;
    </select>
    
    <delete id="deleteQueue" parameterType="java.lang.String">
        delete from queue where name = #{queueName};
    </delete>
    
    <insert id="insertBinding" parameterType="com.example.mq.mqserver.core.Binding">
        insert into binding values(#{exchangeName}, #{queueName}, #{bindingKey});
    </insert>

    <select id="selectAllBindings" resultType="com.example.mq.mqserver.core.Binding">
        select * from binding;
    </select>
    
    <delete id="deleteBinding" parameterType="com.example.mq.mqserver.core.Binding">
        delete from binding where exchangeName = #{exchangeName} and queueName = #{queueName};
    </delete>

实现 DataBaseManager

package com.example.mq.mqserver.datacenter;

import com.example.mq.MqApplication;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.ExchangeType;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.mapper.MetaMapper;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.File;
import java.util.List;

/*
 * 通过这个类, 来整合上述的数据库操作.
 */
public class DataBaseManager {
    // 要做的是从 Spring 中拿到现成的对象
    private MetaMapper metaMapper;

    // 针对数据库进行初始化
    public void init() {
        // 手动的获取到 MetaMapper
        metaMapper = MqApplication.context.getBean(MetaMapper.class);

        if (!checkDBExists()) {
            // 数据库不存在, 就进行建建库表操作
            // 先创建一个 data 目录
            File dataDir = new File("./data");
            dataDir.mkdirs();
            // 创建数据表
            createTable();
            // 插入默认数据
            createDefaultData();
            System.out.println("[DataBaseManager] 数据库初始化完成!");
        } else {
            // 数据库已经存在了, 啥都不必做即可
            System.out.println("[DataBaseManager] 数据库已经存在!");
        }
    }

    public void deleteDB() {
        File file = new File("./data/meta.db");
        boolean ret = file.delete();
        if (ret) {
            System.out.println("[DataBaseManager] 删除数据库文件成功!");
        } else {
            System.out.println("[DataBaseManager] 删除数据库文件失败!");
        }

        File dataDir = new File("./data");
        // 使用 delete 删除目录的时候, 需要保证目录是空的.
        ret = dataDir.delete();
        if (ret) {
            System.out.println("[DataBaseManager] 删除数据库目录成功!");
        } else {
            System.out.println("[DataBaseManager] 删除数据库目录失败!");
        }
    }

    private boolean checkDBExists() {
        File file = new File("./data/meta.db");
        if (file.exists()) {
            return true;
        }
        return false;
    }

    // 这个方法用来建表.
    // 建库操作并不需要手动执行. (不需要手动创建 meta.db 文件)
    // 首次执行这里的数据库操作的时候, 就会自动的创建出 meta.db 文件来 (MyBatis 帮我们完成的)
    private void createTable() {
        metaMapper.createExchangeTable();
        metaMapper.createQueueTable();
        metaMapper.createBindingTable();
        System.out.println("[DataBaseManager] 创建表完成!");
    }

    // 给数据库表中, 添加默认的数据.
    // 此处主要是添加一个默认的交换机.
    // RabbitMQ 里有一个这样的设定: 带有一个 匿名 的交换机, 类型是 DIRECT.
    private void createDefaultData() {
        // 构造一个默认的交换机.
        Exchange exchange = new Exchange();
        exchange.setName("");
        exchange.setType(ExchangeType.DIRECT);
        exchange.setDurable(true);
        exchange.setAutoDelete(false);
        metaMapper.insertExchange(exchange);
        System.out.println("[DataBaseManager] 创建初始数据完成!");
    }

    // 把其他的数据库的操作, 也在这个类中封装一下.
    public void insertExchange(Exchange exchange) {
        metaMapper.insertExchange(exchange);
    }

    public List<Exchange> selectAllExchanges() {
        return metaMapper.selectAllExchanges();
    }

    public void deleteExchange(String exchangeName) {
        metaMapper.deleteExchange(exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        metaMapper.insertQueue(queue);
    }

    public List<MSGQueue> selectAllQueues() {
        return metaMapper.selectAllQueues();
    }

    public void deleteQueue(String queueName) {
        metaMapper.deleteQueue(queueName);
    }

    public void insertBinding(Binding binding) {
        metaMapper.insertBinding(binding);
    }

    public List<Binding> selectAllBindings() {
        return metaMapper.selectAllBindings();
    }

    public void deleteBinding(Binding binding) {
        metaMapper.deleteBinding(binding);
    }
}

测试 DataBaseManager

package com.example.mq;

import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.ExchangeType;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.datacenter.DataBaseManager;
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.util.List;

// 加上这个注解之后, 改类就会被识别为单元测试类.
@SpringBootTest
public class DataBaseManagerTests {
    private DataBaseManager dataBaseManager = new DataBaseManager();

    // 接下来下面这里需要编写多个 方法 . 每个方法都是一个/一组单元测试用例.
    // 还需要做一个准备工作. 需要写两个方法, 分别用于进行 "准备工作" 和 "收尾工作"

    // 使用这个方法, 来执行准备工作. 每个用例执行前, 都要调用这个方法.
    @BeforeEach
    public void setUp() {
        // 由于在 init 中, 需要通过 context 对象拿到 metaMapper 实例的.
        // 所以就需要先把 context 对象给搞出来.
        MqApplication.context = SpringApplication.run(MqApplication.class);
        dataBaseManager.init();
    }

    // 使用这个方法, 来执行收尾工作. 每个用例执行后, 都要调用这个方法.
    @AfterEach
    public void tearDown() {
        // 这里要进行的操作, 就是把数据库给清空~~ (把数据库文件, meta.db 直接删了就行了)
        // 注意, 此处不能直接就删除, 而需要先关闭上述 context 对象!!
        // 此处的 context 对象, 持有了 MetaMapper 的实例, MetaMapper 实例又打开了 meta.db 数据库文件.
        // 如果 meta.db 被别人打开了, 此时的删除文件操作是不会成功的 (Windows 系统的限制, Linux 则没这个问题).
        // 另一方面, 获取 context 操作, 会占用 8080 端口. 此处的 close 也是释放 8080.
        MqApplication.context.close();
        dataBaseManager.deleteDB();
    }

    @Test
    public void testInitTable() {
        // 由于 init 方法, 已经在上面 setUp 中调用过了. 直接在测试用例代码中, 检查当前的数据库状态即可.
        // 直接从数据库中查询. 看数据是否符合预期.
        // 查交换机表, 里面应该有一个数据(匿名的 exchange); 查队列表, 没有数据; 查绑定表, 没有数据.
        List<Exchange> exchangeList = dataBaseManager.selectAllExchanges();
        List<MSGQueue> queueList = dataBaseManager.selectAllQueues();
        List<Binding> bindingList = dataBaseManager.selectAllBindings();

        // 直接打印结果, 通过肉眼来检查结果, 固然也可以. 但是不优雅, 不方便.
        // 更好的办法是使用断言.
        // System.out.println(exchangeList.size());
        // assertEquals 判定结果是不是相等.
        // 注意这俩参数的顺序. 虽然比较相等, 谁在前谁在后, 无所谓.
        // 但是 assertEquals 的形参, 第一个形参叫做 expected (预期的), 第二个形参叫做 actual (实际的)
        Assertions.assertEquals(1, exchangeList.size());
        Assertions.assertEquals("", exchangeList.get(0).getName());
        Assertions.assertEquals(ExchangeType.DIRECT, exchangeList.get(0).getType());
        Assertions.assertEquals(0, queueList.size());
        Assertions.assertEquals(0, bindingList.size());
    }

    private Exchange createTestExchange(String exchangeName) {
        Exchange exchange = new Exchange();
        exchange.setName(exchangeName);
        exchange.setType(ExchangeType.FANOUT);
        exchange.setAutoDelete(false);
        exchange.setDurable(true);
        exchange.setArguments("aaa", 1);
        exchange.setArguments("bbb", 2);
        return exchange;
    }

    @Test
    public void testInsertExchange() {
        // 构造一个 Exchange 对象, 插入到数据库中. 再查询出来, 看结果是否符合预期.
        Exchange exchange = createTestExchange("testExchange");
        dataBaseManager.insertExchange(exchange);
        // 插入完毕之后, 查询结果
        List<Exchange> exchangeList = dataBaseManager.selectAllExchanges();
        Assertions.assertEquals(2, exchangeList.size());
        Exchange newExchange = exchangeList.get(1);
        Assertions.assertEquals("testExchange", newExchange.getName());
        Assertions.assertEquals(ExchangeType.FANOUT, newExchange.getType());
        Assertions.assertEquals(false, newExchange.isAutoDelete());
        Assertions.assertEquals(true, newExchange.isDurable());
        Assertions.assertEquals(1, newExchange.getArguments("aaa"));
        Assertions.assertEquals(2, newExchange.getArguments("bbb"));
    }

    @Test
    public void testDeleteExchange() {
        // 先构造一个交换机, 插入数据库; 然后再按照名字删除即可!
        Exchange exchange = createTestExchange("testExchange");
        dataBaseManager.insertExchange(exchange);
        List<Exchange> exchangeList = dataBaseManager.selectAllExchanges();
        Assertions.assertEquals(2, exchangeList.size());
        Assertions.assertEquals("testExchange", exchangeList.get(1).getName());

        // 进行删除操作
        dataBaseManager.deleteExchange("testExchange");
        // 再次查询
        exchangeList = dataBaseManager.selectAllExchanges();
        Assertions.assertEquals(1, exchangeList.size());
        Assertions.assertEquals("", exchangeList.get(0).getName());
    }

    private MSGQueue createTestQueue(String queueName) {
        MSGQueue queue = new MSGQueue();
        queue.setName(queueName);
        queue.setDurable(true);
        queue.setAutoDelete(false);
        queue.setExclusive(false);
        queue.setArguments("aaa", 1);
        queue.setArguments("bbb", 2);
        return queue;
    }

    @Test
    public void testInsertQueue() {
        MSGQueue queue = createTestQueue("testQueue");
        dataBaseManager.insertQueue(queue);

        List<MSGQueue> queueList = dataBaseManager.selectAllQueues();

        Assertions.assertEquals(1, queueList.size());
        MSGQueue newQueue = queueList.get(0);
        Assertions.assertEquals("testQueue", newQueue.getName());
        Assertions.assertEquals(true, newQueue.isDurable());
        Assertions.assertEquals(false, newQueue.isAutoDelete());
        Assertions.assertEquals(false, newQueue.isExclusive());
        Assertions.assertEquals(1, newQueue.getArguments("aaa"));
        Assertions.assertEquals(2, newQueue.getArguments("bbb"));
    }

    @Test
    public void testDeleteQueue() {
        MSGQueue queue = createTestQueue("testQueue");
        dataBaseManager.insertQueue(queue);
        List<MSGQueue> queueList = dataBaseManager.selectAllQueues();
        Assertions.assertEquals(1, queueList.size());
        // 进行删除
        dataBaseManager.deleteQueue("testQueue");
        queueList = dataBaseManager.selectAllQueues();
        Assertions.assertEquals(0, queueList.size());
    }

    private Binding createTestBinding(String exchangeName, String queueName) {
        Binding binding = new Binding();
        binding.setExchangeName(exchangeName);
        binding.setQueueName(queueName);
        binding.setBindingKey("testBindingKey");
        return binding;
    }

    @Test
    public void testInsertBinding() {
        Binding binding = createTestBinding("testExchange", "testQueue");
        dataBaseManager.insertBinding(binding);

        List<Binding> bindingList = dataBaseManager.selectAllBindings();
        Assertions.assertEquals(1, bindingList.size());
        Assertions.assertEquals("testExchange", bindingList.get(0).getExchangeName());
        Assertions.assertEquals("testQueue", bindingList.get(0).getQueueName());
        Assertions.assertEquals("testBindingKey", bindingList.get(0).getBindingKey());
    }

    @Test
    public void testDeleteBinding() {
        Binding binding = createTestBinding("testExchange", "testQueue");
        dataBaseManager.insertBinding(binding);
        List<Binding> bindingList = dataBaseManager.selectAllBindings();
        Assertions.assertEquals(1, bindingList.size());

        // 删除
        Binding toDeleteBinding = createTestBinding("testExchange", "testQueue");
        dataBaseManager.deleteBinding(toDeleteBinding);
        bindingList = dataBaseManager.selectAllBindings();
        Assertions.assertEquals(0, bindingList.size());
    }
}