文章目录

• 消息队列与索引文件恢复
• • 加载Commitlog文件
  • 加载ConsumeQueue
  • 加载索引文件
  • 根据Broker是否是正常停止执行不同的恢复策略
  • • 恢复所有ConsumeQueue文件
    • 正常停止commitLog文件恢复
    • 异常停止commitLog文件恢复
• 存储流程
• • 校验msg
  • 异步存储消息(默认)
  • 同步存储消息
  • • CommitLog刷盘
    • • 同步刷盘
      • 异步刷盘
• 实时更新消息消费队列与索引文件
• • 根据消息更新ConumeQueue
  • • 异步创建文件
    • 同步创建文件 (默认)
  • 根据消息更新Index 索引文件
  • • 获取或创建IndexFile 文件
    • 添加到Hash索引文件中
• 过期文件删除机制
• 总结

消息队列与索引文件恢复

假设broker上次启动后,由于某些原因宕机了,导致ommitlog 、ConsumeQueue 、IndexFile 文件数据不一致,所以broker初次启动时,会去加载文件,然后尝试故障修复,让数据达到最终一致性.

加载存储文件

public boolean load() {
        boolean result = true;
        try {
            //判断上一次退出是否正常。
            /*
            其实现机制是Broker 在启动时创建$｛ ROCKET_HOME} /store/abort 文件，
            在退出时通过注册NM 钩子函数删除abort 文件。
            如果下一次启动时存在abort文件。说明Broker 是异常退出的，
             Commitlog 与Consumequeue数据有可能不一致，需要进行修复。
             */
            boolean lastExitOK = !this.isTempFileExist();
            log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally");
            //加载延迟队列， RocketMQ 定时消息相关
            if (null != scheduleMessageService) {
                result = result && this.scheduleMessageService.load();
            }
            // load Commit Log 加载Commitlog 文件
            result = result && this.commitLog.load();
            // load Consume Queue 加载消息消费队列
            result = result && this.loadConsumeQueue();
            if (result) {
                //加载存储检测点，检测点主要记录commitlog 文件、Consumequeue 文件、
                //Index 索引文件的刷盘点，将在下文的文件刷盘机制中再次提交。
                this.storeCheckpoint =
                    new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
                //加载索引文件
                this.indexService.load(lastExitOK);
                //根据Broker 是否是正常停止执行不同的恢复策略
                this.recover(lastExitOK);
                log.info("load over, and the max phy offset = {}", this.getMaxPhyOffset());
            }
        } catch (Exception e) {
            log.error("load exception", e);
            result = false;
        }
        //如果没能加载成功
        if (!result) {
            //关闭分配请求
            this.allocateMappedFileService.shutdown();
        }
        return result;
    }

判断上次退出是否正常

private boolean isTempFileExist() {
        String fileName = StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir());
        File file = new File(fileName);
        return file.exists();
    }

也就是这个文件

加载Commitlog文件

public boolean load() {
        boolean result = this.mappedFileQueue.load();
        log.info("load commit log " + (result ? "OK" : "Failed"));
        return result;
    }
public boolean load() {
        //加载$ ｛ ROCKET_HOME }/store/commitlog 目录下所有文件
        File dir = new File(this.storePath);
        File[] files = dir.listFiles();
        if (files != null) {
            // ascending order
            //按照文件名排序
            Arrays.sort(files);
            for (File file : files) {
                //如果文件大小与配置文件的单个文件大小不一致，将忽略该目录下所有文件，
                if (file.length() != this.mappedFileSize) {
                    log.warn(file + "\t" + file.length()
                        + " length not matched message store config value, please check it manually");
                    return false;
                }
                try {
                    //创建MappedFile 对象
                    MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);
                    //将wrotePosition 、flushedPosition ,
                    //committedPosition 三个指针都设置为文件大小。
                    mappedFile.setWrotePosition(this.mappedFileSize);
                    mappedFile.setFlushedPosition(this.mappedFileSize);
                    mappedFile.setCommittedPosition(this.mappedFileSize);
                    this.mappedFiles.add(mappedFile);
                    log.info("load " + file.getPath() + " OK");
                } catch (IOException e) {
                    log.error("load file " + file + " error", e);
                    return false;
                }
            }
        }
        return true;
    }

加载ConsumeQueue

