2024.2.6 模拟实现 RabbitMQ —

引言

选择数据库

环境配置

设计数据库表

实现流程

封装数据库操作

针对 DataBaseManager 单元测试

引言

硬盘保存分为两个部分

数据库：交换机（Exchange）、队列（Queue）、绑定（Binding）
文件：消息（Message）

选择数据库

MySQL 数据库是比较重量的数据库！
此处为了使用更方便，简化环境，采取的数据库是更轻量的 SQLite 数据库

原因：

一个完整的 SQLite 数据库，只有一个单独的可执行文件（不到 1M）
MySQL 是客户端服务器结构的程序，而 SQLite 只是一个本地的数据库，相当于是直接操作本地的硬盘文件

注意：

SQLite 数据库应用非常广泛，在一些性能不高的设备上，SQLite 数据库是首选
尤其是移动端和嵌入式设备（Android 系统就是内置的 SQLite）

环境配置

在 Java 中要想使用 SQLite 数据库，无需额外安装，直接使用 Maven，将 SQLite 的依赖直接引入进来即可！
此时 Maven 依赖会自动加载 jar 包和动态库文件

<dependency><groupId>org.xerial</groupId><artifactId>sqlite-jdbc</artifactId><version>3.41.0.1</version>
</dependency>
编写 yml 配置文件（此处我们使用 MyBatis 操作 SQLite 数据库）
spring:datasource:url: jdbc:sqlite:./data/meta.dbusername:password:driver-class-name: org.sqlite.JDBCmybatis:mapper-locations: classpath:mapper/**Mapper.xml
注意点一：

SQLite 数据库将数据存储在当前硬盘的某个指定的文件中（./data/meta.db）

注意点二：

谈到相对路径，就需要明确 "基准路径" "工作路径"
如果是在 IDEA 中直接运行程序，此时工作路径就是当前项目所在的路径
如果是通过 javr -jar 方式运行部署的，此时你在哪个目录下执行的命令，哪个目录就是工作路径

注意点三：

对于 SQLite 数据库来说，并不需要指定用户名密码
MySQL 数据库是一个客户端服务器结构的程序，而一个数据库服务器，就会对应很多个客户端来访问它
相比之下，SQLite 则不是客户端服务器结构的程序，其数据放在本地文件上，与网络无关，只有本地主机才能访问

注意点四：

SQLite 虽然和 MySQL 不太一样，但是都可以通过 MyBatis 这样的框架来使用

注意点五：

当把上述的配置和依赖都准备好了之后，程序启动便会自动建库！

设计数据库表

需要在数据库中存储的有交换机（Exchange）、队列（Queue）、绑定（Binding）
对照着上述这样的核心类，很容易把这几个表设计出来的

问题：

上述表的建表操作，具体什么时机来执行？

回答：

以往写的程序，都是先将数据库表啥的创建好，再启动服务器
即将建库建表语句写到一个 .sql 文件中，需要建表时，直接复制到 MySQL 客户端中执行即可
这个操作都是在部署阶段完成的
之前大概就部署一次即可，不会反复操作，但是后续接触到的更多的程序可能会涉及到反复部署多次
综上，通过代码自动完成建表操作，简化部署步骤，也是十分关键的！

实现流程

创建一个 interface 接口，描述有哪些方法要给 Java 代码使用
import com.example.demo.mqserver.core.Binding;
import com.example.demo.mqserver.core.Exchange;
import com.example.demo.mqserver.core.MSGQueue;
import org.apache.ibatis.annotations.Mapper;import java.util.List;@Mapper
public interface MetaMapper {
//    提供三个核心建表方法void createExchangeTable();void createQueueTable();void createBindingTable();//    针对上述三个基本概念，进行 插入 和 删除void insertExchange(Exchange exchange);List<Exchange> selectAllExchanges();void deleteExchange(String exchangeName);void insertQueue(MSGQueue msgQueue);List<MSGQueue> selectAllQueues();void deleteQueue(String queueName);void insertBinding(Binding binding);List<Binding> selectAllBindings();void deleteBinding(Binding binding);
}
创建对应的 xml 文件，通过 xml 来实现上述 interface 接口中的抽象方法
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.example.demo.mqserver.mapper.MetaMapper"><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><insert id="insertExchange" parameterType="com.example.demo.mqserver.core.Exchange">insert into exchange values(#{name}, #{type}, #{durable}, #{autoDelete}, #{arguments});</insert><select id="selectAllExchanges" resultType="com.example.demo.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.demo.mqserver.core.MSGQueue">insert into queue values(#{name}, #{durable}, #{exclusive}, #{autoDelete}, #{arguments});</insert><select id="selectAllQueues" resultType="com.example.demo.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.demo.mqserver.core.Binding">insert into binding values(#{exchangeName}, #{queueName}, #{bindingKey});</insert><select id="selectAllBindings" resultType="com.example.demo.mqserver.core.Binding">select * from binding;</select><delete id="deleteBinding" parameterType="com.example.demo.mqserver.core.Binding">delete from binding where exchangeName = #{exchangeName} and queueName = #{queueName};</delete>
</mapper>
注意点一：

此处我们使用 update 标签来实现建表操作！

问题：

当前是将每个建表语句都单独的列为一个 update 标签，并且对应一个 Java 方法
此处我们能否改成一个 update 标签中包含多个建表语句，同时借助一个 Java 方法，完成上述多个表的创建呢？

回答：

MyBatis 支持一个标签中包含多个 sql 语句，其前提为，搭配 MySQL 或 Oracle 使用
对于 SQLite 来说，是无法做到上述功能的
即当一个 update 标签中，写了多个 create table 语句时，只有第一个语句能执行

注意点二：

Exchange 和 Queue 这两个表，由于使用 name 做为主键，直接按照 name 进行删除即可
但对于 Binding 来说，此时没有主键，其删除操作是针对 exchangeName 和 queueName 两个纬度进行筛选

问题：

如果实现把 argument 键值对与数据库中的字符串类型相互转换呢？

回答：

关键要点在于，MyBatis 在完成数据库操作时，会自动的调用到对象的 getter 和 setter
比如 MyBatis 往数据库中写数据时，就会调用对象的 getter 方法，拿到属性值再往数据库中写
如果这个过程中，让 getArgument 得到的结果为 String 类型，此时，就可以直接把这个数据写到数据库了

比如 MyBatis 从数据库读数据时，就会调用对象的 setter 方法，将数据库中读到的结果设置到对象的属性中
如果这个过程中，让 setArgument 的参数为 String 类型，并且在 setArguments 内部针对字符串解析，解析成一个 Map 对象
综上，我们直接在包含 argument 成员变量的 Exchange 实体类和 MSGQueue 实体类中，改写 argument 的 getter 和 setter 方法即可
//    这里的 getter setter 用于和数据库进行交互public String getArguments() {
//        是把当前的 arguments 参数，从 Map 转成 String (JSON)ObjectMapper objectMapper = new ObjectMapper();try {return objectMapper.writeValueAsString(arguments);} catch (JsonProcessingException e) {e.printStackTrace();}
//        如果代码真异常了，返回一个空的 json 字符串就 okreturn "{}";}//    这个方法，是从数据库读数据之后，构造 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();}}
注意：

此处的第二个参数用来描述当前 JSON 字符串，要转成的 Java 对象是啥类型的
如果是个简单类型，直接使用对应类型的类对象即可
如果是集合类这样的复杂类型，可以使用 TypeReference 匿名内部类对象，来描述复杂类型的具体信息（通过泛型参数来描述的）

封装数据库操作

此处我们将专门写一个类来整合上述的数据库操作

import com.example.demo.DemoApplication;
import com.example.demo.mqserver.core.Binding;
import com.example.demo.mqserver.core.Exchange;
import com.example.demo.mqserver.core.ExchangeType;
import com.example.demo.mqserver.core.MSGQueue;
import com.example.demo.mqserver.mapper.MetaMapper;import java.io.File;
import java.util.List;/*
* 通过这个类，来整合上述的数据库操作
* */
public class DataBaseManager {
//    从 Spring 中拿到现成的对象private MetaMapper metaMapper;//    针对数据库进行初始化public void init() {
//        手动的获取到 MetaMappermetaMapper = DemoApplication.context.getBean(MetaMapper.class);if (!checkDBExists()) {
//            如果数据库不存在，就进行建库建表操作
//            先创建一个 data 目录File dataDir = new File("./data");dataDir.mkdir();
//            创建数据库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 里有一个这样的设定: 带有一个 匿名 的交换机,类型是 DIRECTprivate 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);}
}

