从源码角度聊聊RocketMQ 消息(文件)删除机制

傷城~ 2023-10-13 22:51 123阅读 0赞

## RocketMQ版本 ##

*  5.1.0

## 入口 ##

这里消息删除的源码入口其实不太好直接确定，我们可以通过启动`broker`的代码开始即`BrokerStartup`的`main`方法

![format_png][]

这里我们直接进入到`controller.start();`方法看看

里面的代码很长，我们可以看到有一个方法`startBasicService();`

![format_png 1][]

进入`startBasicService();`方法后我们可以看到有一个`this.messageStore.start();`

![format_png 2][]

这个`messageStore`就是封在了我们的一些消息文件的操作 可以简单看看他的一些方法

![format_png 3][]

由于方法众多，我们回到今天的主线，继续查看`start()`方法的实现类即`DefaultMessageStore`

![format_png 4][]

在`start`方法里面我们可以看到一个比较核心的方法`this.addScheduleTask();` 看着似乎是启动一些定时任务，没错，我们的消息过期删除就是通过定时任务扫描然后删除的

进入定时任务我们就能看到我们的文件删除定时任务了

![format_png 5][]

可以看到这个定时任务是broker启动后 60s后再启动，每次执行的时间间隔是可以配置的，默认10s

![format_png 6][]

实际的过期文件删除逻辑还是被封在再`DefaultMessageStore.this.cleanFilesPeriodically();`中

## cleanFilesPeriodically ##

我们进入cleanFilesPeriodically方法看看

csharp复制代码private void cleanFilesPeriodically() {
            this.cleanCommitLogService.run();
            this.cleanConsumeQueueService.run();
            this.correctLogicOffsetService.run();
        }

可以看到有三个清理任务

1.  清理CommitLog
2.  清理ConsumeQueue、index
3.  更新逻辑偏移量即ConsumeQueueStore

## cleanCommitLogService ##

这里我们先看看CommitLog是如何清理的

csharp复制代码
    public void run() {
                try {
                    this.deleteExpiredFiles();
                    this.reDeleteHangedFile();
                } catch (Throwable e) {
                    DefaultMessageStore.LOGGER.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

这里可以看到又分两步

1.  删除已经失效的文件
2.  再次删除第一次未删除掉的文件(比如文件被其他线程占用)

这里我们先看看deleteExpiredFiles方法

## deleteExpiredFiles ##

kotlin复制代码
    private void deleteExpiredFiles() {
                int deleteCount = 0;
                // 获取配置的过期时间 默认72小时 3天
                long fileReservedTime = DefaultMessageStore.this.getMessageStoreConfig().getFileReservedTime();
                // 获取删除文件的时间间隔 默认100毫秒
                int deletePhysicFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteCommitLogFilesInterval();
                // 1000 * 120 120s 强制删除文件映射
                int destroyMappedFileIntervalForcibly = DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly();         
                // 每次删除文件的最大数量 默认10
                int deleteFileBatchMax = DefaultMessageStore.this.getMessageStoreConfig().getDeleteFileBatchMax();
                //是否是删除时间 默认凌晨 4点(04) 04:00~04:59
                boolean isTimeUp = this.isTimeToDelete();
                // 磁盘是否已满
                boolean isUsageExceedsThreshold = this.isSpaceToDelete();
                boolean isManualDelete = this.manualDeleteFileSeveralTimes > 0;
    
                if (isTimeUp || isUsageExceedsThreshold || isManualDelete) {
    
                    if (isManualDelete) {
                        this.manualDeleteFileSeveralTimes--;
                    }
    
                    boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable() && this.cleanImmediately;
    
                    LOGGER.info("begin to delete before {} hours file. isTimeUp: {} isUsageExceedsThreshold: {} manualDeleteFileSeveralTimes: {} cleanAtOnce: {} deleteFileBatchMax: {}",
                        fileReservedTime,
                        isTimeUp,
                        isUsageExceedsThreshold,
                        manualDeleteFileSeveralTimes,
                        cleanAtOnce,
                        deleteFileBatchMax);
    
                    fileReservedTime *= 60 * 60 * 1000;
                    // 实际清理逻辑
                    deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(fileReservedTime, deletePhysicFilesInterval,
                        destroyMappedFileIntervalForcibly, cleanAtOnce, deleteFileBatchMax);
                    if (deleteCount > 0) {
                        // 如果是 controller模式，还要清理 EpochFile
                        // If in the controller mode, we should notify the AutoSwitchHaService to truncateEpochFile
                        if (DefaultMessageStore.this.brokerConfig.isEnableControllerMode()) {
                            if (DefaultMessageStore.this.haService instanceof AutoSwitchHAService) {
                                final long minPhyOffset = getMinPhyOffset();
                                ((AutoSwitchHAService) DefaultMessageStore.this.haService).truncateEpochFilePrefix(minPhyOffset - 1);
                            }
                        }
                    } else if (isUsageExceedsThreshold) {
                        LOGGER.warn("disk space will be full soon, but delete file failed.");
                    }
                }
            }

上面有一些代码都已经注释了，可以看到这里主要是做一些判断，是否徐要清理commitLog 有如下几个条件

1.  是否在配置的时间内,默认4点-5点
2.  是否磁盘已满
3.  是否手动调度，默认执行20次

3这里的条件我们可以在这里确定

![format_png 7][]

![format_png 8][]

可以看到这个参数主要是给外部控制台去使用的

三个条件任意满足一个就去执行删除逻辑

实际的文件删除逻辑是在方法

ini复制代码                deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(fileReservedTime, deletePhysicFilesInterval,
                        destroyMappedFileIntervalForcibly, cleanAtOnce, deleteFileBatchMax);

## commitLog.deleteExpiredFile ##

最终是调用的MappedFileQueue的deleteExpiredFileByTime方法

java复制代码
    public int deleteExpiredFileByTime(final long expiredTime,
            final int deleteFilesInterval,
            final long intervalForcibly,
            final boolean cleanImmediately,
            final int deleteFileBatchMax) {
            //获取映射文件列表 commitlog文件可能随时有写入，copy一份不影响写入
            Object[] mfs = this.copyMappedFiles(0);
    
            if (null == mfs)
                return 0;
    
            int mfsLength = mfs.length - 1;
            // 删除的文件数量
            int deleteCount = 0;
            // 需要删除的 MappedFile
            List<MappedFile> files = new ArrayList<>();
            int skipFileNum = 0;
            if (null != mfs) {
                //do check before deleting
                checkSelf();
                for (int i = 0; i < mfsLength; i++) {
                    MappedFile mappedFile = (MappedFile) mfs[i];
                    long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
                    // 当前时间大于等于 liveMaxTimestamp 或者 cleanImmediately 为 true(手动清理或磁盘空间达到85%)，表示文件过期或需要立即清理
                    if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) {
                        if (skipFileNum > 0) {
                            log.info("Delete CommitLog {} but skip {} files", mappedFile.getFileName(), skipFileNum);
                        }
                        // 真正的删除逻辑
                        if (mappedFile.destroy(intervalForcibly)) {
                            files.add(mappedFile);
                            deleteCount++;
    
                            if (files.size() >= deleteFileBatchMax) {
                                break;
                            }
    
                            if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
                                try {
                                    Thread.sleep(deleteFilesInterval);
                                } catch (InterruptedException e) {
                                }
                            }
                        } else {
                            break;
                        }
                    } else {
                        skipFileNum++;
                        //avoid deleting files in the middle
                        break;
                    }
                }
            }
            //从文件映射队列中删除对应的文件映射
            deleteExpiredFile(files);
            //返回删除的文件个数
            return deleteCount;
        }