private boolean loadConsumeQueue() {
        File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
        File[] fileTopicList = dirLogic.listFiles();
        if (fileTopicList != null) {
            //遍历消息消费队列根目录
            for (File fileTopic : fileTopicList) {
                //获取目录名
                String topic = fileTopic.getName();
                //加载每个消息消费队列下的文件，
                File[] fileQueueIdList = fileTopic.listFiles();
                if (fileQueueIdList != null) {
                    for (File fileQueueId : fileQueueIdList) {
                        int queueId;
                        try {
                            //获取文件名,就是queueId
                            queueId = Integer.parseInt(fileQueueId.getName());
                        } catch (NumberFormatException e) {
                            continue;
                        }
                        //构建ConsumeQueue 对象
                        //初始化ConsumeQueue 的topic 、queueld 、storePath 、mappedFileSize 属性
                        ConsumeQueue logic = new ConsumeQueue(
                            topic,
                            queueId,
                            StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
                            this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),
                            this);
                        //存入consumeQueueTable
                        this.putConsumeQueue(topic, queueId, logic);
                          //加载ConsumeQueue
                        if (!logic.load()) {
                            return false;
                        }
                    }
                }
            }
        }
        log.info("load logics queue all over, OK");
        return true;
    }

加载索引文件

IndexService

//加载索引文件
    public boolean load(final boolean lastExitOK) {
        File dir = new File(this.storePath);
        File[] files = dir.listFiles();
        if (files != null) {
            // ascending order
            Arrays.sort(files);
            for (File file : files) {
                try {
                    IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
                    //加载索引文件
                    f.load();
                    //如果上次异常退出
                    if (!lastExitOK) {
                        //且索引文件上次刷盘时间小于该索引文件最大的消息时间戳该文件将立即销毁。
                        if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint()
                            .getIndexMsgTimestamp()) {
                            f.destroy(0);
                            continue;
                        }
                    }
                    log.info("load index file OK, " + f.getFileName());
                    //加入索引文件集合
                    this.indexFileList.add(f);
                } catch (IOException e) {
                    log.error("load file {} error", file, e);
                    return false;
                } catch (NumberFormatException e) {
                    log.error("load file {} error", file, e);
                }
            }
        }
        return true;
    }

加载索引头部信息
IndexFile

public void load() {
        this.indexHeader.load();
    }

IndexHeader

public void load() {
        //分别解析文件头部信息
        this.beginTimestamp.set(byteBuffer.getLong(beginTimestampIndex));
        this.endTimestamp.set(byteBuffer.getLong(endTimestampIndex));
        this.beginPhyOffset.set(byteBuffer.getLong(beginPhyoffsetIndex));
        this.endPhyOffset.set(byteBuffer.getLong(endPhyoffsetIndex));
        this.hashSlotCount.set(byteBuffer.getInt(hashSlotcountIndex));
        this.indexCount.set(byteBuffer.getInt(indexCountIndex));
        if (this.indexCount.get() <= 0) {
            this.indexCount.set(1);
        }
    }

根据Broker是否是正常停止执行不同的恢复策略

DefaultMessageStore

private void recover(final boolean lastExitOK) {
      //恢复所有ConsumeQueue文件,返回最大的在ConsumeQueue存储的comlog偏移量
        long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();
        if (lastExitOK) {
            //正常停止commitLog文件恢复
            this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue);
        } else {
            //异常停止commitLog文件恢复
            this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue);
        }
        this.recoverTopicQueueTable();
    }

恢复所有ConsumeQueue文件

返回最大的在ConsumeQueue存储的comlog偏移量

//恢复所有ConsumeQueue文件
    private long recoverConsumeQueue() {
        long maxPhysicOffset = -1;
        for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
            //遍历每个ConsumeQueue
            for (ConsumeQueue logic : maps.values()) {
                //恢复
                logic.recover();
                if (logic.getMaxPhysicOffset() > maxPhysicOffset) {
                    maxPhysicOffset = logic.getMaxPhysicOffset();
                }
            }
        }
        //返回最大的在ConsumeQueue存储的comlog偏移量
        return maxPhysicOffset;
    }

恢复ConsumeQueue文件
ConsumeQueue

