废话

很久没写过看过RocketMQ源码了，就刚开始用MQ的时候，顺便看了下源码，由于当时水平较差，写了几篇文章，现在感觉真JB乱....有种想重新写的冲动(￣▽￣) 所以从这篇文章开始估计会有好些内容是重复的

正文

了解过RocketMQ的原理或者看过我第一篇RocketMQ文章的应该知道ConsumeQueue是CommitLog的一个索引，查找消息的时候会先去ConsumeQueue获取offset，然后再去CommitLog拿消息，所以了解一下ConsumeQueue原理是非常必要的

先来看下ConsumeQueue几个重要的字段

    private final String topic;
    private final int queueId;//队列id
    private final ByteBuffer byteBufferIndex;// 写索引时用到的ByteBuffer
    private long maxPhysicOffset = -1;// 最后一个消息对应的物理Offset

ConsumeQueue的主要方法

先来看下其中的putMessagePostionInfo，这个是ConsumeQueue和CommitLog的offset建立关系的地方

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

        if (offset <= this.maxPhysicOffset) {
            return true;
        }
        //写入ConsumeQueue的临时ByteBuffer
        this.byteBufferIndex.flip();
        // 在第一篇文章或者网上介绍ConsumeQueue的结构的时候，我们知道ConsumeQueue有3部分组成
        // 知道了这个，看到下面4行代就知道分别对应什么值的
        this.byteBufferIndex.limit(CQStoreUnitSize);// ConsumeQueue的大小为20
        this.byteBufferIndex.putLong(offset);
        this.byteBufferIndex.putInt(size);
        this.byteBufferIndex.putLong(tagsCode);

        // 物理位移=消息数*每个consumequeue的大小
        final long expectLogicOffset = cqOffset * CQStoreUnitSize;
        //获取ConsumeQueue对应的MapedFile，没有则创建
        // 以前介绍过MapedFile是对文件的操作的封装，其对应一个磁盘上的ConsumeQueue文件
        MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile(expectLogicOffset);
        if (mapedFile != null) {
            if (mapedFile.isFirstCreateInQueue() && cqOffset != 0 && mapedFile.getWrotePostion() == 0) {
                this.minLogicOffset = expectLogicOffset;
                this.fillPreBlank(mapedFile, expectLogicOffset);
            }

            if (cqOffset != 0) {
                long currentLogicOffset = mapedFile.getWrotePostion() + mapedFile.getFileFromOffset();
                // 校验offset
                if (expectLogicOffset != currentLogicOffset) {
                    //....log
                }
            }
            // 每次构建ConsumeQueue的该值设置为CommitLog的offset
            this.maxPhysicOffset = offset;
            // 写入ConsumeQueue文件对应的ByteBuffer中
            return mapedFile.appendMessage(this.byteBufferIndex.array());
        }

        return false;
    }

该方法前面3个参数就不说了，就是对应ConsumeQueue的结构的3个组成部分，cqOffset是什么呢？
这个参数对应CommitLog里QUEUEOFFSET 这个组成，意思就是topic+queueId下的消息个数-1，所以，计算ConsumeQueue的物理位移的时候是=个数*大小

这个方法就是构建的核心，那么什么时候会调用这个方法呢

ReputMessageService定时构建ConsumeQueue

在Broker启动的时候，会启动一个线程ReputMessageService，那么入口就在run方法中

        @Override
        public void run() {
            while (!this.isStoped()) {
                try {
                    Thread.sleep(1);
                    this.doReput();
                } catch (Exception e) {
                    DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
        }

可以看到，每1毫秒执行一次doReput方法进行构建(当然也不是每次都会构建)，看下doReput方法

private void doReput() {
    for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {

        if (DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() //
                && this.reputFromOffset >= DefaultMessageStore.this.getConfirmOffset()) {
            break;
        }
        // 批量获取对应offset的数据
        SelectMapedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);
        if (result != null) {
            try {
                // 更新offset
                this.reputFromOffset = result.getStartOffset();

                for (int readSize = 0; readSize < result.getSize() && doNext; ) {
                    // 通过ByteBuffer读取消息并封装成DispatchRequest 返回
                    DispatchRequest dispatchRequest =
                            DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false);
                    int size = dispatchRequest.getMsgSize();//此次读取的消息大小

                    if (dispatchRequest.isSuccess()) {
                        if (size > 0) {
                            // 构建ConsumeQueue的地方
                            DefaultMessageStore.this.doDispatch(dispatchRequest);
                            // 这个是长轮询相关，后续会分析
                            if (BrokerRole.SLAVE != DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole()
                                    && DefaultMessageStore.this.brokerConfig.isLongPollingEnable()) {
                                DefaultMessageStore.this.messageArrivingListener.arriving(dispatchRequest.getTopic(),
                                        dispatchRequest.getQueueId(), dispatchRequest.getConsumeQueueOffset() + 1,
                                        dispatchRequest.getTagsCode());
                            }
                            // 构建ConsumeQueue完成之后，reputFromOffset应该增加，下次就从这后面继续构建
                            this.reputFromOffset += size;
                            readSize += size;
                            // ....
                        } else if (size == 0) {
                            this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);
                            readSize = result.getSize();
                        }
                    } else if (!dispatchRequest.isSuccess()) {
                        if (size > 0) {
                            this.reputFromOffset += size;
                        }
                        else {
                            doNext = false;
                            if (DefaultMessageStore.this.brokerConfig.getBrokerId() == MixAll.MASTER_ID) {
                                log.error("[BUG]the master dispatch message to consume queue error, COMMITLOG OFFSET: {}",
                                        this.reputFromOffset);

                                this.reputFromOffset += (result.getSize() - readSize);
                            }
                        }
                    }
                }
            } finally {
                result.release();
            }
        } else {
            doNext = false;
        }
    }
}

