项目实战 — 消息队列(4){消息持久化}

目录

 一、消息存储格式设计

       🍅 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{
//        直接定义成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 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)){
//               该变量记录当前文件光标位置,初始位置为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. 从旧的文件中, 读取出所有的有效消息对象了. (这个逻辑直接调用上述方法即可, 不必重新写了)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.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);//        检查文件,把消息读出来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());//      比较两个字节数组的内容是否相同,不能直接使用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);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、测试删除消息

//    测试删除消息@Testpublic 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、测试垃圾回收

 @Testpublic 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));}// 手动调用 gcmessageFileManager.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 + 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)垃圾回收

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/79251.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

迅为全国产龙芯3A5000电脑运行统信UOS、银河麒麟、loongnix系统

iTOP-3A5000开发板采用全国产龙芯3A5000处理器&#xff0c;基于龙芯自主指令系统 (LoongArch) 的LA464微结构&#xff0c;并进一步提升频率&#xff0c;降低功耗&#xff0c;优化性能。在与龙芯3A4000处理器保持引脚兼容的基础上&#xff0c;频率提升至2.5GHZ&#xff0c;功耗降…

谷粒商城第九天-解决商品品牌问题以及前后端使用检验框架检验参数

目录 一、总述 二、商品分类问题 三、前端检验 四、后端检验 五、总结 一、总述 在完成完商品分类的时候&#xff0c;后来测试的时候还是发现了一些问题&#xff0c;现在将其进行解决&#xff0c;问题如下&#xff1a; 1. 取消显示的时候&#xff0c;如果取消了显示&…

九度OJ → 题目1368:二叉树中和为某一值的路径 ← DFS

【题目来源】 由于九度OJ&#xff08;http://ac.jobdu.com/&#xff09;已经永久关闭&#xff0c;故无法在其上进行在线提交代码。 幸运的是&#xff0c;在AcWing上有此题目“二叉树中和为某一值的路径”&#xff0c;但描述有些不同。可详见&#xff1a;https://www.acwing.com…

redis 集群 1:李代桃僵 —— Sentinel

目前我们讲的 Redis 还只是主从方案&#xff0c;最终一致性。读者们可思考过&#xff0c;如果主节点凌晨 3 点突发宕机怎么办&#xff1f;就坐等运维从床上爬起来&#xff0c;然后手工进行从主切换&#xff0c;再通知所有的程序把地址统统改一遍重新上线么&#xff1f;毫无疑问…

绿色项目管理:为环境和效益双赢

绿色项目管理&#xff1a;为环境和效益双赢 在21世纪的今天&#xff0c;我们正面临着各种全球性的环境问题&#xff0c;从气候变化到资源枯竭。作为项目经理&#xff0c;我们有责任和机会确保我们的项目对环境的影响最小&#xff0c;并在可能的情况下为环境做出积极的贡献。 …

P9503 『MGOI』Simple Round I | B. 魔法照相馆

题目背景 照片留下了值得留恋的瞬间&#xff0c;但对于魔法士来说最重要的是向前看。——殿堂魔法士 W 题目描述 小 M 正在准备入学所必需的魔法士证件&#xff0c;因此他来到了纵深巷的魔法照相馆。 在等待的时候&#xff0c;小 M 注意到魔法照相馆有三个幕布&#xff0c;颜…

建筑行业是不是一个夕阳产业?一建值得考吗?

建筑行业行业是不是夕阳产业&#xff0c;提出这个问题的人已经了解了现在建筑行业的不景气&#xff0c;从3年疫情到放开疫情管控&#xff0c;2023年是疫情放开后最好的一年。 建筑上下游各产业链行业失业率攀升&#xff0c;房屋建筑土地没人卖&#xff0c;多修建公共建筑&…

RISC-V公测平台发布:如何在SG2042上玩转OpenMPI

About HS-2 HS-2 RISC-V通用主板是澎峰科技与合作伙伴共同研发的一款专为开发者设计的标准mATX主板&#xff0c;它预装了澎峰科技为RISC-V高性能服务器定制开发的软件包&#xff0c;包括各种标准bencmark、支持V扩展的GCC编译器、计算库、中间件以及多种典型服务器应用程序。…

音频客观感知MOS对比,对ViSQOL、PESQ、MosNet(神经网络MOS分)和polqa一致性对比和可信度验证

原创&#xff1a;转载需附链接&#xff1a; https://blog.csdn.net/qq_37100442/article/details/132057139?spm1001.2014.3001.5502 一、背景 Mos分评价音质重要指标&#xff0c;最近也有很多机构和公司在研究适合自己的评价体系。目前Mos分主要分为主观评测和客观感知评价。…

