RocketMQ ConsumeQueue與IndexFile實時更新機制源碼解析

// 構建Message參數
Message msg = new Message("TopicTest",  // 消息topic
    "TagA",															// 消息Tag
    "key1 key2 key3",										// 消息keys，多個key用" "隔開
    "hello linshifu!".getBytes(RemotingHelper.DEFAULT_CHARSET)); // 消息體

// org.apache.rocketmq.store.index.IndexHeader
public class IndexHeader {
    // 索引文件第一條消息在commitLog中的存儲時間
    private final AtomicLong beginTimestamp = new AtomicLong(0);
    // 索引文件最後一條消息在commitLog中的存儲時間
    private final AtomicLong endTimestamp = new AtomicLong(0);
    // 索引文件第一條消息的偏移量
    private final AtomicLong beginPhyOffset = new AtomicLong(0);
    // 索引文件最後一條消息的偏移量
    private final AtomicLong endPhyOffset = new AtomicLong(0);
    // 已經填充slot的hash槽數量
    private final AtomicInteger hashSlotCount = new AtomicInteger(0);
    // 該indexFile種包含的索引單元數量
    private final AtomicInteger indexCount = new AtomicInteger(1);
}

// org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService#run
@Override
public void run() {
		// 死循環
    while (!this.isStopped()) {
        try {
             // 睡眠1ms
            Thread.sleep(1);
						// 更新consumeQueue和IndexFile
            this.doReput();
        } catch (Exception e) {
            DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }
}

// org.apache.rocketmq.store.SelectMappedBufferResult
public class SelectMappedBufferResult {
    // mappedFile文件起始偏移量+position
    private final long startOffset;
    // reputFromOffset開始的緩沖
    private final ByteBuffer byteBuffer;
    // 消息size
    private int size;
    // commitLog的MappedFile
    private MappedFile mappedFile;
}

// org.apache.rocketmq.store.DefaultMessageStore.ReputMessageService#doReput
private void doReput() {
    // 1.reputFromOffset ≤ commitLog最大offset，則循環重放
    for (boolean doNext = true; this.isCommitLogAvailable()/*reputFromOffset≤commitLog最大offset*/&&doNext; ) {
        // 2.根據reputFromOffset的物理偏移量找到mappedFileQueue中對應的CommitLog文件的MappedFile
        // 然後從該MappedFile中截取一段自reputFromOffset偏移量開始的ByteBuffer，這段內存存儲著將要重放的消息
        SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);
        if (result != null) {
            try {
                // 遍歷消息，開始reput
                for (int readSize = 0; readSize < result.getSize() && doNext; ) {
                    // 3. 檢查消息屬性，並構建一個消息的dispatchRequest
                    DispatchRequest dispatchRequest =
                        DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
                    if (dispatchRequest.isSuccess()) {
                        if (size > 0) {
                            // 4.消息分發，寫consumeQueue和Index
                            DefaultMessageStore.this.doDispatch(dispatchRequest);
                            // 設置reputOffset加上當前消息大小
                            this.reputFromOffset += size;
                            // 設置讀取的大小加上當前消息大小
                            readSize += size;
                             //如果size=0，說明讀取到瞭MappedFile的文件結尾
                        } else if (size == 0) {
                            // 5. 獲取下個文件的起始offset
                            this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);
                            // 設置readSize=0，結束循環
                            readSize = result.getSize();
                        }
                    } else if (!dispatchRequest.isSuccess()) {
                       // ...
                }
            } finally {
                result.release();
            }
        } else {
            doNext = false;
        }
    }
}

public void doDispatch(DispatchRequest req) {
    for (CommitLogDispatcher dispatcher : this.dispatcherList) {
        dispatcher.dispatch(req);
    }
}

// org.apache.rocketmq.store.DefaultMessageStore#putMessagePositionInfo
public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
    // 找到ConsumeQueue，如果找不到會創建一個ConsumeQueue
    ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId()); 
    // 消息保存到consumeQueue中
    cq.putMessagePositionInfoWrapper(dispatchRequest, checkMultiDispatchQueue(dispatchRequest));
}

// org.apache.rocketmq.store.ConsumeQueue#putMessagePositionInfo
private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,
        final long cqOffset) {
    this.byteBufferIndex.flip();
    // consumeQueue存儲單元的長度
    this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);
    // 消息物理偏移量
    this.byteBufferIndex.putLong(offset);
    // 消息長度
    this.byteBufferIndex.putInt(size);
    // 消息tags的哈希碼
    this.byteBufferIndex.putLong(tagsCode);
    final long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;
		// 獲取最後一個mappedFile
    MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
    if (mappedFile != null) {
        // 更新物理offset
        this.maxPhysicOffset = offset + size;
      	// 數據保存到consumeQueue
        return mappedFile.appendMessage(this.byteBufferIndex.array());
    }
    return false;
}

public void buildIndex(DispatchRequest req) {
    // 獲取或者創建最新索引文件，支持重試最多3次
    IndexFile indexFile = retryGetAndCreateIndexFile();
    if (indexFile != null) {
        // 獲取結束物理索引
        long endPhyOffset = indexFile.getEndPhyOffset();
        DispatchRequest msg = req;
        // 獲取topic和keys
        String topic = msg.getTopic();
        String keys = msg.getKeys();
        // 如果當前消息的commitLogOffset小於當前IndexFile的endPhyOffset時，說明當前消息已經構建過Index索引，因此直接返回
        if (msg.getCommitLogOffset() < endPhyOffset) {
            return;
        }
        // 獲取客戶端生成的uniqueId(msgId)，代表客戶端生成的唯一一條消息
        // 消息解密時生成的
        if (req.getUniqKey() != null) {
            indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
        }
        // 客戶端傳遞的keys，消息是從keys屬性中獲取的
        if (keys != null && keys.length() > 0) {
            String[] keyset = keys.split(MessageConst.KEY_SEPARATOR/*空格*/);
            for (int i = 0; i < keyset.length; i++) {
                String key = keyset[i];
                if (key.length() > 0) {
                    indexFile = putKey(indexFile, msg, buildKey(topic, key));
                    if (indexFile == null) {
                        return;
                    }
                }
            }
        }
    } else {
        log.error("build index error, stop building index");
    }
}

    public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        // 索引數量小於2000W，否則說明當前索引文件已經滿瞭，不能添加索引
        if (this.indexHeader.getIndexCount() < this.indexNum) {
            // keyHashCode
            int keyHash = indexKeyHashMethod(key);
            // 索引槽位置
            int slotPos = keyHash % this.hashSlotNum;
            // 絕對位置
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
            try {
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
                int absIndexPos =
                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize/*哈希槽數量*哈希槽大小=500w*4*/
                        + this.indexHeader.getIndexCount() * indexSize;
                // 更新IndexFile索引單元信息
              	// keyHash(4)+消息在commitLog中的偏移量(8)+消息存儲時間-索引文件開始存儲時間(4)+前置消息索引值(4)
                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);
								// 更新slots的indexCount
                this.mappedByteBuffer.putInt(absSlotPos/*hash槽的絕對位置*/, this.indexHeader.getIndexCount());
              	//...
                // 更新IndexHeader信息
                this.indexHeader.incIndexCount();
                this.indexHeader.setEndPhyOffset(phyOffset);
                this.indexHeader.setEndTimestamp(storeTimestamp);
                return true;
            } catch (Exception e) {
                log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);
            }
        } 
        return false;
    }