一、消息存储格式设计

🍅 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进行序列化。

针对二进制序列化，有很多解决方案：

（1）Java标准库中提供了序列化的方案：ObjectInputStream 和 ObjectOutputStream

（2）Hession

（3）protobuffer

（4）thrift

这里使用第一种，这样就不用再引入额外的依赖。

在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{
//        直接定义成public
        public 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\t0
        Stat 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 LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException,MqException,ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
//               该变量记录当前文件光标位置,初始位置为0
                long 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，根据总消息数目和有效消息数目判断。

//    检查当前是否要针对该队列的消息数据文件进行GC
    public 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. 从旧的文件中, 读取出所有的有效消息对象了. (这个逻辑直接调用上述方法即可, 不必重新写了)
            LinkedList<Message> 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.txt
            ok = 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";

//  每个用例执行之前的准备工作
    @BeforeEach
    public void setUp() throws IOException {
//        准备阶段，创建出两个队列，以备后用
        messageFileManger.createQueueFiles(queueName1);
        messageFileManger.createQueueFiles(queueName2);

    }


//    每个用例执行之后的收尾工作
    @AfterEach
    public void tearDown() throws IOException {
//        收尾阶段，把创建出的队列销毁掉
        messageFileManger.destroyQueueFiles(queueName1);
        messageFileManger.destroyQueueFiles(queueName2);
    }
}

🍅 2、测试创建文件是否存在

@Test
    public 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是否能够通过

@Test
    public 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：

@Test
    public 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);

//        检查文件,把消息读出来
        LinkedList<Message> 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());

//      比较两个字节数组的内容是否相同，不能直接使用asserEquals
        Assertions.assertArrayEquals(message.getBody(),curMessage.getBody());

        System.out.println("message = "+ curMessage);
    }

构造100条消息，并且读取出来

 @Test
    public void testLoadAllMessageFromQueue() throws IOException, MqException, ClassNotFoundException {
//      往队列中插入100条消息，验证100条消息从文件中读取之后，是否和最初是一致的
        MSGQueue queue = createTestQueue(queueName1);
        List<Message> expectedMessages = new LinkedList<>();
        for (int i = 0; i < 100; i++) {
            Message message = createTestMessage("testMessge" + 1);
            messageFileManger.sendMessage(queue,message);
            expectedMessages.add(message);
        }

//        读取所有消息
        LinkedList<Message> 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、测试删除消息

//    测试删除消息
    @Test
    public void testDeleteMessage() throws IOException, MqException, ClassNotFoundException {
        // 创建队列, 写入 10 个消息. 删除其中的几个消息. 再把所有消息读取出来, 判定是否符合预期.
        MSGQueue queue = createTestQueue(queueName1);
        List<Message> 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));

        // 对比这里的内容是否正确.
        LinkedList<Message> 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、测试垃圾回收

 @Test
    public void testGC() throws IOException, MqException, ClassNotFoundException {
        // 先往队列中写 100 个消息. 获取到文件大小.
        // 再把 100 个消息中的一半, 都给删除掉(比如把下标为偶数的消息都删除)
        // 再手动调用 gc 方法, 检测得到的新的文件的大小是否比之前缩小了.
        MSGQueue queue = createTestQueue(queueName1);
        List<Message> 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));
        }

        // 手动调用 gc
        messageFileManager.gc(queue);

        // 重新读取文件, 验证新的文件的内容是不是和之前的内容匹配
        LinkedList<Message> 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 + 1
            Message 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);
    }