rocketmq的broker恢复commit-log的时候如何恢复consumeQueueindexfile

Posted 2020-11-24 notlate

如果一个broker正常退出，是会删除abort文件的。那么启动broker的时候发现abort文件还存在，那么说明上次是异常终止，会进入到commit-log的recoverAbnormally逻辑里面，因为所有其他的信息都是从commit-log获取到的，所以追根溯源只能从commit-log开始着手。

    public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
        // recover by the minimum time stamp
        boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
        final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            // Looking beginning to recover from which file
            int index = mappedFiles.size() - 1;
            MappedFile mappedFile = null;
            for (; index >= 0; index--) {
                mappedFile = mappedFiles.get(index);
                if (this.isMappedFileMatchedRecover(mappedFile)) {
                    log.info("recover from this mapped file " + mappedFile.getFileName());
                    break;
                }
            }

　从最新的mapped文件开始找，直到发现一个有效的mappedfile，对这个mappedfile文件执行操作：

　　　　　　　　ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();

　　　　　　　　while (true) {
                DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                int size = dispatchRequest.getMsgSize();

                if (dispatchRequest.isSuccess()) {
                    // Normal data
                    if (size > 0) {
                        mappedFileOffset += size;

                        if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {
                            if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {
                                this.defaultMessageStore.doDispatch(dispatchRequest);
                            }
                        } else {
                            this.defaultMessageStore.doDispatch(dispatchRequest);
                        }
                    }

　　checkMessageAndReturnSize的逻辑：

    public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC,
        final boolean readBody) {
        try {
            // 1 TOTAL SIZE
            int totalSize = byteBuffer.getInt();

　　通过不变遍历这个最新的mappedfile的bytebuffer，把他的每一条消息取出来，并且消息一个字段就是total-size，累计求和就可以拿到这个mappedfile上次挂掉的时候到底写到哪了。对于其他的mappedfile默认认为都是写满的。

 

            processOffset += mappedFileOffset;
            this.mappedFileQueue.setFlushedWhere(processOffset);
            this.mappedFileQueue.setCommittedWhere(processOffset);
            this.mappedFileQueue.truncateDirtyFiles(processOffset);

　　

这里已经把commit-log完全恢复了。接下来就是恢复consumequeue、indexfile了。前一节提到生成消息的时候，其实也会同步更新consumeQueue，并且刷盘。

也就是不管是故障恢复还是正常生成消息，都会涉及commit-log、consumeQueue、indexfile的更新、落盘。只不过后面两个是异步的。

他们都是通过this.defaultMessageStore.doDispatch(dispatchRequest)更新的

 

另外topicQueueTable完全可以通过consumeQueue得到，因为前者是后者子集：

 

    private void recoverTopicQueueTable() {
        HashMap<String/* topic-queueid */, Long/* offset */> table = new HashMap<String, Long>(1024);
        long minPhyOffset = this.commitLog.getMinOffset();
        for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueue logic : maps.values()) {
                String key = logic.getTopic() + "-" + logic.getQueueId();
                table.put(key, logic.getMaxOffsetInQueue());
                logic.correctMinOffset(minPhyOffset);
            }
        }

        this.commitLog.setTopicQueueTable(table);
    }