做网站后端要什么技术,网站如何做百度权重,小米商城官方网站入口,分析企业网站建设流程目录 一、消息存储格式设计 #x1f345; 1、queue_data.txt#xff1a;保存消息的内容 #x1f345; 2、queue_stat.txt#xff1a;保存消息的统计信息
二、消息序列化
三、自定义异常类
四、创建MessageFileManger类
#x1f345; 1、约定消息文件所在的目录和文件名…目录 一、消息存储格式设计 1、queue_data.txt保存消息的内容 2、queue_stat.txt保存消息的统计信息
二、消息序列化
三、自定义异常类
四、创建MessageFileManger类 1、约定消息文件所在的目录和文件名字 2、队列的统计信息 3、创建队列对应的目录和功能 4、实现删除队列的文件和目录 5、检查队列的目录和文件是否都存在 6、把消息写入到文件中 7、删除文件中的消息 8、将硬盘中的数据加载到内存中 9、实现消息文件的垃圾回收 检测是否要进行GC 构造新目录 进行GC操作
五、测试MessageFileManager类 1、“准备工作”和“收尾工作” 2、测试创建文件是否存在 3、测试writetStat和readStat是否能够通过 4、测试sendMessage 5、测试删除消息 6、测试垃圾回收
六、小结 一、消息存储格式设计 对于消息并不打算存储在数据库中 1消息操作并不会涉及到复杂的增删改查 2消息的数量可能会非常多数据库的访问效率并不高 所以我们直接把消息存储在文件中。
那么消息要如何在文件中存储呢
首先消息它是依附于队列的所以在存储的时候就把消息按照队列的维度展开。
我们会将队列存储在和数据库同级的data目录中在data中创建一些子目录子目录的名字就是队列名。 然后在每个队列的子目录下面再分配两个文件用来存储消息。主要是以下两个文件。 1、queue_data.txt保存消息的内容 Message是一个二进制格式的文件包含若干个消息每个消息都以二进制的方式存储每个消息都由这几个部分构成Message对象序列化之后。 Messag对象会分别再内存和硬盘上都记录一份。内存中的衣服呢会记录offsetBegin和offsetEnd。这样就可以找到内存中的Message对象能够找到对应的硬盘上的message对象。 关于isValid是用来标识当前消息在文件中是否有效为1就是有效的消息为0就是无效的消息。当为0时就相当于逻辑上的删除消息功能。但是随着时间的推移消息的增多那么该消息可能就会大部分都是无效的消息针对这种情况就需要对当前消息的文件进行垃圾回收。 以下是本程序中实现的垃圾回收功能 垃圾回收GC使用复制算法针对消息数据文件中的垃圾回收进行回收。直接遍历原有的消息数据文件把有效的数据拷贝到一个新的文件中然后把之前的旧文件都删除掉逻辑删除。 作出约定可以不按这个来当总消息数目超过了2000并且有效的消息数目低于总数目的50%就出发一次垃圾回收。 对于RabbitMQ解决垃圾回收的方式如下 如果某个消息队列中消息很多都是有效消息就会导致整个消息的数据文件特别答后续针对文件的各种操作成本就会很高。RabbitMQ解决方案如下 文件拆分当单个文件长度达到一定阈值以后就会拆分成两个文件拆分次数越多文件就越多 文件合并每个单独的文件都会进行GC如果GC之后就会发现文件变小很多当小到一定程度就会和其他文件合并。 这样就会再消息特别多的时候也能保证性能上的及时响应。 2、queue_stat.txt保存消息的统计信息 使用这个文件来保存消息的统计信息。 只存一行数据文本格式。然后一行包括两列两者使用 \t 来分割形如2000\t500 第一列是queue_data.txt中总的消息的数目 第二列是queue_data.txt中有效消息的数目。 二、消息序列化
消息序列化把一个对象结构化数据转成一个字符串/字节数组。
在序列化之后对象的信息是不会丢失的这样就会方便与存储和传输在文件中存储时只能以字符串/二进制数据的方式存储对象。后面需要用的时候就再反序列化。
由于消息的body是二进制数据所以这里不会使用JSON进行序列化。 针对二进制序列化有很多解决方案 1Java标准库中提供了序列化的方案ObjectInputStream 和 ObjectOutputStream 2Hession 3protobuffer 4thrift 这里使用第一种这样就不用再引入额外的依赖。
在commen中创建一个BinaryTool因为后面客户端还会用到这个类所以放在公共的包中。 /*
* 序列化和反序列化
* 实现Serializable接口才能让这个对象进行序列化和反序列化
* */
public class BinaryTool {// 把一个对象序列化成一个字节数组public static byte[] toBytes(Object object) throws IOException {
// 这个流对象相当于一个边长的字节数组
// 就可以把object序列化的数组逐渐写入到byteArrayOutputStream中然后统一转成byte[]try (ByteArrayOutputStream byteArrayOutputStream new ByteArrayOutputStream()) {try (ObjectOutputStream objectOutputStream new ObjectOutputStream(byteArrayOutputStream)) {
// 此处的writeObject就会把该对象进行序列化生成二进制字节数据就会写入到ObjectOutputStream中
// 由于ObjectOutputStream又关联到了ByteArrayOutputStream最终结果就写入到ByteArrayOutputStream里面了objectOutputStream.writeObject(object);}
// 该操作就是把byteArrayOutputStream中持有的二进制数据取出来转成byte[]return byteArrayOutputStream.toByteArray();}}// 把一个字节数组反序列化成一个对象public static Object fromBytes(byte[] data) throws IOException, ClassNotFoundException {Object object null;try (ByteArrayInputStream byteArrayInputStream new ByteArrayInputStream(data)) {try (ObjectInputStream objectInputStream new ObjectInputStream(byteArrayInputStream)) {
// 此处的readObject,就是从data的byte[]中读取数据并且进行反序列化object objectInputStream.readObject();}}return object;}
} 三、自定义异常类 自定义一个异常类如果是mq的业务逻辑中出的异常就抛出这个异常类
/*
*自定义异常类
*/
public class MqException extends Exception{public MqException(String reason){super(reason);}
} 四、创建MessageFileManger类 对硬盘上的消息进行管理的类。 1、约定消息文件所在的目录和文件名字
//这里的init()方法暂时不用只是列在这后面可能扩展
public void init(){
// 后续扩展}// 约定的那个消息文件所在的目录和文件名// 用来获取到指定队列对应的消息文件所在的路径private String getQueueDir(String queueName){return ./data/ queueName;}// 用来获取该队列的消息数据文件路径private String getQueueDataPath(String queueName){return getQueueDir(queueName) /queue_data.txt;}// 用来获取该队列列的消息统计文件路径private String getQueueStatPath(String queueName){return getQueueDir(queueName) /queue_stat.txt;} 2、队列的统计信息
定义一个内部类来表示该队列的统计信息
// 定义一个内部类来表示该队列的统计信息static public class Stat{
// 直接定义成publicpublic int totalCount; //总消息数量public int validCount;//有效消息的数量}
然后实现消息统计的读写功能
private Stat readStat(String queueName){
// 由于当前的消息统计文件是文本文件直接使用scanner读取文件内容Stat stat new Stat();try (InputStream inputStream new FileInputStream(getQueueStatPath(queueName))) {Scanner scanner new Scanner(inputStream);stat.totalCount scanner.nextInt();stat.validCount scanner.nextInt();return stat;} catch (IOException e) {e.printStackTrace();}return null;}private void writeStat(String queueName,Stat stat){
// 使用PrinWrite来写
// OutputStream打开文件默认情况下会直接把源文件清空新数据会覆盖原数据
// 这里直接覆盖就可以了try (OutputStream outputStream new FileOutputStream(getQueueStatPath(queueName))) {PrintWriter printWriter new PrintWriter(outputStream);printWriter.write(stat.totalCount \t stat.validCount);printWriter.flush();} catch (IOException e) {e.printStackTrace();}} 3、创建队列对应的目录和功能
// 创建队列对应的文件和目录public void createQueueFiles(String queueName) throws IOException {
// 1.创建队列对应的消息目录File baseDir new File(getQueueDir(queueName));if (!baseDir.exists()){
// 不存在就创建这个目录boolean ok baseDir.mkdirs();if (!ok){throw new IOException(创建目录失败baseDir baseDir.getAbsolutePath());}}// 2.创建队列数据文件File queueDataFile new File(getQueueDataPath(queueName));if (!queueDataFile.exists()){boolean ok queueDataFile.createNewFile();if (!ok){throw new IOException(创建文件失败queueDataFile queueDataFile.getAbsolutePath());}}
// 3.创建消息统计文件File queueStatFile new File(getQueueStatPath(queueName));if (!queueStatFile.exists()){boolean ok queueStatFile.createNewFile();if (!ok){throw new IOException(创建文件失败queueStatFile queueStatFile.getAbsolutePath());}}// 4.给消息统计文件设定初始值 0\t0Stat stat new Stat();stat.totalCount 0;stat.validCount 0;writeStat(queueName,stat);} 4、实现删除队列的文件和目录
// 实现删除队列的文件和目录public void destroyQueueFiles(String queueName) throws IOException {
// 先删除文件在删除目录File queueDataFile new File(getQueueDataPath(queueName));boolean ok1 queueDataFile.delete();File queueStatFile new File(getQueueStatPath(queueName));boolean ok2 queueStatFile.delete();File baseDir new File(getQueueDir(queueName));boolean ok3 baseDir.delete();if (!ok1 || !ok2 || !ok3){
// 有任意一个删除失败就算整体删除失败throw new IOException(删除队列目录和文件失败baseDir baseDir.getAbsolutePath());}}5、检查队列的目录和文件是否都存在
后续有生产者给brocker server生产消息了这个消息就可能需要记录到文件上取决消息是否要持久化。但是判断是否要持久化之前需要检查队列中的文件是否存在。
// 检查队列的目录和文件是否存在public boolean checkFileExists(String queueName){
// 判断队列的数据文件和统计文件是否都存在File queueDataFile new File(getQueueDataPath(queueName));if (!queueDataFile.exists()){return false;}File queueStatFile new File(getQueueStatPath(queueName));if (!queueStatFile.exists()){return false;}return true;} 6、把消息写入到文件中 把一个新的消息放到队列对应的文件中。主要包含以下三步 1检查写入的队列是否存在 2把Message对象进行序列化转成二进制的字节数组 3获取到当前数据的文件长度使用[offsetBegin,offsetEnd]。把新的Message数据写入到队列数据文件的末尾此时 Message对象的offsetBegin就是当前文件长度 4 offsetEnd就是当前文件长度 4 message自身长度 4写入消息到数据文件追加到数据文件末尾 5更新消息统计文件 写入文件时的线程安全问题 * 如果两个线程是往同一个队列中写消息此时就需要阻塞等待 假设现在有两个线程t1,t2。如果没有加锁那么他们的目的就是将一个message写入到104~124之间去。但是此时可能就会导致t1计算长度以后没有进行写文件t2就开始计算长度了并且执行了写文件操作写完以后t1才开始写但是此时t1就不是从104写了而是从124开始写。这样会导致queue_data多出一段。 所以这里我们就需要对队列进行加锁。 * 如果两个线程需要往不同的队列中些消息此时就不需要阻塞等待。 总体代码如下
// 该方法用于把一个新的消息,放到队列对应的文件中
// queue表示要把消息写入的队列message则是要写的消息public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
// 1、检查要写入的队列是否存在
// 如果不存在if (!checkFileExists(queue.getName())){throw new MqException([MessageFileManager] 队列对应的文件不存在queueName queue.getName());}// 2、把Message对象进行序列化转成二进制的字节数组byte[] messageBinary BinaryTool.toBytes(message);// 这个锁是,当有两个对象针对同一个对象操作时,锁才会有效synchronized (queue){
// 3、先获取到当前的队列数据文件长度使用[offsetBegin,offsetEnd]
// 把新的Message数据写入到队列数据文件的末尾此时Message对象的offsetBegin,就是当前文件长度4
// offsetEnd就是当前文件长度 4 message自身长度File queueDataFile new File(getQueueDataPath(queue.getName()));
// 获取到文件的长度queueDataFile.length();单位字节message.setOffsetBegin(queueDataFile.length() 4);message.setOffsetEnd(queueDataFile.length() 4 messageBinary.length);// 4.写入消息到数据文件,追加到数据文件末尾try(OutputStream outputStream new FileOutputStream(queueDataFile,true)){try(DataOutputStream dataOutputStream new DataOutputStream(outputStream)){
// 先写当前消息的长度占据4个字节
// writeInt()方法用于将给定的整数值作为4个字节(即32位)写入基本DataOutputStream并且成功执行时变量计数器加4。dataOutputStream.writeInt(messageBinary.length);
// 写入消息本体dataOutputStream.write(messageBinary);}}// 5.更新消息统计文件Stat stat readStat(queue.getName());stat.totalCount 1;stat.validCount 1;writeStat(queue.getName(),stat);}} 7、删除文件中的消息 这里就是逻辑删除将isValid设置为0. 主要分为3步 1把文件中需要删除的一段数据读出来 2还原回Message对象反序列化 3把isValid改为0 4将上面的数据又写回到文件。 5更新统计文件 这里的message对象必须要包含offsetBegin和offsetEnd。因为这里是对文件中指定的位置进行读写的把这个随机访问。随机访问用到的类RandomAcessFile。 关于RandomAcessFile.seek()是用于设置文件指针相当于光标位置设置后光标会从当前指针的下一位读取到或写入到。 // 删除消息的方法public void deleteMessage(MSGQueue queue,Message message) throws IOException,ClassNotFoundException{synchronized (queue){try(RandomAccessFile randomAccessFile new RandomAccessFile(getQueueDataPath(queue.getName()),rw)){
// 1.先从文件中读取对应的Message数据byte[] bufferSrc new byte[(int)(message.getOffsetEnd() - message.getOffsetBegin())];randomAccessFile.seek(message.getOffsetBegin());randomAccessFile.read(bufferSrc);
// 2.把当前读出来的二进制数据,转回成Message对象Message diskMessage (Message) BinaryTool.fromBytes(bufferSrc);
// 3.把isValid设置为无效diskMessage.setIsValid((byte) 0x0);
// 4.重新写入文件byte[] buffserDest BinaryTool.toBytes(diskMessage);
// 这里还需要设置光标的位置,因为,上面的光标已经随着读出数据而发生了改变,已经走到了下一条message的offsetBegin,
// 这里为了重新写入数据到文件中,就需要将光标移到对应的位置上面randomAccessFile.seek(message.getOffsetBegin());randomAccessFile.write(buffserDest);
// 5.统计文件-1
// 因为有一条数据无效了Stat stat readStat(queue.getName());if (stat.validCount 0){stat.validCount - 1;}writeStat(queue.getName(),stat);}}} 8、将硬盘中的数据加载到内存中 将数据从文件中,读取出所有的消息内容,加载到内存当中(放到一个链表中)这个方法会在程序启动的时候调用主要又以下几步 1.读取当前消息的长度 2.按照该长度读取消息内容 3.将读取到的二进制数据反序列化回Message对象 4.判断消息对象是不是无效对象 4.将有效的Message对象插入到链表中 // 将数据从文件中,读取出所有的消息内容,加载到内存当中(放到一个链表中)
// 由于该方法实在程序启动时调用, 此时服务器还不能处理请求,不涉及线程操作文件.public LinkedListMessage loadAllMessageFromQueue(String queueName) throws IOException,MqException,ClassNotFoundException {LinkedListMessage messages new LinkedList();try(InputStream inputStream new FileInputStream(getQueueDataPath(queueName))){try(DataInputStream dataInputStream new DataInputStream(inputStream)){
// 该变量记录当前文件光标位置,初始位置为0long currentOffset 0;
// 一个文件中包含了很多消息,这里要循环读取while (true){
// 1.读取当前消息的长度,这里可能会读到文件末尾
// reaIn()方法读到文件末尾会抛出EOFException异常
// readInt()读取出4个字节int messageSize dataInputStream.readInt();
// 2.按照这个长度,读取消息内容
// buffer是一个盛放消息容器,和消息的长度一般大小byte[] buffer new byte[messageSize];int actualSize dataInputStream.read(buffer);if (messageSize ! actualSize){
// 如果不匹配说明文件有问题格式错乱了throw new MqException([MessageFileManager] 文件格式错误queueName queueName);}
// 3.将读取到的二进制数据反序列化回Message对象Message message (Message) BinaryTool.fromBytes(buffer);// 4.判定这个消息对象是不是无效对象if(message.getIsValid() ! 0x1){
// 无效数据直接跳过
// 虽然消息是无效消息但是offset要更新currentOffset (4 messageSize);continue;}
// 有效数据则需要把这个Message对象加入到链表中加入前还需要填写offsetBegin和offsetEnd;
// 进行offset的时候需要知道当前光标的位置由于当下使用的DataInputStream并不方便计算光标位置
// 因此这里手动计算文件光标位置message.setOffsetBegin(currentOffset 4);message.setOffsetEnd(currentOffset 4 messageSize);currentOffset (4 messageSize);messages.add(message);}} catch (EOFException e){System.out.println([MessageFileManager]恢复Message数据完成);}}return messages;} 9、实现消息文件的垃圾回收 为什么要实现垃圾回收GC 由于当前会不停的往消息文件中写入新消息而且删除也只是逻辑删除isValid这样就可能导致消息文件越来越大并且里面又包含了大量的无效消息。 此处的垃圾回收使用的是复制算法 判断当文件中消息总数超过了2000并且有效消息的数目不足50%时就触发垃圾回收。然后将文件中有效的消息复制出来单独写入到一个新的文件中删除旧文件使用新文件代替。 检测是否要进行GC
检查当前是否要针对该队列的消息数据文件进行GC判断是否要GC根据总消息数目和有效消息数目判断。
// 检查当前是否要针对该队列的消息数据文件进行GCpublic boolean checkGC(String queueName){
// 判断是否要GC,根据总消息数目和有效消息数目这两个值都是在消息统计文件中实现的。Stat stat readStat(queueName);if (stat.totalCount 2000 (double)stat.validCount / (double) stat.totalCount 0.5){return true;}return false;} 构造新目录
构造一个新目录放置有效的复制信息。
// 构造一个目录结构放置复制的信息private String getQueueDataNewPath(String queueName){return getQueueDir(queueName) /queue_data_new.txt;}进行GC操作
执行消息数据文件的垃圾回收操作使用复制算法完成主要分为以下几步 1创建一个新的文件queue_data_new.txt 2从旧文件中读取出所有的有效消息对象 3把有效消息写入到queue_data_new.txt 4删出旧的文件并把新的文件重命名queue_data_new.txt queue_data.txt 5更新统计文件 public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
// 进行gc的时候是针对消息数据文件作出整体性的一个操作在这个过程中
// 进行加锁操作让其他线程不能对该队列的消息文件作出任何修改synchronized (queue) {// 由于 gc 操作可能比较耗时, 此处统计一下执行消耗的时间.long gcBegin System.currentTimeMillis();// 1. 创建一个新的文件File queueDataNewFile new File(getQueueDataNewPath(queue.getName()));if (queueDataNewFile.exists()) {// 正常情况下, 这个文件不应该存在. 如果存在, 就是意外~~ 说明上次 gc 了一半, 程序意外崩溃了.throw new MqException([MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在! queueName queue.getName());}boolean ok queueDataNewFile.createNewFile();if (!ok) {throw new MqException([MessageFileManager] 创建文件失败! queueDataNewFile queueDataNewFile.getAbsolutePath());}// 2. 从旧的文件中, 读取出所有的有效消息对象了. (这个逻辑直接调用上述方法即可, 不必重新写了)LinkedListMessage messages loadAllMessageFromQueue(queue.getName());// 3. 把有效消息, 写入到新的文件中.try (OutputStream outputStream new FileOutputStream(queueDataNewFile)) {try (DataOutputStream dataOutputStream new DataOutputStream(outputStream)) {for (Message message : messages) {byte[] buffer BinaryTool.toBytes(message);// 先写四个字节消息的长度dataOutputStream.writeInt(buffer.length);dataOutputStream.write(buffer);}}}// 4. 删除旧的数据文件, 并且把新的文件进行重命名File queueDataOldFile new File(getQueueDataPath(queue.getName()));ok queueDataOldFile.delete();if (!ok) {throw new MqException([MessageFileManager] 删除旧的数据文件失败! queueDataOldFile queueDataOldFile.getAbsolutePath());}// 把 queue_data_new.txt queue_data.txtok queueDataNewFile.renameTo(queueDataOldFile);if (!ok) {throw new MqException([MessageFileManager] 文件重命名失败! queueDataNewFile queueDataNewFile.getAbsolutePath() , queueDataOldFile queueDataOldFile.getAbsolutePath());}// 5. 更新统计文件Stat stat readStat(queue.getName());stat.totalCount messages.size();stat.validCount messages.size();writeStat(queue.getName(), stat);long gcEnd System.currentTimeMillis();System.out.println([MessageFileManager] gc 执行完毕! queueName queue.getName() , time (gcEnd - gcBegin) ms);}} 五、测试MessageFileManager类 1、“准备工作”和“收尾工作”
创建MessagefileManagerTests SpringBootTest
public class MessageFileManagerTests {private MessageFileManger messageFileManger new MessageFileManger();private static final String queueName1 testQueue1;private static final String queueName2 testQueue2;// 每个用例执行之前的准备工作BeforeEachpublic void setUp() throws IOException {
// 准备阶段创建出两个队列以备后用messageFileManger.createQueueFiles(queueName1);messageFileManger.createQueueFiles(queueName2);}// 每个用例执行之后的收尾工作AfterEachpublic void tearDown() throws IOException {
// 收尾阶段把创建出的队列销毁掉messageFileManger.destroyQueueFiles(queueName1);messageFileManger.destroyQueueFiles(queueName2);}
} 2、测试创建文件是否存在
Testpublic void testCreateFiles(){
// 创建队列文件在准备工作已经执行过了这里主要是为了验证文件是否存在File queueDataFile1 new File(./data/ queueName1 /queue_data.txt);
// assertEquals(预期值实际值)Assertions.assertEquals(true,queueDataFile1.isFile());File queueStatFile1 new File(./data/ queueName1 /queue_stat.txt);Assertions.assertEquals(true,queueStatFile1.isFile());File queueDataFile2 new File(./data/ queueName2 /queue_data.txt);Assertions.assertEquals(true,queueDataFile2.isFile());File queueStatFile2 new File(./data/ queueName2 /queue_stat.txt);Assertions.assertEquals(true,queueStatFile2.isFile());} 这里为了方便查看文件是否创建就把收尾工作注释掉了 3、测试writetStat和readStat是否能够通过
Testpublic void testReadWriteStat(){MessageFileManger.Stat stat new MessageFileManger.Stat();stat.totalCount 100;stat.validCount 50;// 由于writeStat和readStat是私有方法此处就需要使用反射的方式
// 使用Spring封装好的反射的工具类ReflectionTestUtils.invokeMethod(messageFileManger,writeStat, queueName1,stat);// 写入完毕之后调用读取验证读取的结果和写入的数据是一致的MessageFileManger.Stat newStat ReflectionTestUtils.invokeMethod(messageFileManger,readStat,queueName1);Assertions.assertEquals(100,newStat.totalCount);Assertions.assertEquals(50,newStat.validCount);System.out.println(writetStat和readStat测试通过);}4、测试sendMessage
构造创建queue和message的方法 private MSGQueue createTestQueue(String queueName){MSGQueue queue new MSGQueue();queue.setName(queueName);queue.setDurable(true); //是否要持久化return queue;}
// 构造出一条消息private Message createTestMessage(String content){Message message Message.createMessageWithId(testRoutingKey,null,content.getBytes());return message;}
测试sendMessage
Testpublic void testSendMessage() throws IOException, MqException, ClassNotFoundException {
// 构造出消息并且构造出队列Message message createTestMessage(testMessage);
// 创建queue对象MSGQueue queue createTestQueue(queueName1);// 调用发送消息的方法messageFileManger.sendMessage(queue,message);// 检查stat文件MessageFileManger.Stat stat ReflectionTestUtils.invokeMethod(messageFileManger,readStat,queueName1);Assertions.assertEquals(1,stat.totalCount);Assertions.assertEquals(1,stat.validCount);// 检查文件,把消息读出来LinkedListMessage messages messageFileManger.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(1,messages.size());Message curMessage messages.get(0);Assertions.assertEquals(message.getMessageId(), curMessage.getMessageId());Assertions.assertEquals(message.getDeliverMode(),curMessage.getDeliverMode());// 比较两个字节数组的内容是否相同不能直接使用asserEqualsAssertions.assertArrayEquals(message.getBody(),curMessage.getBody());System.out.println(message curMessage);} 构造100条消息 并且读取出来 Testpublic void testLoadAllMessageFromQueue() throws IOException, MqException, ClassNotFoundException {
// 往队列中插入100条消息验证100条消息从文件中读取之后是否和最初是一致的MSGQueue queue createTestQueue(queueName1);ListMessage expectedMessages new LinkedList();for (int i 0; i 100; i) {Message message createTestMessage(testMessge 1);messageFileManger.sendMessage(queue,message);expectedMessages.add(message);}// 读取所有消息LinkedListMessage actualMessages messageFileManger.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(expectedMessages.size(),actualMessages.size());for (int i 0; i expectedMessages.size(); i) {Message expectedMessage expectedMessages.get(i);Message actualMessage actualMessages.get(i);System.out.println([ i ]actualMessage actualMessages);Assertions.assertEquals(expectedMessage.getMessageId(),actualMessage.getMessageId());Assertions.assertEquals(expectedMessage.getRoutingKey(),actualMessage.getRoutingKey());Assertions.assertEquals(expectedMessage.getDeliverMode(),actualMessage.getDeliverMode());Assertions.assertArrayEquals(expectedMessage.getBody(),actualMessage.getBody());Assertions.assertEquals(0x1,actualMessage.getIsValid());}} 5、测试删除消息
// 测试删除消息Testpublic void testDeleteMessage() throws IOException, MqException, ClassNotFoundException {// 创建队列, 写入 10 个消息. 删除其中的几个消息. 再把所有消息读取出来, 判定是否符合预期.MSGQueue queue createTestQueue(queueName1);ListMessage expectedMessages new LinkedList();for (int i 0; i 10; i) {Message message createTestMessage(testMessage i);messageFileManager.sendMessage(queue, message);expectedMessages.add(message);}// 删除其中的三个消息messageFileManager.deleteMessage(queue, expectedMessages.get(7));messageFileManager.deleteMessage(queue, expectedMessages.get(8));messageFileManager.deleteMessage(queue, expectedMessages.get(9));// 对比这里的内容是否正确.LinkedListMessage actualMessages messageFileManager.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(7, actualMessages.size());for (int i 0; i actualMessages.size(); i) {Message expectedMessage expectedMessages.get(i);Message actualMessage actualMessages.get(i);System.out.println([ i ] actualMessage actualMessage);Assertions.assertEquals(expectedMessage.getMessageId(), actualMessage.getMessageId());Assertions.assertEquals(expectedMessage.getRoutingKey(), actualMessage.getRoutingKey());Assertions.assertEquals(expectedMessage.getDeliverMode(), actualMessage.getDeliverMode());Assertions.assertArrayEquals(expectedMessage.getBody(), actualMessage.getBody());Assertions.assertEquals(0x1, actualMessage.getIsValid());}} 6、测试垃圾回收 Testpublic void testGC() throws IOException, MqException, ClassNotFoundException {// 先往队列中写 100 个消息. 获取到文件大小.// 再把 100 个消息中的一半, 都给删除掉(比如把下标为偶数的消息都删除)// 再手动调用 gc 方法, 检测得到的新的文件的大小是否比之前缩小了.MSGQueue queue createTestQueue(queueName1);ListMessage expectedMessages new LinkedList();for (int i 0; i 100; i) {Message message createTestMessage(testMessage i);messageFileManager.sendMessage(queue, message);expectedMessages.add(message);}// 获取 gc 前的文件大小File beforeGCFile new File(./data/ queueName1 /queue_data.txt);long beforeGCLength beforeGCFile.length();// 删除偶数下标的消息for (int i 0; i 100; i 2) {messageFileManager.deleteMessage(queue, expectedMessages.get(i));}// 手动调用 gcmessageFileManager.gc(queue);// 重新读取文件, 验证新的文件的内容是不是和之前的内容匹配LinkedListMessage actualMessages messageFileManager.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(50, actualMessages.size());for (int i 0; i actualMessages.size(); i) {// 把之前消息偶数下标的删了, 剩下的就是奇数下标的元素了.// actual 中的 0 对应 expected 的 1// actual 中的 1 对应 expected 的 3// actual 中的 2 对应 expected 的 5// actual 中的 i 对应 expected 的 2 * i 1Message expectedMessage expectedMessages.get(2 * i 1);Message actualMessage actualMessages.get(i);Assertions.assertEquals(expectedMessage.getMessageId(), actualMessage.getMessageId());Assertions.assertEquals(expectedMessage.getRoutingKey(), actualMessage.getRoutingKey());Assertions.assertEquals(expectedMessage.getDeliverMode(), actualMessage.getDeliverMode());Assertions.assertArrayEquals(expectedMessage.getBody(), actualMessage.getBody());Assertions.assertEquals(0x1, actualMessage.getIsValid());}// 获取新的文件的大小File afterGCFile new File(./data/ queueName1 /queue_data.txt);long afterGCLength afterGCFile.length();System.out.println(before: beforeGCLength);System.out.println(after: afterGCLength);Assertions.assertTrue(beforeGCLength afterGCLength);} 六、小结 MessageFileManager主要是负责管理消息在文件中的存储 1设计了目录结构和文件格式 2实现了目录创建和删除 3实现了统计文件的读写 4实现了消息的写入 5实现了消息的删除 6实现了加载所有消息 7垃圾回收