public void recover() {
        final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            //从倒数第三个文件开始进行恢复
            int index = mappedFiles.size() - 3;
            //不足3 个文件，则从第一个文件开始恢复。
            if (index < 0)
                index = 0;
            int mappedFileSizeLogics = this.mappedFileSize;
            MappedFile mappedFile = mappedFiles.get(index);
            ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
            //获取该文件的初始偏移量
            long processOffset = mappedFile.getFileFromOffset();
            //当前文件已校验通过的offset
            long mappedFileOffset = 0;
            long maxExtAddr = 1;
            while (true) {
                for (int i = 0; i < mappedFileSizeLogics; i += CQ_STORE_UNIT_SIZE) {
                    //每次获取一条消息
                    long offset = byteBuffer.getLong();
                    int size = byteBuffer.getInt();
                    long tagsCode = byteBuffer.getLong();
                    if (offset >= 0 && size > 0) {
                        mappedFileOffset = i + CQ_STORE_UNIT_SIZE;
                        //重新记录消息体总长度加上消息在comlog偏移量
                        this.maxPhysicOffset = offset + size;
                        if (isExtAddr(tagsCode)) {
                            maxExtAddr = tagsCode;
                        }
                    } else {
                        //说明遍历到没数据了
                        log.info("recover current consume queue file over,  " + mappedFile.getFileName() + " "
                            + offset + " " + size + " " + tagsCode);
                        break;
                    }
                }
                //说明恢复到了结尾
                if (mappedFileOffset == mappedFileSizeLogics) {
                    //重新计算下个遍历文件索引
                    index++;
                    if (index >= mappedFiles.size()) {
                    //遍历到最后一个,则结束遍历
                        log.info("recover last consume queue file over, last mapped file "
                            + mappedFile.getFileName());
                        break;
                    } else {
                        //恢复下一个文件
                        mappedFile = mappedFiles.get(index);
                        byteBuffer = mappedFile.sliceByteBuffer();
                        processOffset = mappedFile.getFileFromOffset();
                        mappedFileOffset = 0;
                        log.info("recover next consume queue file, " + mappedFile.getFileName());
                    }
                } else {
                    //恢复消息队列结束
                    log.info("recover current consume queue queue over " + mappedFile.getFileName() + " "
                        + (processOffset + mappedFileOffset));
                    break;
                }
            }
            //记录该文件的恢复的物理偏移量
            processOffset += mappedFileOffset;
            //设置刷盘指针
            this.mappedFileQueue.setFlushedWhere(processOffset);
            //当前数据提交指针
            this.mappedFileQueue.setCommittedWhere(processOffset);
            //删除offset 之后的所有文件
            this.mappedFileQueue.truncateDirtyFiles(processOffset);
            if (isExtReadEnable()) {
                this.consumeQueueExt.recover();
                log.info("Truncate consume queue extend file by max {}", maxExtAddr);
                this.consumeQueueExt.truncateByMaxAddress(maxExtAddr);
            }
        }
    }

删除offset 之后的所有文件
MappedFileQueue

//删除offset 之后的所有文件
    public void truncateDirtyFiles(long offset) {
        List<MappedFile> willRemoveFiles = new ArrayList<MappedFile>();
        //遍历目录下文件
        for (MappedFile file : this.mappedFiles) {
            //获取文件尾部偏移
            long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;
            //若文件尾部偏移>offset
            if (fileTailOffset > offset) {
                //文件开始偏移>=offset
                //说明当前文件包含了有效偏移
                if (offset >= file.getFileFromOffset()) {
                    //分别设置当前文件的刷盘,提交,写入指针.
                    file.setWrotePosition((int) (offset % this.mappedFileSize));
                    file.setCommittedPosition((int) (offset % this.mappedFileSize));
                    file.setFlushedPosition((int) (offset % this.mappedFileSize));
                } else {
                    //说明该文件是有效文件后面创建的
                    //释放MappedFile 占用的内存资源（内存映射与内存通道等）
                    file.destroy(1000);
                    //加入待删除集合
                    willRemoveFiles.add(file);
                }
            }
        }
        //删除文件
        this.deleteExpiredFile(willRemoveFiles);
    }

MappedFile文件销毁
MappedFile