reputFromOffset：指的是开始解析物理队列的位置，当其小于物理队列的最大位置时isCommitLogAvailable方法返回true，这个值在初始化的时候为0

看下checkMessageAndReturnSize几个核心的地方，省略部分代码

    public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC, final boolean readBody) {
        try {
            // ....
            // 2 MAGIC CODE
            int magicCode = byteBuffer.getInt();
            switch (magicCode) {
                case MessageMagicCode:
                    break;
                case BlankMagicCode:
                    return new DispatchRequest(0, true /* success */);
                default:
                    log.warn("found a illegal magic code 0x" + Integer.toHexString(magicCode));
                    return new DispatchRequest(-1, false /* success */);
            }
            // ....
            short propertiesLength = byteBuffer.getShort();
            if (propertiesLength > 0) {
                byteBuffer.get(bytesContent, 0, propertiesLength);
                String properties = new String(bytesContent, 0, propertiesLength, MessageDecoder.CHARSET_UTF8);
                Map<String, String> propertiesMap = MessageDecoder.string2messageProperties(properties);

                // 延时消息将tagsCode设置为时间戳，为什么tagsCode要设置为时间戳后续讲延时消息的时候会分析
                {
                    String t = propertiesMap.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
                    if (ScheduleMessageService.SCHEDULE_TOPIC.equals(topic) && t != null) {
                        int delayLevel = Integer.parseInt(t);
                        if (delayLevel > 0) {
                            tagsCode = this.defaultMessageStore.getScheduleMessageService().computeDeliverTimestamp(delayLevel,
                                    storeTimestamp);
                        }
                    }
                }
            }
            // ....
            return new DispatchRequest(//
                topic, // 1
                queueId, // 2
                physicOffset, // 3
                totalSize, // 4
                tagsCode, // 5
                storeTimestamp, // 6
                queueOffset, // 7
                keys, // 8
                uniqKey, //9
                sysFlag, // 9
                preparedTransactionOffset// 10
            );
        } catch (Exception e) {
        }

        return new DispatchRequest(-1, false /* success */);
    }

重点看的就两个地方吧，一个是延迟消息tagsCode重置，注释已经写了，另一个是magicCode判断的地方，还记得BlankMagicCode是哪里设置的吗？知道消息存储相关的应该知道，当一个消息过来的时候，发现文件最后剩下的大小不够装载这条消息，那么就会在后面设置一条消息，消息大小为剩余的空间，magicCode为BlankMagicCode，证明已经到文件末尾了。这种时候DispatchRequest返回size是0，reputFromOffset重置

接下来就是重点doDispatch方法的实现了

    public void doDispatch(DispatchRequest req) {
        final int tranType = MessageSysFlag.getTransactionValue(req.getSysFlag());
        switch (tranType) {
            case MessageSysFlag.TransactionNotType:
            case MessageSysFlag.TransactionCommitType:
                DefaultMessageStore.this.putMessagePostionInfo(req.getTopic(), req.getQueueId(), req.getCommitLogOffset(), req.getMsgSize(),
                        req.getTagsCode(), req.getStoreTimestamp(), req.getConsumeQueueOffset());
                break;
            case MessageSysFlag.TransactionPreparedType:
            case MessageSysFlag.TransactionRollbackType:
                break;
        }

        if (DefaultMessageStore.this.getMessageStoreConfig().isMessageIndexEnable()) {
            DefaultMessageStore.this.indexService.buildIndex(req);
        }
    }

TransactionPreparedType和TransactionRollbackType应该是事务相关的，暂时没研究过，主要看下putMessagePostionInfo方法

    public void putMessagePostionInfo(String topic, int queueId, long offset, int size, long tagsCode, long storeTimestamp,long logicOffset) {
        ConsumeQueue cq = this.findConsumeQueue(topic, queueId);
        cq.putMessagePostionInfoWrapper(offset, size, tagsCode, storeTimestamp, logicOffset);
    }

这就是调用CQ的putMessagePostionInfoWrapper的地方了，logicOffset传入的是CommitLog的QueueOffset

到这里整个ConsumeQueue的构建过程就理清了