上面的代码有一些注释，实际也是检测MappedFile是否需要删除，最终的删除逻辑在 mappedFile.destroy(intervalForcibly)方法

我们进去看看

kotlin复制代码@Override
        public boolean destroy(final long intervalForcibly) {
            // 释放内存
            this.shutdown(intervalForcibly);
    
            if (this.isCleanupOver()) {
                try {
                    long lastModified = getLastModifiedTimestamp();
                    // 关闭文件通道
                    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)
                        + "," + (System.currentTimeMillis() - lastModified));
                } 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;
        }

## cleanConsumeQueueService 清理queue ##

这里我们直接查看deleteExpiredFiles方法，也是从最外层的this.cleanConsumeQueueService.run();方法进入的

kotlin复制代码private void deleteExpiredFiles() {
                int deleteLogicsFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteConsumeQueueFilesInterval();
                // 获取commitlog 最小的偏移量
                long minOffset = DefaultMessageStore.this.commitLog.getMinOffset();
                // 如果最小物理偏移量大于上次记录的最小物理偏移量 说明commitLog被删除过了
                if (minOffset > this.lastPhysicalMinOffset) {
                    this.lastPhysicalMinOffset = minOffset;
                     // 获取消费队列表
                    ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.getConsumeQueueTable();
    
                    for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
                        for (ConsumeQueueInterface logic : maps.values()) {
                            //删除过期文件，并返回删除的文件数量
                            int deleteCount = DefaultMessageStore.this.consumeQueueStore.deleteExpiredFile(logic, minOffset);
                            if (deleteCount > 0 && deleteLogicsFilesInterval > 0) {
                                try {
                                    //当上一个ConsumeQueue成功删除之后，下一个ConsumeQueue删除需要等待0.1s
                                    Thread.sleep(deleteLogicsFilesInterval);
                                } catch (InterruptedException ignored) {
                                // 忽略中断异常
                                }
                            }
                        }
                    }
    
                    DefaultMessageStore.this.indexService.deleteExpiredFile(minOffset);
                }
            }

这里最终的删除逻辑还是在deleteExpiredFile中封在的

kotlin复制代码DefaultMessageStore.this.consumeQueueStore.deleteExpiredFile(logic, minOffset);

实际和上面commitLog删除类似，这里我们不过多深究MappedByteBuffer这种偏底层的细节。主要是了解主流程，后面再单独聊聊一些有深度的细节