注意点一：

谈到初始化，我们一般都会用到构造方法
但是此处的 init() 为一个普通方法
构造方法一般是用来初始化类的属性，即一般不太会涉及到太多的业务逻辑
但是此处的初始化是带有业务逻辑的，还是单独拎出来，手动来调用比较合适一些

注意点二：

数据库初始化 = 建库建表 + 插入一些默认数据
此处我们期望在 broker server 启动时，做出下列逻辑判定：

如果数据库已经存在了，即表啥的都有了，则不做任何操作
如果数据库不存在，则创建库，创建表，构造默认数据

实例理解

例如，现在将 broker server 部署到一个新的服务器上
显然，此时是没有数据库的，需让 broker server 启动时，自动将对应的数据库创建好
但是如果是一个已经部署过的机器，当 broker server 重启时，就会发现数据库已经有了，此时将不做任何数据库相关操作
综上，判定数据库是否存在，就等同于判定 meta.db 这个文件是否存在即可！

注意点三：

当然手动获取 MetaMapper 的对象前提是改写启动类代码

针对 DataBaseManager 单元测试

在设计单元测试时，要求单元测试用例与用例之间需相互独立，互不干扰

实例理解

测试用例A => 测试过程中，给数据库里插入了一些数据
测试用例B => 再针对 B 进行测试，可能 A 这里的数据，就会对 B 造成干扰
测试用例C => 再针对 C 测试，A 和 B 都可能因为数据库的数据影响到 C

