RocketMQ源码:broker文件清理

r囧r小猫 2024-03-24 13:42 10阅读 0赞

## 1. broker 清理文件介绍 ##

### 1.1 哪些文件需要清理 ###

首先我们需要介绍下在RocketMQ中哪些文件需要清理，其实可以想一想，在RocketMQ中哪些文件是一直在往里面写入东西的，最容易想到的就是`commitlog` 了，因为在一个broker 进程中，所有的普通消息，事务消息，系统消息啥的都往这个`commitlog`中写，随着时间的越来越长，然后`commitlog`就会越积攒越多，肯定会有磁盘放不下的那一天，而且我们消息消费完成后，那些被消费完成后的消息其实作用就很小了，可能会有这么一个场景，比如说我线上出现了某个问题，我想看下关于这个问题的消息有没有被消费到，可能你会用到这个消息，但是这种问题一般就是比较紧急的，最近实效的，之前那些消息其实作用就基本没有了，所以就需要清理掉之前的消息。其实不光`commitlog`需要清理，还需要清理一下`ConsumeQueue` 与`indexFile` ， 因为你`commitlog`里面的消息都被清理了，`ConsumeQueue` 与`indexFile` 再保存着之前的一些数据，就是纯粹浪费空间了。

所以说`broker` 文件清理主要是清理`commitlog ， ConsumeQueue ， indexFile`。

### 1.2 RocketMQ文件清理的机制 ###

我们介绍下`RocketMQ`文件清理的机制，`RocketMQ`默认是`清理72小时之前的消息`，然后它有几个触发条件， 默认是`凌晨4点触发清`理， 除非你你这个`磁盘空间占用到75%` 以上了。在清理`commitlog` 的时候，并不是一条消息一条消息的清理，拿到所有的`MappedFile`（抛去现在还在用着的，也就是最后一个） ，然后比对每个`MappedFile`的最后一条消息的时间，如果是`72小时之前`的就把`MappedFile`对应的文件删除了，销毁对应`MappedFile`，这种情况的话只要你`MappedFile` `最后一条消息还在存活实效内的话`，它就不会清理你这个`MappedFile`，就算你这个`MappedFile` 靠前的消息过期了。但是有一种情况它不管你消息超没超过72小时，直接就是删，那就是磁盘空间不足的时候，也就是占了`85%`以上了，就会立即清理。

清理完成`commitlog` 之后，就会拿到`commitlog`中最小的`offset` ，然后去`ConsumeQueue`与`indexFile`中把小于`offset` 的记录删除掉。清理`ConsumeQueue` 的时候也是遍历`MappedFile` ，然后它的最后一条消息（unit）小于`commitlog`中最小的`offset` 的话，就说明这个`MappedFile`都小于`offset` ，因为他们是顺序追加写的，这个MappedFile 就会清理掉，如果你`MappedFile` 最后一个unit不是小于`offset` 的话，这个`MappedFile` 就不删了。

## 2. 源码解析 ##

我们来看下源码是怎样实现的： 在broker 存储器`DefaultMessageStore` 启动（start）的时候，会添加几个任务调度，其中有一个就是文件清理的:

private void addScheduleTask() {
    
        // todo 清理过期文件 每隔10s
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                // todo
                DefaultMessageStore.this.cleanFilesPeriodically();
            }
        }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);
        ...
    }
    复制代码

默认是`10s`执行一次，可以看到它调用了`DefaultMessageStore` 的`cleanFilesPeriodically`方法：

private void cleanFilesPeriodically() {
        // todo 清除CommitLog文件
        this.cleanCommitLogService.run();
        // todo 清除ConsumeQueue文件
        this.cleanConsumeQueueService.run();
    }
    复制代码

### 2.1 清理commitlog ###

我们先来看下关于`commitlog`的清理工作:

public void run() {
        try {
            // todo 删除过期文件
            this.deleteExpiredFiles();
    
            this.redeleteHangedFile();
        } catch (Throwable e) {
            DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }
    复制代码

我们看下`deleteExpiredFiles` 方法的实现:

private void deleteExpiredFiles() {
        int deleteCount = 0;
        // 文件保留时间，如果超过了该时间，则认为是过期文件，可以被删除
        long fileReservedTime = DefaultMessageStore.this.getMessageStoreConfig().getFileReservedTime();
        // 删除物理文件的间隔时间，在一次清除过程中，可能需要被删除的文件不止一个，该值指定两次删除文件的间隔时间
        int deletePhysicFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteCommitLogFilesInterval();
        // 在清除过期文件时，如
        //果该文件被其他线程占用（引用次数大于0，比如读取消息），此时会
        //阻止此次删除任务，同时在第一次试图删除该文件时记录当前时间
        //戳，destroyMapedFileIntervalForcibly表示第一次拒绝删除之后能
        //保留文件的最大时间，在此时间内，同样可以被拒绝删除，超过该时
        //间后，会将引用次数设置为负数，文件将被强制删除
        int destroyMapedFileIntervalForcibly = DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly();
    
        // 指定删除文件的时间点，RocketMQ通过deleteWhen设置每天在
        //固定时间执行一次删除过期文件操作，默认凌晨4点
        boolean timeup = this.isTimeToDelete();
        // todo 检查磁盘空间是否充足，如果磁盘空间不充足，则返回true，表示应该触发过期文件删除操作
        boolean spacefull = this.isSpaceToDelete();
        // 预留手工触发机制，可以通过调用excuteDeleteFilesManualy
        //方法手工触发删除过期文件的操作，目前RocketMQ暂未封装手工触发
        //文件删除的命令
        boolean manualDelete = this.manualDeleteFileSeveralTimes > 0;
    
        if (timeup || spacefull || manualDelete) {
    
            if (manualDelete)
                this.manualDeleteFileSeveralTimes--;
    
            boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable() && this.cleanImmediately;
    
            log.info("begin to delete before {} hours file. timeup: {} spacefull: {} manualDeleteFileSeveralTimes: {} cleanAtOnce: {}",
                fileReserv