## idnex删除 ##

实际我们在queue里面删除逻辑中会看到这个么一个方法

kotlin复制代码DefaultMessageStore.this.indexService.deleteExpiredFile(minOffset);

![format_png 9][]

![get_code_OWU1YzhhZDUzOTY5NzA0N2ZlMDUwZjc2YzhjOTE2YzYsMTY5MDE2NTU0MzQwMQ][]

没错这里面就是删除索引index的文件

csharp复制代码public void deleteExpiredFile(long offset) {
            Object[] files = null;
            try {
                // 获取读锁
                this.readWriteLock.readLock().lock();
                // 如果索引文件列表为空，则直接返回
                if (this.indexFileList.isEmpty()) {
                    return;
                }
                // 获取第一个索引文件的结束物理偏移量
                long endPhyOffset = this.indexFileList.get(0).getEndPhyOffset();    
                // 如果第一个索引文件的结束物理偏移量小于commitLog的最小偏移量，说明索引无效需要删除
                if (endPhyOffset < offset) {
                    // 将索引文件列表转换为数组
                    files = this.indexFileList.toArray();
                }
            } catch (Exception e) {
                LOGGER.error("destroy exception", e);
            } finally {
                this.readWriteLock.readLock().unlock();
            }
    
            if (files != null) {
                List<IndexFile> fileList = new ArrayList<>();
                for (int i = 0; i < (files.length - 1); i++) {
                    IndexFile f = (IndexFile) files[i];
                    // 如果文件的结束物理偏移量小于commitlog的最小偏移量
                    if (f.getEndPhyOffset() < offset) {
                        fileList.add(f);
                    } else {
                        break;
                    }
                }
                // 调用删除过期文件的方法
                this.deleteExpiredFile(fileList);
            }
        }

可以看到索引的删除文件也是通过commitlog的最小偏移量来删除的。只要一个索引中的最大偏移量小于commitlog的最小偏移量，说明这个索引是不需要的就可以删除

## correctLogicOffsetService ##

最后一个是更新逻辑偏移量，即consumeQueueStore

scss复制代码private void correctLogicMinOffset() {
    
                long lastForeCorrectTimeCurRun = lastForceCorrectTime;
                // 获取commitlog的最小偏移量
                long minPhyOffset = getMinPhyOffset();
                ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.getConsumeQueueTable();
                for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
                    for (ConsumeQueueInterface logic : maps.values()) {
                        if (Objects.equals(CQType.SimpleCQ, logic.getCQType())) {
                            // cq is not supported for now.
                            continue;
                        }
                        // 如果需要纠正
                        if (needCorrect(logic, minPhyOffset, lastForeCorrectTimeCurRun)) {
                            // 执行纠正操作
                            doCorrect(logic, minPhyOffset);
                        }
                    }
                }
            }

*  doCorrect

csharp复制代码private void doCorrect(ConsumeQueueInterface logic, long minPhyOffset) {
                // 删除逻辑队列中过期的文件
                DefaultMessageStore.this.consumeQueueStore.deleteExpiredFile(logic, minPhyOffset);
                int sleepIntervalWhenCorrectMinOffset = DefaultMessageStore.this.getMessageStoreConfig().getCorrectLogicMinOffsetSleepInterval();
                if (sleepIntervalWhenCorrectMinOffset > 0) {
                    try {
                        Thread.sleep(sleepIntervalWhenCorrectMinOffset);
                    } catch (InterruptedException ignored) {
                    }
                }
            }

## 总结 ##

总的来说RocketMQ执行文件清理操作主要是在设置的凌晨4点(4:00-5:00),要清理的文件也是我们常见的CommitLog、consumequeue、Index. 清理的条件就是三个

1.  磁盘超过设置的默认85%
2.  到达设置的清理时间默认凌晨4点
3.  手动执行清理操作

三个条件满足一个即可

至于一些文件操作比如MappedByteBuffer等不在本次探讨范围内，限于篇幅我们这次是看整理的逻辑。后续会基于单个点深入分析

[format_png]: https://img-blog.csdnimg.cn/img_convert/382ae1532fc26659fbb48ccc3c30da1c.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/97f09049d9b43684e80498c747fd9ed8.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/7823b08e8abf8e5880f8c4b3dcb50017.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/23567769dc4a41600a6f641fc75e019c.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/6903f80fbb2e5c745511c6aa2650997d.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/b7dfc0d8978329620d433b4c3b1359c9.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/dd98b1c61d57801b5440cbe30e055e19.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/5e7235ed4c69ffa227812e202a2ff09c.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/d259ed19134cfe0b4b60c3a7250d86df.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/c318c78d6c4c8b4bc64a3b408cc6271f.webp?x-oss-process=image/format,png
[get_code_OWU1YzhhZDUzOTY5NzA0N2ZlMDUwZjc2YzhjOTE2YzYsMTY5MDE2NTU0MzQwMQ]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=OWU1YzhhZDUzOTY5NzA0N2ZlMDUwZjc2YzhjOTE2YzYsMTY5MDE2NTU0MzQwMQ==