浅析RocketMQ-存储文件

RocketMQ中存储文件主要分为CommitLog，Comsumequeue,Index 三类文件。

1. CommitLog

CommitLog文件用于存储消息数据，所有请求的消息都与记录在这个文件。在commitlog文件夹下，会存在如下文件：

文件名会以20位的数字组成，以0开始，每个文件相差1G大小，下个文件名在上个文件名基础上+1G的数值。文件名数值不足20位，自动补0.
每个msg在commitlog中存储结构如下，4个字节的消息长度+其他消息数据

具体的存储格式可以参考CommitLog类下的MessageExtEncoder.encode方法

protected ByteBuffer encode(final MessageExtBatch messageExtBatch, PutMessageContext putMessageContext) {
           		....
                // 1 TOTALSIZE
                this.encoderBuffer.putInt(msgLen);
                // 2 MAGICCODE
                this.encoderBuffer.putInt(CommitLog.MESSAGE_MAGIC_CODE);
                // 3 BODYCRC
                this.encoderBuffer.putInt(bodyCrc);
                // 4 QUEUEID
                this.encoderBuffer.putInt(messageExtBatch.getQueueId());
                // 5 FLAG
                this.encoderBuffer.putInt(flag);
                // 6 QUEUEOFFSET
                this.encoderBuffer.putLong(0);
                // 7 PHYSICALOFFSET
                this.encoderBuffer.putLong(0);
                // 8 SYSFLAG
                this.encoderBuffer.putInt(messageExtBatch.getSysFlag());
                // 9 BORNTIMESTAMP
                this.encoderBuffer.putLong(messageExtBatch.getBornTimestamp());
                // 10 BORNHOST
                this.resetByteBuffer(bornHostHolder, bornHostLength);
                this.encoderBuffer.put(messageExtBatch.getBornHostBytes(bornHostHolder));
                // 11 STORETIMESTAMP
                this.encoderBuffer.putLong(messageExtBatch.getStoreTimestamp());
                // 12 STOREHOSTADDRESS
                this.resetByteBuffer(storeHostHolder, storeHostLength);
                this.encoderBuffer.put(messageExtBatch.getStoreHostBytes(storeHostHolder));
                // 13 RECONSUMETIMES
                this.encoderBuffer.putInt(messageExtBatch.getReconsumeTimes());
                // 14 Prepared Transaction Offset, batch does not support transaction
                this.encoderBuffer.putLong(0);
                // 15 BODY
                this.encoderBuffer.putInt(bodyLen);
                if (bodyLen > 0)
                    this.encoderBuffer.put(messagesByteBuff.array(), bodyPos, bodyLen);
                // 16 TOPIC
                this.encoderBuffer.put((byte) topicLength);
                this.encoderBuffer.put(topicData);
                // 17 PROPERTIES
                this.encoderBuffer.putShort((short) totalPropLen);
                if (propertiesLen > 0) {
                    this.encoderBuffer.put(messagesByteBuff.array(), propertiesPos, propertiesLen);
                }
                if (batchPropLen > 0) {
                    if (needAppendLastPropertySeparator) {
                        this.encoderBuffer.put((byte) MessageDecoder.PROPERTY_SEPARATOR);
                    }
                    this.encoderBuffer.put(batchPropData, 0, batchPropLen);
                }
           	....
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

2. Comsumequeue

Comsumequeue 出现是为了提升消息查询的效率，毕竟在大的commitlog的查找某个topic的效率太低了。
在存储结构上，它首先在comsumumequeue文件中以topic文件夹进行区分，然后再topic文件夹下又以队列0~n的方式做划分，最后才是到具体的文件.结构如下图:



每个文件默认包含30w个条目，每个条目包含8个字节commitlog的偏移量+消息的总长度+消息的tag的hashCode.
每个条目存储结构如下:

对应的代码在ComsumeQueue的putMessagePositionInfo方法中

private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
        final long cqOffset) {
        ...

        this.byteBufferIndex.flip();
        // CQ_STORE_UNIT_SIZE 为20
        this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
        this.byteBufferIndex.putLong(offset);
        this.byteBufferIndex.putInt(size);
        this.byteBufferIndex.putLong(tagsCode);

       ...
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

3. Index

Index使用与提升检索的索引文件，用于提升topic和消息队列之间的检索速度的。
该文件由三部分组成:IndexHead，Hash槽，条目列表

1. IndexHeader 头部，包含 40 个字节，记录该 IndexFile 的统计信息 ,包含如下子弹
   beginTimestamp 该索 文件中 含消息 存储时间
   end Times tamp 该索引文件中包含消息的最大存储时
   beginPhyoffset 该索引文件中包含消息的最小物理偏移量（ ommitlog 文件偏移量）
   endPhyoffset ：该 文件中包含消息 最大物理偏移量（ commitlog 件偏移量）
   hashslotCount: hashslot 数，并不是 hash 槽使用的个数，在这里意义不大
2. Hash 槽
   Index文件默认包含 500 万个 Hash 槽，每个 Hash 槽存储的是落在该槽的hashcode最新的Index的索引
3. 条目列表，默认 个索引文件包含 2000 万个条目，每个条目结构如下
   hashcode: key hashcode
   phyoffset 消息对应的物理偏移
   timedif：该消息存储时间与第 条消息的时间戳的差值，小于0该消息无效
   prelndexNo ：该条目的前一条 Index 引
   整体结构如下:
   
   对应的代码在InexFile的putKey中

public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        ...

                int absIndexPos =
                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                        + this.indexHeader.getIndexCount() * indexSize;

                this.mappedByteBuffer.putInt(absIndexPos, keyHash);
                this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);

                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());

                if (this.indexHeader.getIndexCount() <= 1) {
                    this.indexHeader.setBeginPhyOffset(phyOffset);
                    this.indexHeader.setBeginTimestamp(storeTimestamp);
                }

                if (invalidIndex == slotValue) {
                    this.indexHeader.incHashSlotCount();
                }
                this.indexHeader.incIndexCount();
                this.indexHeader.setEndPhyOffset(phyOffset);
                this.indexHeader.setEndTimestamp(storeTimestamp);

              ....
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28