public boolean destroy(final long intervalForcibly) {
        //关闭MappedFile
        this.shutdown(intervalForcibly);
        //判断是否清理完成
        if (this.isCleanupOver()) {
            try {
                //关闭通道
                this.fileChannel.close();
                log.info("close file channel " + this.fileName + " OK");
                long beginTime = System.currentTimeMillis();
                //删除整个物理文件
                boolean result = this.file.delete();
                log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
                    + (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePosition() + " M:"
                    + this.getFlushedPosition() + ", "
                    + UtilAll.computeElapsedTimeMilliseconds(beginTime));
            } catch (Exception e) {
                log.warn("close file channel " + this.fileName + " Failed. ", e);
            }
            return true;
        } else {
            log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
                + " Failed. cleanupOver: " + this.cleanupOver);
        }
        return false;
    }

关闭MappedFile
ReferenceResource

public void shutdown(final long intervalForcibly) {
        //默认true
        if (this.available) {
            //初次调用时available 为true ，设置available为fal se
            this.available = false;
            //设置初次关闭的时间戳
            this.firstShutdownTimestamp = System.currentTimeMillis();
            //释放资源,引用次数小于1 的情况下才会释放资源
            this.release();
        } else if (this.getRefCount() > 0) {
            //如果引用次数大于0
            //对比当前时间与firstShutdownTimestamp ，如果已经超过了其最大拒绝存活期，每执行
            //一次，将引用数减少1000 ，直到引用数小于0 时通过执行release方法释放资源。
            if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
                this.refCount.set(-1000 - this.getRefCount());
                this.release();
            }
        }
    }

释放引用和资源
ReferenceResource

public void release() {
        //引用减1
        long value = this.refCount.decrementAndGet();
        if (value > 0)
            return;
        synchronized (this) {
            //释放堆外内存
            this.cleanupOver = this.cleanup(value);
        }
    }

释放堆外内存
MappedFile

public boolean cleanup(final long currentRef) {
        //如果available为true ，表示MappedFile当前可用，无须清理，
        if (this.isAvailable()) {
            log.error("this file[REF:" + currentRef + "] " + this.fileName
                + " have not shutdown, stop unmapping.");
            return false;
        }
        //如果资源已经被清除，返回true
        if (this.isCleanupOver()) {
            log.error("this file[REF:" + currentRef + "] " + this.fileName
                + " have cleanup, do not do it again.");
            return true;
        }
        //如果是堆外内存，调用堆外内存的cleanup 方法清除
        clean(this.mappedByteBuffer);
        //维护虚拟内存
        TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(this.fileSize * (-1));
        //对象个数-1
        TOTAL_MAPPED_FILES.decrementAndGet();
        log.info("unmap file[REF:" + currentRef + "] " + this.fileName + " OK");
        return true;
    }

删除过期文件
MappedFileQueue

void deleteExpiredFile(List<MappedFile> files) {
        if (!files.isEmpty()) {
            Iterator<MappedFile> iterator = files.iterator();
            while (iterator.hasNext()) {
                MappedFile cur = iterator.next();
                //mappedFiles,不包含,就跳过.
                if (!this.mappedFiles.contains(cur)) {
                    iterator.remove();
                    log.info("This mappedFile {} is not contained by mappedFiles, so skip it.", cur.getFileName());
                }
            }
            try {
                //从mappedFiles删除所有
                if (!this.mappedFiles.removeAll(files)) {
                    log.error("deleteExpiredFile remove failed.");
                }
            } catch (Exception e) {
                log.error("deleteExpiredFile has exception.", e);
            }
        }
    }

正常停止commitLog文件恢复

CommitLog