注意：

此处的影响不一定是数据库，也可能是其他方面，如是否搞了个文件，是否占用了端口等

解决方案：

每个用例执行之前，先执行一段逻辑，搭建测试环境，准备好测试需要用到的一些东西
每个用例执行之后，再执行一段逻辑，把用例执行过程中产生的中间结果，一些影响，给消除掉
综上，我们可以使用 @BeforeEach 和 @AfterEach 这两个注解

import com.example.demo.mqserver.core.Binding;
import com.example.demo.mqserver.core.Exchange;
import com.example.demo.mqserver.core.ExchangeType;
import com.example.demo.mqserver.core.MSGQueue;
import com.example.demo.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();//    接下来下面这里需要编写多个 方法,每个方法都是一个/一组单元测试用例
//    还需要做一个准备工作,需要写两个方法,分别用于 "准备工作" 和 "收尾工作"//    使用这个方法,来执行准备工作,每个用例执行前,都要调用这个方法\@BeforeEachpublic void setUp() {
//        由于在 init 中,需要通过 context 对象拿到 metaMapper 实例的
//        所以就需要先把 context 对象给搞出来DemoApplication.context = SpringApplication.run(DemoApplication.class);dataBaseManager.init();}//    使用这个方法,来执行收尾工作,每个用例执行后,都要调用这个方法@AfterEachpublic void tearDown() {
//        这里要进行的操作,就是把数据库给清空(把数据库文件,meta.db 直接删了就行)
//        注意,此处不能直接就删除,而需要先关闭上述 context 对象!
//        此处的 context 对象,持有了 Meta 的实例, MetaMapper 实例又打开了 meta.db 数据库文件
//        如果 Meta.db 被别人打开了,此时的删除文件操作是不会成功的(Windows 系统的限制,Linux 则没有这个问题)
//        另一方面,获取 context 操作,会占用 8080 端口,此处的 close 也是释放 8080DemoApplication.context.close();dataBaseManager.deleteDB();}@Testpublic 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;}@Testpublic 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"));}@Testpublic 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;}@Testpublic 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.isAutoDelete());Assertions.assertEquals(1,newQueue.getArguments("aaa"));Assertions.assertEquals(2,newQueue.getArguments("bbb"));}@Testpublic 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;}@Testpublic 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());}@Testpublic 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());}
}

注意点一：