息持久化

消息操作并不涉及到复杂的增删改查.消息数量可能会非常多数据库访问的效率不是很高.

1.1.2、存储格式约定

文件中包含若干个消息每个消息都以二进制的方式存储每个消息由两个部分构成

第一个部分约定占用

个字节用来保存消息的长度防止粘包问题.第二个部分为具体的二进制消息数据Message

如下图

validCount;}通过以下方法获取队列对应消息文件的路径以及队列

数据/统计

FileInputStream(getQueueStatPath(queueName)))

{Scanner

Scanner(inputStream);stat.totalCount

scanner.nextInt();stat.validCount

stat;}

FileOutputStream(getQueueStatPath(queueName)))

{PrintWriter

PrintWriter(outputStream);printWriter.write(stat.totalCount

stat.validCount);printWriter.flush();}

catch

RuntimeException(e);}}通过以下方法来创建和销毁文件和*

param

此方法的用处队列也是可以被删除的队列删除之后就需要调用此方法删除对应的消息文件之类的*

param

baseDir.getAbsolutePath());}}1.2.2、实现消息的写入

Message

生产消息了这个消息就需要被记录到文件上(持久化的前提是文件必须要存在)*

param

File(getQueueDataPath(queueName));if(!queueDataFile.exists())

{return

File(getQueueStatPath(queueName));if(!queueStatFile.exists())

{return

将一个新的消息(message)放到队列文件中(queue)*

param

{//1.先检查当前文件是否存在if(!checkFilesExits(queue.getName()))

{throw

MqException([MessageFileManager]

queueName

BinaryTool.toBytes(message);//3.根据当前队列文件长度计算出

Message

长度//这里为了避免写操作引发线程安全问题synchronized(queue)

{File

File(getQueueDataPath(queue.getName()));message.setOffsetBeg(queueDataFile.length()

4);message.setOffsetEnd(queueDataFile.length()

message

FileOutputStream(queueDataFile,

true))

DataOutputStream(outputStream))

{//这里用

只会写一个字节)dataOutputStream.writeInt(messageBinary.length);//写入消息体dataOutputStream.write(messageBinary);dataOutputStream.flush();}}//5.更新消息统计文件内容Stat

stat

readStat(queue.getName());stat.validCount

1;stat.totalCount

随着时间的推移文件可能会越来越大并且可能存在大量的无效消息针对这种情况就需要对当前消息数据文件进行垃圾回收机制后续会讲到.

public

{//读写文件注意线程安全问题synchronized(queue)

{try

RandomAccessFile(getQueueDataPath(queue.getName()),

rw))

message.getOffsetBeg())];randomAccessFile.seek(message.getOffsetBeg());randomAccessFile.read(bufferSrc);

Message

BinaryTool.fromBytes(bufferSrc);//3.把

isValid

设置成无效diskMessage.setIsValid((byte)

message

对象//这个对象马上就会被从内存中删除//4.重新写入文件byte[]

bufferDest

BinaryTool.toBytes(diskMessage);//这里还需要将光标移动到最初这个消息的位置因为

read

操作也会挪动光标randomAccessFile.seek(message.getOffsetBeg());randomAccessFile.write(bufferDest);//

通过上述折腾对于文件来说只有一个字节发生改变了而已}//更新统计文件消息无效了消息个数就需要

-1Stat

readStat(queue.getName());if(stat.validCount

{stat.validCount

1)消息内容加载到内存中(此方法准备在程序启动的时候进行调用因此也不需要加锁)

queueName

FileInputStream(getQueueDataPath(queueName)))

{try

dataInputStream.readInt();//2.按照长度获取消息内容byte[]

buffer

inputStream.read(buffer);//比较理论和实际消息长度if(messageSize

actualSize)

MqException([MessageFileManager]

文件格式错误

BinaryTool.fromBytes(buffer);//4.判断这个消息是否是无效对象if(message.getIsValid()

0x1)

messageSize);continue;}//5.有效数据就加入到链表中加入前计算一下

offsetBeg

offsetEnd//这个位置需要知道当前文件光标的位置由于当下使用的

DataInputStream

手动记录一下message.setOffsetBeg(currentOffset

4);message.setOffsetEnd(currentOffset

messageSize);//6.最后加入到链表当中messages.add(message);}}

catch

业务逻辑这是为了当消息读完了能得到一个反馈(有点顺水推舟的感觉)//因为当消息读取到文件末尾readInt

就会引发异常(EOF异常)System.out.println([MessageFileManager]

Message

具体的我们直接遍历原有的消息数据文件把所有的有效数据拷贝到一个新的文件中再把之前整个旧的文件都删除然后将新文件的名字改为旧文件的名字.

什么时候触发一次

复制算法比较合适的前提是当前空间里有效的数据不多大部分是无效的数据减少搬运数据的开销

2000

readStat(queueName);if(stat.totalCount

2000

这个过程中其他线程不得干预synchronized(queue)

{//由于

System.currentTimeMillis();//1.创建一个新文件File

queueDataNewFile

File(getQueueDataNewPath(queue.getName()));if(queueDataNewFile.exists())

new

MqException([MessageFileManager]

queue_data_new

queueDataNewFile.createNewFile();if(!ok)

{throw

MqException([MessageFileManager]

创建文件失败!

queueDataNewFile.getName());}//2.从旧文件中读出所有的有效消息LinkedListMessage

messages

loadAllMessageFromQueue(queue.getName());//3.把有效消息写入新的文件try(OutputStream

outputStream

FileOutputStream(queueDataNewFile))

dataOutputStream

DataOutputStream(outputStream))

{for(Message

BinaryTool.toBytes(message);//先写消息长度dataOutputStream.writeInt(buffer.length);//再写消息内容dataOutputStream.write(buffer);}}}//4.删除旧文件File

queueDataOldFile

File(getQueueDataPath(queue.getName()));ok

queueDataOldFile.delete();if(!ok)

{throw

MqException([MessageFileManager]

queueDataOldFile

queueDataOldFile.getName());}//把

queue_data_new.txt

queueDataNewFile.renameTo(queueDataOldFile);if(!ok)

{throw

MqException([MessageFileManager]

文件重命名失败!

queueDataNewFile.getAbsolutePath()

queueDataOldFile

queueDataOldFile.getAbsolutePath());}//5.跟新统计文件Stat

stat

readStat(queue.getName());stat.totalCount

messages.size();stat.validCount

messages.size();writeStat(queue.getName(),

stat);long

System.currentTimeMillis();System.out.println([MessageFileManager]

执行完毕queueName

当某个队列中消息特别多并且很多都是有效的消息就会导致后续对这个文件操作的成本上升很多例如文件大小是

10G此时如果触发一次

文件拆分当单个文件长度到达一定阈值以后就会拆分成两个文件.拆着拆着就成了很多文件.文件合并每个单独的文件都会进行

具体实现思路

需要专门的数据结构来存储当前队列中有多少个数据文件每个文件大小是多少消息数目是多少无效消息是多少.设计策略什么时候触发消息拆分什么时候触发文件合并.

Ps这里可以先不给出具体实现需要的可以私信我前提是备注微信号.

二、统一硬盘操作

上层逻辑需要操作硬盘统一通过这个类来操作(上层代码不关心当前数据是存储再数据库还是文件中的)提高了代码的内聚可维护性.

public

DataBaseManager();//这个实例用来管理数据文件中的数据private

MessageFileManager

是一个空方法只是先列在这里一旦后续需要扩展就在这里进行初始化即可messageFileManager.init();}//封装交换机操作public

void

{dataBaseManager.insertExchange(exchange);}public

void

{dataBaseManager.deleteExchange(exchangeName);}public

ListExchange

dataBaseManager.selectAllExchanges();}//封装队列操作public

void

{dataBaseManager.insertQueue(queue);//创建队列的同时不仅需要把队列写入到数据库中还需要创建出对应的删除messageFileManager.destroyQueueFiles(queueName);}public

ListMSGQueue

dataBaseManager.selectAllQueues();}//封装绑定操作public

void

{dataBaseManager.insertBinding(binding);}public

void

{dataBaseManager.deleteBinding(binding);}public

ListBinding

dataBaseManager.selectAllBindings();}//封装消息操作public

void

{messageFileManager.sendMessage(queue,

message);}public

{messageFileManager.deleteMessage(queue,

message);//这里删除消息以后还需要看以下文件中是否有太多的无效文件需要进行清除if(messageFileManager.checkGC(queue.getName()))

{messageFileManager.gc(queue);}}public

ListMessage

selectAllMessagesFromQueue(String

queueName)

messageFileManager.loadAllMessageFromQueue(queueName);}}Ps这里对队列和消息的封装都是具有一定的逻辑的

创建队列的同时不仅需要把队列写入到数据库中还需要创建出对应的删除

删除消息以后还需要看以下文件中是否有太多的无效文件需要进行清除GC