public void recoverNormally(long maxPhyOffsetOfConsumeQueue) {
        //在进行文件恢复时查找消息时是否验证CRC
        boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
        final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            // Began to recover from the last third file
            //从倒数第三个文件开始进行恢复
            int index = mappedFiles.size() - 3;
            //如果不足3个文件，则从第一个文件开始恢复。
            if (index < 0)
                index = 0;
            MappedFile mappedFile = mappedFiles.get(index);
            ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
            //Commitlog 文件已确认的物理偏移量,等于mappedFile.getFileFromOffset 加上mappedFileOffset 。
            long processOffset = mappedFile.getFileFromOffset();
            //当前文件已校验通过的offset ,
            long mappedFileOffset = 0;
            while (true) {
                DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                //取出一条消息
                int size = dispatchRequest.getMsgSize();
                // Normal data
                //查找结果为true 并且消息的长度大于0 表示消息正确，
                if (dispatchRequest.isSuccess() && size > 0) {
                    //mappedFileOffset 指针向前移动本条消息的长度
                    mappedFileOffset += size;
                }
                // Come the end of the file, switch to the next file Since the
                // return 0 representatives met last hole,
                // this can not be included in truncate offset
                else if (dispatchRequest.isSuccess() && size == 0) {
                    //如果查找结果为true 并且消息的长度等于0 ，表示已到该文件的末尾
                    index++;
                    //没有文件则退出
                    if (index >= mappedFiles.size()) {
                        // Current branch can not happen
                        log.info("recover last 3 physics file over, last mapped file " + mappedFile.getFileName());
                        break;
                    } else {
                        //若还有文件
                        mappedFile = mappedFiles.get(index);
                        //重置变量,继续遍历下个文件,并重新进入循环
                        byteBuffer = mappedFile.sliceByteBuffer();
                        processOffset = mappedFile.getFileFromOffset();
                        mappedFileOffset = 0;
                        log.info("recover next physics file, " + mappedFile.getFileName());
                    }
                }
                // Intermediate file read error
                else if (!dispatchRequest.isSuccess()) {
                    //读取消息错误,直接结束
                    log.info("recover physics file end, " + mappedFile.getFileName());
                    break;
                }
            }
            processOffset += mappedFileOffset;
            this.mappedFileQueue.setFlushedWhere(processOffset);
            this.mappedFileQueue.setCommittedWhere(processOffset);
            //删除offset 之后的所有文件
            this.mappedFileQueue.truncateDirtyFiles(processOffset);
            // Clear ConsumeQueue redundant data
            //queue记录的最大commitlog偏移若大于commitlog存储的最大偏移
            if (maxPhyOffsetOfConsumeQueue >= processOffset) {
                log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
                //删除processOffset之后存储的ConsumeQueue脏数据文件
                this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
            }
        } else {
            // Commitlog case files are deleted
            log.warn("The commitlog files are deleted, and delete the consume queue files");
            //Commitlog文件不存在
            this.mappedFileQueue.setFlushedWhere(0);
            this.mappedFileQueue.setCommittedWhere(0);
            //销毁所有ConsumeQueue文件
            this.defaultMessageStore.destroyLogics();
        }
    }

删除processOffset之后存储的ConsumeQueue脏数据文件

public void truncateDirtyLogicFiles(long phyOffset) {
        ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> tables = DefaultMessageStore.this.consumeQueueTable;
        for (ConcurrentMap<Integer, ConsumeQueue> maps : tables.values()) {
            //遍历每个队列目录
            for (ConsumeQueue logic : maps.values()) {
                //每个目录执行删除
                logic.truncateDirtyLogicFiles(phyOffset);
            }
        }
    }

截断phyOffet之后的文件
ConsumeQueue

public void truncateDirtyLogicFiles(long phyOffet) {
        //获取逻辑文件大小
        int logicFileSize = this.mappedFileSize;
        //设置commitlog最大偏移
        this.maxPhysicOffset = phyOffet;
        long maxExtAddr = 1;
        while (true) {
            //获取最后一个文件
            MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
            if (mappedFile != null) {
                ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
                //清空刷盘,写入,提交位置
                mappedFile.setWrotePosition(0);
                mappedFile.setCommittedPosition(0);
                mappedFile.setFlushedPosition(0);
                for (int i = 0; i < logicFileSize; i += CQ_STORE_UNIT_SIZE) {
                    //获取每条消息
                    long offset = byteBuffer.getLong();
                    int size = byteBuffer.getInt();
                    long tagsCode = byteBuffer.getLong();
                    //说明为该文件第一条消息
                    if (0 == i) {
                        //该文件记录第一条的commitlog偏移>=phyOffet
                        //则说明该文件记录消息无效
                        if (offset >= phyOffet) {
                            //删除文件
                            this.mappedFileQueue.deleteLastMappedFile();
                            break;
                        } else {
                            //继续遍历下条消息
                            int pos = i + CQ_STORE_UNIT_SIZE;
                            //重新设置刷盘,写入,提交位置
                            mappedFile.setWrotePosition(pos);
                            mappedFile.setCommittedPosition(pos);
                            mappedFile.setFlushedPosition(pos);
                            //设置刷入commitlog偏移
                            this.maxPhysicOffset = offset + size;
                            // This maybe not take effect, when not every consume queue has extend file.
                            if (isExtAddr(tagsCode)) {
                                maxExtAddr = tagsCode;
                            }
                        }
                    } else {
                        //不是第一条消息的处理
                        //说明消息有效
                        if (offset >= 0 && size > 0) {
                            if (offset >= phyOffet) {
                                //这里直接返回,而不是删除文件
                                //是因为该文件记录的之前消息是有效的
                                //为什么不清空后面的消息了,这个采用后续消息覆盖解决
                                //而为了保证消息刷盘,写入,以及提交的位置正确,在前一次执行消息解析的时候
                                //已经存储了
                                return;
                            }
                            //继续遍历下条消息
                            int pos = i + CQ_STORE_UNIT_SIZE;
                            //重新设置刷盘,写入,提交位置
                            mappedFile.setWrotePosition(pos);
                            mappedFile.setCommittedPosition(pos);
                            mappedFile.setFlushedPosition(pos);
                            //设置刷入commitlog偏移
                            this.maxPhysicOffset = offset + size;
                            if (isExtAddr(tagsCode)) {
                                maxExtAddr = tagsCode;
                            }
                            //遍历到最后一条消息,则返回
                            if (pos == logicFileSize) {
                                return;
                            }
                        } else {
                            //遍历到文件无效消息,则返回
                            return;
                        }
                    }
                }
            } else {
                //没有文件,直接退出
                break;
            }
        }
        if (isExtReadEnable()) {
            this.consumeQueueExt.truncateByMaxAddress(maxExtAddr);
        }
    }

