目录
一、引言
消息需要在硬盘上存储，但是并不直接放到数据库中，而是直接使用文件存储。因为对于消息的操作没有复杂的增删改查，并且文件操作的效率是远远高于数据库的。所以我们就用文件进行消息存储
二、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和offsetEnd
File 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";
@BeforeEach
public void setUp() throws IOException {
messageFileManager.createQueueFile(queueName1);
messageFileManager.createQueueFile(queueName2);
}
@AfterEach
public void tearDown() throws IOException {
messageFileManager.destroyQueueFile(queueName1);
messageFileManager.destroyQueueFile(queueName2);
}2.测试创建文件
@Test
public 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.测试读写文件
// 读写统计文件
@Test
public 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;
}
// 测试发送消息
@Test
public 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.加载所有消息
// 测试加载所有消息
@Test
public 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.测试删除信息
// 测试删除信息
@Test
public 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
@Test
public void testGC() throws IOException, MqException, ClassNotFoundException {
// 先往队列中写100个消息
// 再把100个消息中的一半给删除掉
// 再手动调用GC
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);
}
// 获取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));
}
// 手动调用GC
messageFileManager.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，所以测试的时候要有耐心，下一篇文章，我们将进行统一硬盘处理代码的编写，感谢观看！