目录
一、引言
二、MessageFileManager类
一、基础部分
二、实现文件读写操作
三、文件实现工作
1.创建文件
2.销毁文件
3.检查队列文件是否存在
4.实现消息的序列化和反序列化
5.实现写入消息文件
6.实现消息加载
7.实现垃圾回收
三、测试MessageFileManager类
1.测试准备工作
2.测试创建文件
3.测试读写文件
4.测试发送消息
5.加载所有消息
6.测试GC回收机制
四、总结
一、引言
消息需要在硬盘上存储，但是并不直接放到数据库中，而是直接使用文件存储。因为对于消息的操作没有复杂的增删改查，并且文件操作的效率是远远高于数据库的。所以我们就用文件进行消息存储
二、MessageFileManager类
一、基础部分
代码：
public class MessageFileManager {static public class Stat{public int totalCount;public int validCount;}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";}
}queue_data.txt:消息数据文件，用来保存信息
queue_stat.txt:消息统计文件，用来保存消息统计信息
二、实现文件读写操作
/*实现消息统计文件读写*/private Stat readStat(String queueName){Stat stat = new Stat();try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){Scanner scanner = new Scanner(inputStream);stat.validCount = scanner.nextInt();stat.totalCount = scanner.nextInt();return stat;} catch (IOException e) {e.printStackTrace();}return null;}/*向统计文件写入结果*/private void writeStat(String queueName,Stat stat){try(OutputStream outputStream = new FileOutputStream(getQueueDataPath(queueName))){PrintWriter printWriter = new PrintWriter(outputStream);printWriter.write(stat.validCount+"\t"+stat.totalCount);printWriter.flush();}catch (IOException e){e.printStackTrace();}}三、文件实现工作
1.创建文件
/*创建文件*/public void createQueueFile(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.给消息文件设定初始值Stat stat = new Stat();stat.validCount=0;stat.totalCount=0;writeStat(queueName,stat);}2.销毁文件
/*销毁文件*/public void destroyQueueFile(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());}}3.检查队列文件是否存在
/*检查队列文件是否存在*/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;}4.实现消息的序列化和反序列化
// 这个逻辑不仅Message能够使用，任何的Java对象都可以使用，来进行序列化和反序列化的操作.
// 如果要使用这样的操作必须实现Serializable 接口
public class BinaryTool {// 将一个对象序列化成字节数组public static byte[] toBytes(Object object) throws IOException {try(ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()){try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream)){objectOutputStream.writeObject(object);}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)){object = objectInputStream.readObject();}}return object;}
}5.实现写入消息文件
自定义业务逻辑异常：
public class MqException extends Exception{public MqException(String reason){super(reason);}
}/*实现写入消息文件*/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.获取当前队列长度，用于计算offsetBeg和offsetEndFile queueDataFile = new File(getQueueDataPath(queue.getName()));message.setOffsetBeg(queueDataFile.length()+4);message.setOffsetEnd(queueDataFile.length()+4 + messageBinary.length);// 4.将消息写入数据文件中，是追加写入，不是重新写入、try (OutputStream outputStream = new FileOutputStream(queueDataFile,true)){try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){dataOutputStream.writeInt(messageBinary.length);dataOutputStream.write(messageBinary);}}// 5.更新消息统计文件Stat stat = readStat(queue.getName());stat.totalCount +=1;stat.validCount +=1;writeStat(queue.getName(),stat);}}6.实现删除消息
此处利用逻辑删除，将isValid改为0
/*删除消息*/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.getOffsetBeg())];randomAccessFile.seek(message.getOffsetBeg());randomAccessFile.read(bufferSrc);// 2.把当前读到的二进制数据转换成Message对象Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);// 3.把isValid设置为无效diskMessage.setIsValid((byte) 0x0);// 4.重新写入文件byte[] bufferDest = BinaryTool.toBytes(diskMessage);randomAccessFile.seek(message.getOffsetBeg());randomAccessFile.write(bufferDest);}// 更新统计文件Stat stat = readStat(queue.getName());if(stat.validCount>0){stat.validCount -=1;}writeStat(queue.getName(),stat);}}6.实现消息加载
/*实现消息加载*/// 使用这个方法将消息从文件中加载到内存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)){long currentSize = 0;while(true){int messageSize = dataInputStream.readInt();byte[] buffer = new byte[messageSize];int actualSize = dataInputStream.read(buffer);if(messageSize != actualSize){throw new MqException("[MessageFileManager] 消息格式有误！queueName="+queueName);}Message message = (Message) BinaryTool.fromBytes(buffer);if(message.getIsValid() != 0x1){currentSize += (4+messageSize);continue;}message.setOffsetBeg(currentSize+4);message.setOffsetEnd(currentSize+4+messageSize);currentSize += (4+messageSize);messages.add(message);}} catch (EOFException e) {System.out.println("[MessageFileManager] 恢复Message数据完成,已处理到末尾，整个文件全部处理完毕！");}}return messages;}7.实现垃圾回收
/*判断是否需要GC操作*/public boolean checkGC(String queueName){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";}/*实现垃圾回收*/public void gc(MsgQueue queue) throws MqException, IOException, ClassNotFoundException {synchronized (queue){// 记录gc开始和结束时间long gcBeg = System.currentTimeMillis();// 1.创建一个新文件File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));if(queueDataNewFile.exists()){throw new MqException("[MessageFileManager] gc时发现该队列的queue_data_new_file已经存在了！queueName="+queue.getName());}boolean ok = queueDataNewFile.createNewFile();if(!ok){throw new MqException("[MessgaeFileManager] 创建文件失败！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());}ok = queueDataNewFile.renameTo(queueDataNewFile);if(!ok){throw new MqException("[MessageFileManager] 更改名字失败！queueDataOldFile="+queueDataOldFile+",queueDataNewFile="+queueDataNewFile);}// 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()+",消耗时间："+(gcEnd-gcBeg)+"ms");}}三、测试MessageFileManager类
1.测试准备工作
// 实例化private MessageFileManager messageFileManager = new MessageFileManager();private static final String queueName1 = "testQueue1";private static final String queueName2 = "testQueue2";@BeforeEachpublic void setUp() throws IOException {messageFileManager.createQueueFile(queueName1);messageFileManager.createQueueFile(queueName2);}@AfterEachpublic void tearDown() throws IOException {messageFileManager.destroyQueueFile(queueName1);messageFileManager.destroyQueueFile(queueName2);}2.测试创建文件
@Testpublic void testCreateFile(){File queueDataFile1 = new File("./data/"+queueName1+"/queue_data.txt");File queueStatFile1 = new File("./data/"+queueName1+"/queue_stat.txt");Assertions.assertEquals(true,queueDataFile1.isFile());Assertions.assertEquals(true,queueStatFile1.isFile());File queueDataFile2 = new File("./data/"+queueName2+"/queue_data.txt");File queueStatFile2 = new File("./data/"+queueName2+"/queue_stat.txt");Assertions.assertEquals(true,queueDataFile2.isFile());Assertions.assertEquals(true,queueStatFile2.isFile());}3.测试读写文件
// 读写统计文件@Testpublic void testReadWriteStat(){MessageFileManager.Stat stat = new MessageFileManager.Stat();stat.validCount = 50;stat.totalCount = 100;// 此处要调用writeStat和readStat，但是这两个方法的修饰词都是private// 所以此处使用反射的方式来进行ReflectionTestUtils.invokeMethod(messageFileManager,"writeStat",queueName1,stat);MessageFileManager.Stat newStat = ReflectionTestUtils.invokeMethod(messageFileManager,"readStat",queueName1);Assertions.assertEquals(50,newStat.validCount);Assertions.assertEquals(100,newStat.totalCount);}4.测试发送消息
// 创建一个测试队列private MsgQueue createTestQueue(String queueName){MsgQueue queue = new MsgQueue();queue.setName(queueName);queue.setDurable(true);queue.setExclusive(false);queue.setAutoDelete(false);return queue;}// 创建一个测试消息private Message createTestMessage(String content){Message message = Message.createMessageWithId("testRoutingKey",null,content.getBytes());return message;}// 测试发送消息@Testpublic void testSendMessage() throws IOException, MqException, ClassNotFoundException {Message message = createTestMessage("testMessage");MsgQueue queue = createTestQueue(queueName1);messageFileManager.sendMessage(queue,message);MessageFileManager.Stat stat = ReflectionTestUtils.invokeMethod(messageFileManager,"readStat",queueName1);Assertions.assertEquals(1,stat.totalCount);Assertions.assertEquals(1,stat.validCount);LinkedList<Message> messages = messageFileManager.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(1,messages.size());Message curMessage = messages.get(0);Assertions.assertEquals(message.getMessageId(),curMessage.getMessageId());Assertions.assertEquals(message.getRoutingKey(),curMessage.getRoutingKey());Assertions.assertEquals(message.getDeliverMode(),curMessage.getDeliverMode());Assertions.assertArrayEquals(message.getBody(),curMessage.getBody());System.out.println("message:"+curMessage);}5.加载所有消息
// 测试加载所有消息@Testpublic void testLoadAllMessageFromQueue() throws IOException, MqException, ClassNotFoundException {MsgQueue queue = createTestQueue(queueName1);LinkedList<Message> expectedMessages = new LinkedList<>();for(int i=0;i<100;i++){Message message = createTestMessage("testMessage"+i);messageFileManager.sendMessage(queue,message);expectedMessages.add(message);}LinkedList<Message> actualMessages = messageFileManager.loadAllMessageFromQueue(queueName1);Assertions.assertEquals(expectedMessages.size(),actualMessages.size());for(int i=0;i<actualMessages.size();i++){Message expectedMessage = expectedMessages.get(i);Message actualMessage = actualMessages.get(i);Assertions.assertEquals(expectedMessage.getMessageId(),actualMessage.getMessageId());Assertions.assertEquals(expectedMessage.getDeliverMode(),actualMessage.getDeliverMode());Assertions.assertEquals(expectedMessage.getRoutingKey(),actualMessage.getRoutingKey());Assertions.assertArrayEquals(expectedMessage.getBody(),actualMessage.getBody());Assertions.assertEquals(0x1,actualMessage.getIsValid());}}6.测试删除信息
// 测试删除信息@Testpublic void testDeleteMessage() throws IOException, MqException, ClassNotFoundException {MsgQueue queue = createTestQueue(queueName1);LinkedList<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("["+"] 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.测试GC回收机制
// 测试垃圾回收GC@Testpublic void testGC() throws IOException, MqException, ClassNotFoundException {// 先往队列中写100个消息// 再把100个消息中的一半给删除掉// 再手动调用GCMsgQueue queue = createTestQueue(queueName1);LinkedList<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();System.out.println("1");// 删除偶数下标的消息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++){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);}四、总结
本篇文章主要介绍了一下消息文件的存储以及消息发送删除加载到内存以及gc操作的代码，在每写一个部分的代码之后都要编写测试用例用于检测代码是否有误，编写测试用例的时候需要有耐心，因为小编自己在写测试代码的时候也会碰到各种各样的问题，并要花很长的时间去解决代码bug，所以测试的时候要有耐心，下一篇文章，我们将进行统一硬盘处理代码的编写，感谢观看！