销毁所有ConsumeQueue文件
DefaultMessageStore

public void destroyLogics() {
        for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
            //遍历每一个ConsumeQueue目录
            for (ConsumeQueue logic : maps.values()) {
                //销毁每个目录所有ConsumeQueue文件
                logic.destroy();
            }
        }
    }

ConsumeQueue

public void destroy() {
        this.maxPhysicOffset = -1;
        this.minLogicOffset = 0;
        //将消息消费队列目录下的所有文件全部删除。
        this.mappedFileQueue.destroy();
        if (isExtReadEnable()) {
            this.consumeQueueExt.destroy();
        }
    }
public void destroy() {
        //遍历目录下每个consumeQueue文件
        for (MappedFile mf : this.mappedFiles) {
            //销毁通道,物理文件
            mf.destroy(1000 * 3);
        }
        //清除集合
        this.mappedFiles.clear();
        this.flushedWhere = 0;
        // delete parent directory
        //删除上级目录
        File file = new File(storePath);
        if (file.isDirectory()) {
            file.delete();
        }
    }

异常停止commitLog文件恢复

CommitLog

//Broker 异常停止文件恢复
    @Deprecated
    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;
                }
            }
            //遍历到最后一个文件都没找到,则遍历最后一个文件消息
            if (index < 0) {
                index = 0;
                mappedFile = mappedFiles.get(index);
            }
            ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
            long processOffset = mappedFile.getFileFromOffset();
            long mappedFileOffset = 0;
            //遍历消息
            while (true) {
                //从result 返回的ByteBuffer 中循环读取消息，一次读取一条，
                //反序列化并创建DispatchRequest对象,主要记录一条消息数据
                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()) {
                        //消息物理偏移量>CommitLog的提交指针,则结束
                            if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {
                                //调用文件转发请求,分别同步index和queue文件
                                this.defaultMessageStore.doDispatch(dispatchRequest);
                            }
                        } else {
                            this.defaultMessageStore.doDispatch(dispatchRequest);
                        }
                    }
                    // Come the end of the file, switch to the next file
                    // Since the return 0 representatives met last hole, this can
                    // not be included in truncate offset
                    else if (size == 0) {
                        //无效数据
                        //遍历下一个文件
                        index++;
                        //遍历到了最后一个文件,直接退出
                        if (index >= mappedFiles.size()) {
                            // The current branch under normal circumstances should
                            // not happen
                            log.info("recover physics file over, last mapped file " + mappedFile.getFileName());
                            break;
                        } else {
                            mappedFile = mappedFiles.get(index);
                            byteBuffer = mappedFile.sliceByteBuffer();
                            processOffset = mappedFile.getFileFromOffset();
                            mappedFileOffset = 0;
                            log.info("recover next physics file, " + mappedFile.getFileName());
                        }
                    }
                } else {
                    //解析数据不成功,直接结束
                    log.info("recover physics file end, " + mappedFile.getFileName() + " pos=" + byteBuffer.position());
                    break;
                }
            }
            processOffset += mappedFileOffset;
            this.mappedFileQueue.setFlushedWhere(processOffset);
            this.mappedFileQueue.setCommittedWhere(processOffset);
            //删除offset 之后的所有文件
            this.mappedFileQueue.truncateDirtyFiles(processOffset);
            // Clear ConsumeQueue redundant data
            if (maxPhyOffsetOfConsumeQueue >= processOffset) {
                log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
                this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
            }
        }
        // Commitlog case files are deleted
        else {
            log.warn("The commitlog files are deleted, and delete the consume queue files");
            //尚未找到文件
            //设置commitlog 目录的flushedWhere 、committedWhere指针都为0
            this.mappedFileQueue.setFlushedWhere(0);
            this.mappedFileQueue.setCommittedWhere(0);
            //销毁消息消费队列文件。
            this.defaultMessageStore.destroyLogics();
        }
    }

