RocketMQ的数据存储部分存储Producer生产的消息,Consume的逻辑队列,索引,以及主从复制
/RocketMQ的数据存储部分也是一个重头戏,他主要用于存储Producer生产的消息,Consume的逻辑队列,索引,以及主从复制,这里也是一个非常好的范例,我们可以看到如何处理数据存储,如何提高IO效率。
东西太多信息量略大有点乱,之后再整理吧......
一、总体结构
队列逻辑结构:
逻辑上来看,每一个Topic有很多queue,他们各自处于自己的ConsumeQueue进行处理
二、基础数据结构
MappedFileQueue:包含了很多MapedFile,以及每个MappedFile的真实大小;
MappedFile:包含了具体的文件信息,包括文件路径,文件名,文件起始偏移,写位移,读位移等等信息,同事使用了虚拟内存映射来提高IO效率;
这两个数据结构是真实的保存了存放在物理机器上的文件信息,后续的很多模块如果涉及到文件存储,都会使用到这两个数据结构。
MappedFileQueue:
// 每次触发删除文件,最多删除多少个文件
private static final int DELETE_FILES_BATCH_MAX = 10;
// 文件存储位置
private final String storePath;
// 每个文件的大小
private final int mappedFileSize;
// 各个文件,这里是一个CopyOnWriteArrayList的使用实例,用于读多写少的线程安全的list容器
private final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();
// 预分配MapedFile对象服务
private final AllocateMappedFileService allocateMappedFileService;
// 刷盘刷到哪里
private long flushedWhere = 0;
private long committedWhere = 0;
// 最后一条消息存储时间
private volatile long storeTimestamp = 0;
MappedFile:保存了文件的详细信息,包括:
TotalMapedVitualMemory:JVM中映射的虚拟内存总大小
TotalMapedFiles:JVM中mmap的数量
fileName:文件名
fileFromOffset:文件的起始偏移量
fileSize:文件大小
file:文件句柄
mappedByteBuffer:映射的内存对象
wrotePostion:当前文件的写位置
committedPosition:当前文件Flush到的位置
fileChannel:映射的FileChannel对象
storeTimestamp:最后一条消息保存时间
firstCreateInQueue:是不是刚刚创建的Map
三、主要模块介绍
3.1 CommitLog(物理队列):
CommitLog是用于存储真实的物理消息的结构,ConsumeQueue是逻辑队列,仅仅存储了CommitLog的位移而已,真实的存储都在本结构中。
1. 首先这里会使用CommitLog.this.topicQueueTable.put(key, queueOffset),其中的key是 topic-queueId, queueOffset是当前这个key中的消息数,每增加一个消息增加一(不会自减);
这里queueOffset的用途如下:每次用户请求putMessage的时候,将queueOffset返回给客户端使用
2. 其次就是将消息写入真正的MappedFile,MappedFile中,每条消息的格式如下:
MagicCode:MessageMagicCode = 0xAABBCCDD ^ 1880681586 + 8
BodyCRC:对Body的校验码
queueId:队列Id
queueOffset:topic-queueId对应的队列数目
physicalOffset:真实的地址偏移,包括所有文件的总偏移
3. putMessage所做操作
3.1 将数据写入mapedFile,并将topic-queueId,queueOffset写入topicQueueTable
3.2 创建DispatchRequest,将消息放入DispatchMessageService的List中,进行后续处理
3.3 进行同步刷盘,或者异步刷盘
FlushCommitLogService:刷盘基类;
GroupCommitService:同步刷盘;每次写入消息时,调用swapRequests进行数据交换,然后将所有的请求进行刷盘
FlushRealTimeService:异步刷盘;调用mappedByteBuffer.force()进行刷盘操作
3.2 ConsumeQueue(逻辑队列):
1.ConsumeQueue的结构
该结构对应于消费者逻辑队列,为什么要将一个topic抽象出很多的queue呢?这样的话,对集群模式更有好处,可以使多个消费者共同消费,而不用上锁;
对应于consumeQueueTable <topic, <queueId,ConsumeQueue>>,其中每一个ConsumeQueue的结构如下:
offset: CommitLog中的物理位移
size: CommitLog中的日志大小
tagsCode:和storeTimestamp相关
2. ConsumeQueue创建过程:
1. 首先会创建CommitLog,在将数据写入CommitLog之后,会创建DispatchRequest,调用defaultMessageStore.putDispatchRequest
2. DispatchMessageService调用putMessagePostionInfo将数据写入ConsumeQueue
3.3 IndexService(索引)
IndexService用于创建索引文件集合,当用户想要查询某个topic下某个key的消息时,能够快速响应;这里注意不要与上述的ConsumeQueue混合,ConsumeQueue只是为了抽象出多个queue,方便并发情况下,用户put/get消息。
IndexService由一系列的IndexFile文件组成:
// 索引文件集合
private final ArrayList<IndexFile> indexFileList
1. IndexFile的结构:
IndexFile的格式如下所示:
其中:
A. IndexHeader的结构如下
beginTimestamp = new AtomicLong(0);
endTimestamp = new AtomicLong(0);
beginPhyOffset = new AtomicLong(0);
endPhyOffset = new AtomicLong(0);
hashSlotCount = new AtomicInteger(0);
indexCount = new AtomicInteger(1);
B. HashSlot里面的每一项保存了这个topic-key计算出的hash的Index,所有的链表,所以他的每个大小是4字节,一共有500W项
C.Index的结构如下:
keyHash:topic-key(key是消息的key)的hashCode组成
phyOffset:commitLog真实的物理位移
timeOffset:时间位移
slotValue:下一个记录的slot位置
2. IndexFile的创建过程:
1. 首先在DispatchMessageService写入ConsumeQueue后,会再调用indexService.putRequest,创建索引请求
2. 调用IndexService的buildIndex创建索引
3.4 HAService(主从复制模块)
他的启动有3个部分:
3.4.1
AcceptSocketService:作为HAServer进行监听
beginAccept,start;
主要就是在socketAddressListen监听Slave的请求服务,如果有请求进入,新建HAConnection,并start;
同时将其加入到HAService的connectionList中
3.4.2
HAConnection:用于管理Master接收的连接,readSocketService和writeSocketService分别启动;
有slaveRequestOffset和slaveAckOffset两个变量:
slaveRequestOffset:它只使用一次,即第一次slave发送给master时,表示slave请求的起始offset;
slaveAckOffset:由slave上传到master,表示其确认接收到的offset
A. ReadSocketService:接收处理slave的请求
首先接收到slave的请求后,slave的请求是一个8字节的Long:
先判断slaveRequestOffset是否小于0,小于0则表示第一次请求,将slaveRequestOffset置为此值,需要从这个位置进行commitLog拉取;
否则将slaveAckOffset置为slave请求的offset,表示确认接收到该请求;
再调用notifyTransferSome(slaveAckOffset)
B. WriteSocketService:发送给slave物理数据CommitLog
A. 初始化:如果nextTransferFromWhere=-1,则表示是第一次传输;slaveRequestOffset如果也为0,表示slave上面没有数据,直接从master的最后一个文件开始上传,将nextTransferFromWhere置为最后一个文件的位移;如果slaveRequestOffset不为0,则从slave请求的位置开始传输;
B. 心跳包:如果本次传输距离上次的时间超过了5S的心跳时间,那么发送心跳包:8(下次传输的位置)+4(固定值0)
C. 数据包:获取从nextTransferFromWhere开始的SelectMapedBufferResult(MapFile,size, startOffset);如果size大于传输的BatchSize,则设置size为BatchSize;接下来:
设置nextTranserFromWhere;
设置byteBufferHeader设置头部:8(传输的offset)+4(内容长度)
调用transferData()传输数据:首先传输头部,再根据selectMapedBufferResult传输body部分;
如果没有数据,则
WaitNotifyObject:感觉用于多个线程之间的同步
notifyTransferSome(slaveAckOffset)
如果客户端ack的值比当前push2SlaveMaxOffset的值大,那么将push2SLaveMaxOffset的值设置为ack,然后调用groupTranserServer唤起线程
groupTransferService:它用于同步slave-master的模式,当master收到commitLog,同时又是同步复制的情况,就会将Requets放入,随后进行复制
push2SlaveMaxOffset的值为slaveAckOffset
3.4.3
HaClient
1. 首先连接到master,同时获取到最大的offset,到currentReportedOffset
2. 向master发送心跳
3. 不断向master发送当前的offset,同时读取master的答复,将master的答复写入commitLog中;如果master的答复不满足我们slave的offset,则关闭与master之间的连接
四、其他模块
4.1 FlushConsumeQueueService:用于将ConsumeQueue的File文件写入入里磁盘;
首先判断是否到达了刷盘时间,如果到达了,那么全盘通刷;
否则,遍历所有的ConsumeQueue,调用cq.commit(flushConsumeQueueLeastPages)进行刷盘,flushConsumeQueueLeastPages是目前文件的未刷盘大小达到flushConsumeQueueLeastPages*OS_PAGE_SIZE(1024*4)个,才进行刷盘
4.2 CleanCommitLogService:用于定期删除CommitLog的物理文件;
如果到达删除时间或者磁盘大小达到阈值:
文件保存时长:72小时
删除文件间隔时间:100ms
强制删除文件间隔时间:1000* 120ms
一次最多删除文件数:10个
4.3 CleanConsumeQueueService
获取commitLog的最小offset,遍历所有的consumeQueue,将小于offset的consumeQueue删除;
同时遍历IndexFile,将offset小于commitLog的索引删除
4.4 DispatchMessageService:这个服务主要是处理DispatchRequest,当写入CommitLog后,会将Dispatch请求写入Queue,然后该服务进行ConsumeQueue和IndexService的写入
4.5 AllocateMapedFileService:包含在MapedFileQueue中,主要用于每次getLastMapedFile(startOffset)时,如果最后一个文件满了,那么调用allocateMapedFileService.putRequestAndReturnMapedFile,将文件创建请求放入allocateMapedFileService的requestTable和requestQueue中,后续会进行文件创建,如果超过5S还没创建,则报错;
4.6 ReputMessageService:这个用于Slave中,当Slave模式时,会启动这个模块,然后首先设置reputFormOffset为初始值0;不断遍历commitLog,得到其SelectMapedBufferResult,然后对每一个Message创建DispatchRequest,创建对应的ConsumeQueue和IndexFile
4.7 ScheduleMessageService
4.8 StoreStatsService
4.9 StoreCheckpoint
4.10 BrokerStatsManager
附:
A.
同步请求的处理方式和异步请求的处理方式统一:
同步请求和异步请求都是将Request放入Queue中,然后再起一个线程进行处理
B. 双缓冲
可以起2个队列,一个读,一个写,这两个队列不断交换;这样可以解决一个问题,写和读不会出现并发问题,不需要上锁,能提高效率,但是仅仅适用于对put顺序没有要求的情况
C. 为什么要逻辑队列:可以使写效率更高