四. 项⽬创建

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

五. 创建核⼼类 ★

创建包 mqserver.core

创建 Exchange（名字、类型、持久化、自动删除、参数）

/** 这个类表示一个交换机*/
public class Exchange {// 此处使用 name 来作为交换机的身份标识. (唯一的)private String name;// 交换机类型, DIRECT, FANOUT, TOPICprivate 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 字符串就 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();}}// 在这里针对 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;}@Overridepublic 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 只是把数据单纯的存储到⼀个⽂件中. ⾮常简单⽅便.

实现创建表【Mysql是手动创建表，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

/** 通过这个类, 来整合上述的数据库操作.*/
public class DataBaseManager {// 要做的是从 Spring 中拿到现成的对象private MetaMapper metaMapper;// 针对数据库进行初始化public void init() {// 手动的获取到 MetaMappermetaMapper = 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

// 加上这个注解之后, 改类就会被识别为单元测试类.
@SpringBootTest
public class DataBaseManagerTests {private DataBaseManager dataBaseManager = new DataBaseManager();// 接下来下面这里需要编写多个 方法 . 每个方法都是一个/一组单元测试用例.// 还需要做一个准备工作. 需要写两个方法, 分别用于进行 "准备工作" 和 "收尾工作"// 使用这个方法, 来执行准备工作. 每个用例执行前, 都要调用这个方法.@BeforeEachpublic void setUp() {// 由于在 init 中, 需要通过 context 对象拿到 metaMapper 实例的.// 所以就需要先把 context 对象给搞出来.MqApplication.context = SpringApplication.run(MqApplication.class);dataBaseManager.init();}// 使用这个方法, 来执行收尾工作. 每个用例执行后, 都要调用这个方法.@AfterEachpublic void tearDown() {// 这里要进行的操作, 就是把数据库给清空~~ (把数据库文件, meta.db 直接删了就行了)// 注意, 此处不能直接就删除, 而需要先关闭上述 context 对象!!// 此处的 context 对象, 持有了 MetaMapper 的实例, MetaMapper 实例又打开了 meta.db 数据库文件.// 如果 meta.db 被别人打开了, 此时的删除文件操作是不会成功的 (Windows 系统的限制, Linux 则没这个问题).// 另一方面, 获取 context 操作, 会占用 8080 端口. 此处的 close 也是释放 8080.MqApplication.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.isExclusive());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());}
}