6.6 实现卷积神经网络LeNet训练并预测手写体数字

模型架构 代码实现 import torch from torch import nn from d2l import torch as d2lnet nn.Sequential(nn.Conv2d(1,6,kernel_size5,padding2),nn.Sigmoid(),#padding2补偿5x5卷积核导致的特征减少。nn.AvgPool2d(kernel_size2,stride2),nn.Conv2d(6,16,kernel_size5),nn.S…

黑马大数据学习笔记5-案例

目录 需求分析背景介绍目标需求数据内容DBeaver连接到Hive建库建表加载数据 ETL数据清洗数据问题需求实现查看结果扩展 指标计算需求需求指标统计 可视化展示BIFineBI的介绍及安装FineBI配置数据源及数据准备 可视化展示 P73~77 https://www.bilibili.com/video/BV1WY4y197g7?…

宋浩概率论笔记(三)随机向量/二维随机变量

第三更&#xff1a;本章的内容最重要的在于概念的理解与抽象&#xff0c;二重积分通常情况下不会考得很难。此外&#xff0c;本次暂且忽略【二维连续型随机变量函数的分布】这一章节&#xff0c;非常抽象且难度较高&#xff0c;之后有时间再更新。

回归决策树模拟sin函数

# -*-coding:utf-8-*- import numpy as np from sklearn import tree import matplotlib.pyplot as pltplt.switch_backend("TkAgg") # 创建了一个随机数生成器对象 rng rngnp.random.RandomState(1) print("rng",rng) #5*rng.rand(80,1)生成一个80行、1列…

恒盛策略:上交所过户费收费标准?

随着我国股市的发展&#xff0c;股票买卖所的过户费成为了广阔投资者关注的焦点之一。在我国股票商场中&#xff0c;上交所是最重要的买卖所之一&#xff0c;因而上交所过户费的收费规范备受到了广泛关注。那么&#xff0c;上交所过户费的收费规范究竟怎么拟定&#xff1f;对投…

【Docker】Docker私有仓库的使用

目录 一、搭建私有仓库 二、上传镜像到私有仓库 三、从私有仓库拉取镜像 一、搭建私有仓库 首先我们需要拉取仓库的镜像 docker pull registry 然后创建私有仓库容器 docker run -it --namereg -p 5000:5000 registry 这个时候我们可以打开浏览器访问5000端口看是否成功&…

python-opencv对极几何 StereoRectify

OpenCV如何正确使用stereoRectify函数 函数介绍 用于双目相机的立体校正环节中&#xff0c;这里只谈谈这个函数怎么使用&#xff0c;参数具体指哪些函数参数 随便去网上一搜或者看官方手册就能得到参数信息&#xff0c;但是&#xff01;&#xff01;相对关系非常容易出错&…

【MySQL】事务的多版本并发控制(MVCC)

目录 一、数据库并发的三种场景二、MVCC2.1 三个记录隐藏字段2.2 undo log&#xff08;撤销日志&#xff09;2.3 模拟MVCC2.3.1 模拟更新&#xff08;update&#xff09;2.3.1 模拟删除&#xff08;delete&#xff09;2.3.1 模拟插入&#xff08;insert&#xff09;2.3.1 模拟查…

maven中常见问题

文章目录 一、配置项提示二、父子打包三、打包之后不显示target四、自定义打包之后的jar包名称五、整个项目打包5.1、父项目管理插件和微服务打包 一、配置项提示 SpringBoot中提示错误信息 表示的是SpringBoot中的注释提示没有配置&#xff01;那么可以来使用一下springboot官…

安全学习DAY14_JS信息打点

信息打点——前端JS框架 文章目录 信息打点——前端JS框架小节概述-思维导图JS安全概述什么是JS渗透测试&#xff1f;前后端差异JS安全问题流行的Js框架如何判定JS开发应用&#xff1f; 测试方法&#xff08;JS文件的获取以及分析方法1、手工搜索分析2、半自动Burp分析插件介绍…

problem(3):python IDE和python解释器

为什么写这篇文章呢&#xff1f;遇到了下面的问题&#xff0c;相同的解释器&#xff0c;如果运行angr库的代码&#xff0c;会出现 这样的情况&#xff0c;但是用spyder IDE 会显示正常&#xff0c;很奇怪 应该就是IDE的原因 IDE的循环导入问题 检查IDE配置&#xff1a; 如果可…