RocketMQ中ConsumeQueue文件与Index文件是怎么样的

//Broker启动初始化@1 BrokerStartup#mainstart(createBrokerController(args))boolean initResult = controller.initialize()@2 BrokerController#initializethis.messageStore = new DefaultMessageStore@3 DefaultMessageStore#DefaultMessageStore()this.reputMessageService = new ReputMessageService();this.dispatcherList = new LinkedList<>();this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());//存储服务启动@4 DefaultMessageStore#start()//允许重复转发reputFromOffset设置为CommitLog的提交指针if (this.getMessageStoreConfig().isDuplicationEnable()) {this.reputMessageService.setReputFromOffset(this.commitLog.getConfirmOffset());} else {//不允许重复转发reputFromOffset设置为CommitLog内存中最大偏移量 this.reputMessageService.setReputFromOffset(this.commitLog.getMaxOffset();}this.reputMessageService.start();

this.dispatcherList = new LinkedList<>();//构建ConsumeQueuethis.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());//构建Indexthis.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());public void doDispatch(DispatchRequest req) {for (CommitLogDispatcher dispatcher : this.dispatcherList) {dispatcher.dispatch(req);}}

@1 DefaultMessageStore#putMessagePositionInfo@2 ConsumeQueue#putMessagePositionInfoWrapper

private void fillPreBlank(final MappedFile mappedFile, final long untilWhere) {ByteBuffer byteBuffer = ByteBuffer.allocate(CQ_STORE_UNIT_SIZE);byteBuffer.putLong(0L);byteBuffer.putInt(Integer.MAX_VALUE);byteBuffer.putLong(0L);int until = (int) (untilWhere % this.mappedFileQueue.getMappedFileSize());for (int i = 0; i < until; i += CQ_STORE_UNIT_SIZE) {mappedFile.appendMessage(byteBuffer.array());}}

this.byteBufferIndex.flip();this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE); //限定每个条目大小this.byteBufferIndex.putLong(offset); //写入消息偏移量this.byteBufferIndex.putInt(size); //写入消息长度this.byteBufferIndex.putLong(tagsCode); //写入tag hashcode

this.dispatcherList = new LinkedList<>();//构建ConsumeQueuethis.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());//构建Indexthis.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());public void doDispatch(DispatchRequest req) {for (CommitLogDispatcher dispatcher : this.dispatcherList) {dispatcher.dispatch(req);}}//IndexService初始化public IndexService(final DefaultMessageStore store) {this.defaultMessageStore = store;this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum();this.indexNum = store.getMessageStoreConfig().getMaxIndexNum();this.storePath =StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir());}

@1 DefaultMessageStore#CommitLogDispatcherBuildIndex#dispatch@2 IdexService#buildIndex

/计算索引数据需要放在哪个位置int absIndexPos =IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize+ this.indexHeader.getIndexCount() * indexSize;//将hashcode存储在MappedByteBuffer中this.mappedByteBuffer.putInt(absIndexPos, keyHash);//将物理偏移量存储在MappedByteBuffer中this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);//落地时间-当前索引的起始时间差值写入MappedByteBufferthis.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);//记录前一条hash桶对应的值（Index条目下标）；注意此处用于解决Hash冲突this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);//将当前index中包含的条目数量存入到Hash槽中，将覆盖原先的值this.mappedByteBuffer.putInt(absSlotPos,this.indexHeader.getIndexCount());