判断一个消息文件是一个正确的文件
CommitLog

//判断一个消息文件是一个正确的文件
    private boolean isMappedFileMatchedRecover(final MappedFile mappedFile) {
        ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
        //通过魔数判断该文件是否符合commitlog 消息文件的存储格式。
        int magicCode = byteBuffer.getInt(MessageDecoder.MESSAGE_MAGIC_CODE_POSTION);
        if (magicCode != MESSAGE_MAGIC_CODE) {
            return false;
        }
        int sysFlag = byteBuffer.getInt(MessageDecoder.SYSFLAG_POSITION);
        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
        int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + bornhostLength;
        //若存储时间为0,说明该消息存储文件中未存储任何消息。
        long storeTimestamp = byteBuffer.getLong(msgStoreTimePos);
        if (0 == storeTimestamp) {
            return false;
        }
        //如果messagelndexEnable 为true ， 表示索引文件的刷盘时间点也参与计算。
        if (this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable()
            && this.defaultMessageStore.getMessageStoreConfig().isMessageIndexSafe()) {
            if (storeTimestamp <= this.defaultMessageStore.getStoreCheckpoint().getMinTimestampIndex()) {
                log.info("find check timestamp, {} {}",
                    storeTimestamp,
                    UtilAll.timeMillisToHumanString(storeTimestamp));
                return true;
            }
        } else {
            //文件第一条消息的时间戳小于文件检测点说明该文件部分消息是可靠的，
            if (storeTimestamp <= this.defaultMessageStore.getStoreCheckpoint().getMinTimestamp()) {
                log.info("find check timestamp, {} {}",
                    storeTimestamp,
                    UtilAll.timeMillisToHumanString(storeTimestamp));
                return true;
            }
        }
        return false;
    }

索引文件的刷盘时间点也参与计算
StoreCheckpoint

public long getMinTimestampIndex() {
        //分别取三者最小值,文件刷盘时间点  消息消费队列文件刷盘时间点  索引文件刷盘时间点
        return Math.min(this.getMinTimestamp(), this.indexMsgTimestamp);
    }

1 ) CommitLog ：消息存储文件，所有消息主题的消息都存储在CommitLog 文件中。
2 ) ConsumeQueue ：消息消费队列，消息到达CommitLog 文件后，将异步转发到消息
消费队列，供消息消费者消费。
3 ) IndexFile ：消息索引文件，主要存储消息Key 与Offset 的对应关系。
4 ）事务状态服务： 存储每条消息的事务状态。
5 ）定时消息服务：每一个延迟级别对应一个消息消费队列，存储延迟队列的消息拉取
进度。

RocketMQ 将所有主题的消息存储在同－个文件中，确保消息发送时顺序写文件

RocketMQ 引入了ConsumeQueue 消息队列文件，每个消息主题包含多个消息消费队列，每一个消息队列有一个消息文件。

IndexFile 索引文件，其主要设计理念就是为了加速消息的检索性能，根据消息的属性快速从Commitlog 文件中检索消息。

存储流程

SendMessageProcessor

