1.绪论
rocketmq的broker中关于消息持久化的组件主要包含三个，分别是：持久化消息到文件中的组件commitLog；根据消息key索引commitLog日志的indexFile；消费者根据topic和queueId查询commitLog日志的consumeQueue。前面已经介绍commitLog和indexFile，这节我们就来学习一下consumeQueue。
2. consumeQueue
4.1.1 consumeQueue的中每条消息的组成
消费者在消息的时候，只知道它在哪个topic的哪个queue里面消费哪个tags的拿条消息，那我们怎么由此定位到一条消息呢？所以我们需要为commitLog建立一条索引。
其实每个topic+queueId都会建立一个ConsumeQueue文件，而这个映射关系存储在broker中consumeQueueTable中，我们查询消息的时候，过consumeQueueTable根据queueId和topic快速定位到我们需要的ConsumeQueue，然后我们再根据消费者所提交的consumeQueueOffset*每条consumequeue索引的大小便能找到我们所以要的consume索引文件的位置，再根据里面存储的commitLog的物理偏移量便能在commitLog中定位到具体的消息的位置。
commitLog的存储结构可以如下所示：
└── TopicTest
├── 0
└── 00000000000000000000
├── 1
└── 00000000000000000000
├── 2
└── 00000000000000000000
└── 3
└── 00000000000000000000
可以通过下图来说明，consumequeue是如何组成的，并且和commitLog的关系：
3.consumeQueue中消息的创建和获取
consumequeue其实底层和commitLog是一样的，其实由多个mappedFile来构成的，这里我们就不在讨论consumequeue的具体存储逻辑。有兴趣的小伙伴可以看这篇文章：
深度解析RocketMq源码-持久化组件（四） CommitLog_rocketmq一主一备commitlog-CSDN博客
接下来我们主要看一看consumequeue是什么时候创建的，并且在消费者知道它需要消费的topic和queueId过后，如何找到它具体要消费的哪条消息的，这其实也是rocketmq的核心之一。
3.1. consumequeue建立消息
其实consumequeue是一个采用mappedFile持久化数据的组件，它写入数据其实发分两步：
1.根据topic和queueId在topic和queueid与consumequeue的映射表（consumeQueueTable）中找到Consumequeue。
2.构造一条consumequeue记录，包括8字节的8字节的commitLog的offset + 4字节的消息大小+8自己的tagsCode，然后顺序写入到consumequeue中。
3.1.1 根据topic和queueId找到需要写入的consumequeue
public void putMessagePositionInfo(DispatchRequest dispatchRequest) {//根据topic和queueId找到对应的consumequeueConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());//根据consumequeue顺序写入consumemequeue的索引数据cq.putMessagePositionInfoWrapper(dispatchRequest, checkMultiDispatchQueue(dispatchRequest));} //根据queueId和topic检索到这个topic和queueId是属于哪个topic的public ConsumeQueue findConsumeQueue(String topic, int queueId) {//根据consumeQueueTable和queueId获取到具体的ConsumeQueueConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);if (null == map) {ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);if (oldMap != null) {map = oldMap;} else {map = newMap;}}ConsumeQueue logic = map.get(queueId);if (null == logic) {//如果consumeQueue为空，便新建一个ConsumequeueConsumeQueue newLogic = new ConsumeQueue(topic,queueId,StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),this);//并且设置到consumeQueueTable中ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);if (oldLogic != null) {logic = oldLogic;} else {if (MixAll.isLmq(topic)) {lmqConsumeQueueNum.getAndIncrement();}logic = newLogic;}}return logic;}3.1.2 向consumequeue中写入一条记录
//commitLog的真正组成8字节的commitLog的offset + 4字节的消息大小+8自己的tagsCodeprivate boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,final long cqOffset) {if (offset + size <= this.maxPhysicOffset) {log.warn("Maybe try to build consume queue repeatedly maxPhysicOffset={} phyOffset={}", maxPhysicOffset, offset);return true;}this.byteBufferIndex.flip();//consumqueue的每条数据占20个字节this.byteBufferIndex.limit(CQ_STORE_UNIT_SIZE);//8字节的commotLog的物理偏移量this.byteBufferIndex.putLong(offset);//4字节消息大小this.byteBufferIndex.putInt(size);//8字节的tagsCodethis.byteBufferIndex.putLong(tagsCode);//写入到哪个offsetfinal long expectLogicOffset = cqOffset * CQ_STORE_UNIT_SIZE;//获取到offset的最后一条数据MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(expectLogicOffset);if (mappedFile != null) {if (mappedFile.isFirstCreateInQueue() && cqOffset != 0 && mappedFile.getWrotePosition() == 0) {//将跟新mappedFilde1写指针为expectLogicOffsetthis.minLogicOffset = expectLogicOffset;this.mappedFileQueue.setFlushedWhere(expectLogicOffset);this.mappedFileQueue.setCommittedWhere(expectLogicOffset);this.fillPreBlank(mappedFile, expectLogicOffset);log.info("fill pre blank space " + mappedFile.getFileName() + " " + expectLogicOffset + " "+ mappedFile.getWrotePosition());}if (cqOffset != 0) {//long currentLogicOffset = mappedFile.getWrotePosition() + mappedFile.getFileFromOffset();if (expectLogicOffset < currentLogicOffset) {log.warn("Build consume queue repeatedly, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",expectLogicOffset, currentLogicOffset, this.topic, this.queueId, expectLogicOffset - currentLogicOffset);return true;}if (expectLogicOffset != currentLogicOffset) {LOG_ERROR.warn("[BUG]logic queue order maybe wrong, expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",expectLogicOffset,currentLogicOffset,this.topic,this.queueId,expectLogicOffset - currentLogicOffset);}}//实际的物理地址为offset+size大小this.maxPhysicOffset = offset + size;//将数据写入到bytebuffer中return mappedFile.appendMessage(this.byteBufferIndex.array());}return false;}3.2 根据topic和messagequeue检索消息
先根据topic和queueId找到以及consumeoffset查询一条消息分三步：
1.根据topic和queueId查询到对应的consumequeue；
2.根据consumeoffset在consumequeue中找到一条consumequeue的记录，里面包含一个属性就是实际消息在commitLog中的物理偏移量和大小;
3.根据物理偏移量和消息大小在commitLog中获取到实际消息内容。
//根据queueId和topic检索到这个topic和queueId是属于哪个topic的public ConsumeQueue findConsumeQueue(String topic, int queueId) {//根据consumeQueueTable和queueId获取到具体的ConsumeQueueConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);if (null == map) {ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);if (oldMap != null) {map = oldMap;} else {map = newMap;}}ConsumeQueue logic = map.get(queueId);if (null == logic) {//如果consumeQueue为空，便新建一个ConsumequeueConsumeQueue newLogic = new ConsumeQueue(topic,queueId,StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),this);//并且设置到consumeQueueTable中ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);if (oldLogic != null) {logic = oldLogic;} else {if (MixAll.isLmq(topic)) {lmqConsumeQueueNum.getAndIncrement();}logic = newLogic;}}return logic;}再根据consumeoffset在consumequeue中获取到具体consumequeue的索引数据。
public boolean get(final long address, final CqExtUnit cqExtUnit) {if (!isExtAddr(address)) {return false;}final int mappedFileSize = this.mappedFileSize;final long realOffset = unDecorate(address);MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(realOffset, realOffset == 0);if (mappedFile == null) {return false;}int pos = (int) (realOffset % mappedFileSize);SelectMappedBufferResult bufferResult = mappedFile.selectMappedBuffer(pos);if (bufferResult == null) {log.warn("[BUG] Consume queue extend unit({}) is not found!", realOffset);return false;}boolean ret = false;try {ret = cqExtUnit.read(bufferResult.getByteBuffer());} finally {bufferResult.release();}return ret;}4.总结
至此，我们已经大概了解消息在进入到broker过后做了什么。在生产者推送消息到broker过后，为了保证数据的能够快速的持久化，是直接按照到达顺序写入到commitLog中的，然后就会给主线程返回生产消息成功的通知。但是消费者需要根据topic和queueId获取到一条消息，并且需要根据消息的key检索一条消息。为了满足上述两个需求，rocketmq会启动一个线程，扫描commitLog，如果有新的消息写入，便会构建IndexFile和consumequeue两个文件，其实相当于两个索引文件。这一步骤在我们后面章节会详细介绍。
其实先持久化文件，然后启动线程对消息做其他处理，这一思想的本质就是为了增大吞吐量。在其他框架中也会应用到这种思想，比如elasticsearch中，在写消息的时候，会同时写入到transLog和memory buffer中后便会返回成功，后续单独启动线程根据memory buffer中的数据来进行其他操作，比如分词，建立倒排索引等，可以看出translog其实就类似于rokcetmq的commitLog。所以万变不离其中，只要有一份持久化数据过后，便可以跟客户端返回成功了，然后再单独的启动线程根据这份持久化数据做定制化处理。