private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,
                                                                SendMessageContext mqtraceContext,
                                                                SendMessageRequestHeader requestHeader) {
        //校验topic,以及创建topic配置文件
        final RemotingCommand response = preSend(ctx, request, requestHeader);
        final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();
        if (response.getCode() != -1) {
            return CompletableFuture.completedFuture(response);
        }
        final byte[] body = request.getBody();
        //获取topic配置
        int queueIdInt = requestHeader.getQueueId();
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
        if (queueIdInt < 0) {
            queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());
        }
        //创建消息扩展,主要封装一些其他参数
        MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
        msgInner.setTopic(requestHeader.getTopic());
        msgInner.setQueueId(queueIdInt);
        if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {
            return CompletableFuture.completedFuture(response);
        }
        msgInner.setBody(body);
        msgInner.setFlag(requestHeader.getFlag());
        MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
        msgInner.setPropertiesString(requestHeader.getProperties());
        msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
        msgInner.setBornHost(ctx.channel().remoteAddress());
        msgInner.setStoreHost(this.getStoreHost());
        msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
        String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
        CompletableFuture<PutMessageResult> putMessageResult = null;
        Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
        String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
        if (transFlag != null && Boolean.parseBoolean(transFlag)) {
            if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
                response.setCode(ResponseCode.NO_PERMISSION);
                response.setRemark(
                        "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                                + "] sending transaction message is forbidden");
                return CompletableFuture.completedFuture(response);
            }
            putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);
        } else {
            //异步存储消息
            putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
        }
        return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);
    }

校验msg

SendMessageProcessor

private RemotingCommand preSend(ChannelHandlerContext ctx, RemotingCommand request,
                                    SendMessageRequestHeader requestHeader) {
        final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
        response.setOpaque(request.getOpaque());
        response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
        response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
        log.debug("Receive SendMessage request command {}", request);
        final long startTimestamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
        if (this.brokerController.getMessageStore().now() < startTimestamp) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimestamp)));
            return response;
        }
        response.setCode(-1);
        //校验消息
        super.msgCheck(ctx, requestHeader, response);
        if (response.getCode() != -1) {
            return response;
        }
        return response;
    }

AbstractSendMessageProcessor

protected RemotingCommand msgCheck(final ChannelHandlerContext ctx,
        final SendMessageRequestHeader requestHeader, final RemotingCommand response) {
        //检查该Broker 是否有写权限。
        if (!PermName.isWriteable(this.brokerController.getBrokerConfig().getBrokerPermission())
            && this.brokerController.getTopicConfigManager().isOrderTopic(requestHeader.getTopic())) {
            response.setCode(ResponseCode.NO_PERMISSION);
            response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                + "] sending message is forbidden");
            return response;
        }
        //验证topic长度以及命名格式
        if (!TopicValidator.validateTopic(requestHeader.getTopic(), response)) {
            return response;
        }
        //检测该topic是否可以发送消息,默认主题不能发送,仅仅供路由查找。
        if (TopicValidator.isNotAllowedSendTopic(requestHeader.getTopic(), response)) {
            return response;
        }
        //获取该topic对应topicConfig
        TopicConfig topicConfig =
            this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
        if (null == topicConfig) {
            int topicSysFlag = 0;
            if (requestHeader.isUnitMode()) {
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
                } else {
                    topicSysFlag = TopicSysFlag.buildSysFlag(true, false);
                }
            }
            log.warn("the topic {} not exist, producer: {}", requestHeader.getTopic(), ctx.channel().remoteAddress());
            //创建topicConfig
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(
                requestHeader.getTopic(),
                requestHeader.getDefaultTopic(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                requestHeader.getDefaultTopicQueueNums(), topicSysFlag);
            if (null == topicConfig) {
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    topicConfig =
                        this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
                            requestHeader.getTopic(), 1, PermName.PERM_WRITE | PermName.PERM_READ,
                            topicSysFlag);
                }
            }
            if (null == topicConfig) {
                response.setCode(ResponseCode.TOPIC_NOT_EXIST);
                response.setRemark("topic[" + requestHeader.getTopic() + "] not exist, apply first please!"
                    + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
                return response;
            }
        }
        int queueIdInt = requestHeader.getQueueId();
        int idValid = Math.max(topicConfig.getWriteQueueNums(), topicConfig.getReadQueueNums());
        if (queueIdInt >= idValid) {
            String errorInfo = String.format("request queueId[%d] is illegal, %s Producer: %s",
                queueIdInt,
                topicConfig.toString(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            log.warn(errorInfo);
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(errorInfo);
            return response;
        }
        return response;
    }