开发者

RocketMQ源码分析之Broker过期消息清理机制

开发者 https://www.devze.com 2023-05-11 10:23 出处:网络 作者: 林师傅
目录前言Broker消息清理机制简介Broker与消息清理相关配置消息清理机制源码分析CommitLog清理源码分析ConsumeQueue和IndexFile清理源码分析ConsumeQueue文件清理IndexFile清理总结前言
目录
  • 前言
  • Broker消息清理机制简介
    • Broker与消息清理相关配置
  • 消息清理机制源码分析
    • CommitLog清理源码分析
    • ConsumeQueue和IndexFile清理源码分析
    • ConsumeQueue文件清理
    • IndexFile清理
  • 总结

    前言

    前面文章讲了消息是如何保存的以及consumeQueue与Index文件更新机制。随着消息的增加,Broker不可能一直保存所有消息,Broker是按照什么规则清理消息的呢?被消费过后的消息就会被清理掉吗?下面我们来介绍Broker消息清理机制。

    Broker消息清理机制简介

    消息是被顺序存储在CommitLog文件中的,且消息长度不定长,因此消息的清理不是以消息为单位进行的,而是以CommitLog为单位进行的。默认情况下,Broker会清理单个CommitLog文件中最后一条消息超过72小时的CommitLog文件,除了用户手动清理为,下面几种情况会被默认清理。

    • CommitLog清理机制

      • CommitLog文件过期(72小时),且达到清理时间点(默认为04:00~05:00),自动清理过期的CommitLog文件

      • CommitLog文件过期(72小时),且CommitLog所在磁盘分区占用率已经达到过期清理警戒线(默认75%),无论是否到达清理时间点都会自动清理过期文件

      • CommitLog所在磁盘分区占用率已经达到清理警戒线(默认85%),无论是否过期,都会从最早的文件开始清理,一次最多清理10个文件

      • CommitLog所在磁盘分区占用率已经达到系统危险警戒线(默认90%),Broker将拒绝消息写入

      • Broker至少会保留最新的CommitLog文件

    • ConsumeQueue清理机制

      • 如果ConsumeQueue文件关联CommitLog都被清理,则清理此ConsumeQueue文件
      • Broker每个Topic-QueueId至少会保留最新的文件
    • IndexFile清理机制

      • 如果IndexFile所有索引单元关联CommitLog都被清理,则清理此IndexFile

    Broker与消息清理相关配置

    # 文件自动清理时间,单位H,默认72
    fileReservedTime=72
    # CommitLog物理文件删除间隔,但是ms,默认100
    deleteCommitLogFilesInterval = 100
    # 文件自动清理时间,默认04,即凌晨4点
    deleteWhen = "04"
    # 硬盘占用率所在分区过期清理警戒线,超过这个值,无论是否到达清理时间,都会自动清理过期文件
    diskMaxUsedSpaceRatio = 75

    消息清理机制源码分析

    消息定时清理的是由DefaultMessageStore类负责的,它在启动时(start)会调用DefaultMessageStore#addScheduleTask添加和消息存储相关的定时任务,其中就包括消息删除相关的定时任务DefaultMessageStore.this.cleanFilesPeriodically(),这个定时任务在Broker启动后60s开始,每隔10秒执行一次

    // org.apache.rocketmq.store.DefaultMessageStore#addScheduleTask
    private void addScheduleTask() {
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                // commitLog、consandroidumeQueue和IndexFile定时删除
                DefaultMessageStore.this.cleanFilesPeriodically();
            }
        }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval()/*10s*/, TimeUnit.MILLISECONDS);
      	// ...
    }
    

    在cleanFilesPeriodically()中有两个方法,cleanCommitLogService.run()负责清理CommitLog,cleanConsumeQueueService.run()负责清理ConsumeQueue和IndexFile。

    // org.a开发者_开发入门pache.rocketmq.store.DefaultMessageStore#cleanFilesPeriodically
    private void cleanFilesPeriodically() {
        // 清理CommitLog
        this.cleanCommitLogService.run();
        // 清理ConsumeQueue和IndexFile
        this.cleanConsumeQueueService.run();
    }
    

    CommitLog清理源码分析

    CommitLog清理方法CleanCommitLogService#run调用了CleanCommitLogService#deleteExpiredFiles,deleteExpiredFiles方法的核心代码逻辑如下,以下三种情况会触发CommitLog文件的删除

    • 当前时间是凌晨4点
    • CommitLog所在磁盘分区硬盘占用率超过75%
    • 手动删除CommitLog
    // org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles
    private void deleteExpiredFiles() {
        // 是否是凌晨4点,用小时匹配[04:00,05:00)
        boolean timeup = this.isTimeToDelete();
        // >75%就会返回true,如果大于85%,则触发强制删除
        boolean spacefull = this.isSpaceToDelete();
        // 手动删除次数是否>0
        boolean manualDelete = this.manualDeleteFileSeveralTimes > 0;
        if (timeup/*凌晨4点*/ || spacefull/*空间满了*/ || manualDelete/*手动删除*/) {
            boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable()/*默认true*/ && this.cleanImmediately/*空间占用超过85%,触发强制删除*/;
    				// 删除CommitLog
            deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(fileReservedTime, deletePhysicFilesInterval,
                destroyMapedFileIntervalForcibly, cleanAtOnce);
            if (deleteCount > 0) {
            } else if (spacefull) {
                log.warn("disk space will be full soon, but delete file failed.");
            }
        }
    }
    

    CommitLog的清理逻辑在MappedFileQueue#deleteExpiredFileByTime,其核心代码如下所示,主要分为下面几个步骤

    • 复制MappedFileQueue中的mappedFiles,循环处理删除逻辑,循环mfsLength-1次,也就是无论如何都会保留最新的MappedFile
    • 如果MappedFile的最后修改时间超过72小时或CommitLog所在磁盘分区硬盘占用率超过85%触发强制删除MappedFile,则会删除MappedFile,每次删除最多删除10个MappedFile,相邻MappedFile删除时间间隔默认100ms
    • 删除MappedFileQueue的mappedFiles数组中已删除的MappedFile,并返回删除MappedFile的数量
    // org.apache.rocketmq.store.MappedFileQueue#deleteExpiredFileByTime
    public int deleteExpiredFileByTime(final long expiredTime,final int deleteFilesInterval,final long intervalForcibly,final boolean cleanImmediately) {
        // 复制一份当前mappedFile
        Object[] mfs = this.copyMappedFiles(0);
    		// 会保留最后一个MappedFile
        int mfsLength = mfs.length - 1;
        int deleteCount = 0;
        List<MappedFile> files = new ArrayList<MappedFile>();
        if (null != mfs) {
            for (int i = 0; i < mfsLength; i++) {
                MappedFile mappedFile = (MappedFile) mfs[i];
                // 最后修改时间+过期时间
                long liveMax编程Timestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
                // 如果commitLog所在磁盘分区总容量超过85%,触发立即删除,或者超过了72小时的mappedFile
                if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) {
                    // 删除mappedFile
                    if (mappedFile.destroy(intervalForcibly)) {
                        files.add(mappedFile);
                        deleteCount++;
                        // 一次最多删除10个mappedFile
                        if (files.size() >= DELETE_FILES_BATCH_MAX/*10*/) {
                            break;
                        }
                        if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
                          php  try {
                                // 删除文件时间间隔,默认100ms
                                Thread.sleep(deleteFilesInterval);
                            } catch (InterruptedException e) {
                            }
                        }
                    } else {
                        break;
                    }
                } else {
                    //avoid deleting files in the middle
                    break;
                }
            }
        }
        // 从MappedFileQueue的mappedFiles中删除这个mappedFile
        deleteExpiredFile(files);
        return deleteCount;
    }
    

    ConsumeQueue和IndexFile清理源码分析

    ConsumeQueue和IndexFile清理方法CleanConsumeQueueService#run调用了CleanConsumeQueueService#deleteExpiredFiles方法清理ConsumeQueue和IndexFile。CleanConsumeQueueService#deleteExpiredFiles核心代码如下,包括两个主要逻辑

    • 遍历consumeQueueTable中的ConsumeQueue,调用ConsumeQueue#deleteExpiredFile删除过期ConsumeQueue
    • 调用IndexService#deleteExpiredFile删除过期IndexFile
    // org.apache.rocketmq.store.DefaultMessageStore.CleanConsumeQueueService#deleteExpiredFiles
    private void deleteExpiredFiles() {
        if (minOffset > this.lastPhysicalMinOffset) {
            ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> tables = DefaultMessageStore.this.consumeQueueTable;
    				// 遍历ConsumeQueue
            for (ConcurrentMap<Integer, ConsumeQueue> maps : tables.values()) {
                for (ConsumeQueue logic : maps.values()) {
                    // 删除consumeQueue
                    int deleteCount = logic.deleteExpiredFile(minOffset);
    								// ... 间隔100ms
                }
            }
    				// 删除indexFile
            DefaultMessageStore.this.indexService.deleteExpiredFile(minOffset);
        }
    }

    ConsumeQupythoneue文件清理

    ConsumeQueue文件底层也是MappedFile,清理ConsumeQueue调用MappedFileQueue#deleteExpiredFileByOffset清理ConsumeQueue的过期MappedFile,源码如下,核心逻辑

    • 复制MappedFileQueue中的mappedFiles,循环处理清理逻辑,循环mfsLength-1次,也就是无论如何都会保留最新的MappedFile
    • 如果ConsumeQueue的MappedFile最后一个存储单元对应消息在CommitLog中的偏移量小于CommitLog的最小偏移量,说明当前MappedFile所有存储单元对应所有CommitLog的消息都已经被清理,因此调用MappedFile#destroy清理当前MappedFile
    • 删除MappedFileQueue缓存的mappedFiles列表中已经被清理MappedFile
    // org.apache.rocketmq.store.MappedFileQueue#deleteExpiredFileByOffset
    public int deleteExpiredFileByOffset(long offset, int unitSize) {
        // 复制一份mappedFiles
        Object[] mfs = this.copyMappedFiles(0);
        List<MappedFile> files = new ArrayList<MappedFile>();
        int deleteCount = 0;
        if (null != mfs) {
            int mfsLength = mfs.length - 1;
            for (int i = 0; i < mfsLength; i++) {
                boolean destroy;
                MappedFile mappedFile = (MappedFile) mfs[i];
                // 取consumeQueue最后一条消息Buffer切片
                SelectMappedBufferResult result = mappedFile.selectMappedBuffer(this.mappedFileSize - unitSize);
                if (result != null) {
                    // consumeQueue最后一个存储单元消息在commitLog的偏移量
                    long maxOffsetInLogicQueue = result.getByteBuffer().getLong();
                    result.release();
                    // 如果consumeQueue最后一条消息已经小于commitLog的最小offset,则说明要删除了
                    destroy = maxOffsetInLogi编程客栈cQueue < offset;
                    if (destroy) {
                        log.info("physic min offset " + offset + ", logics in current mappedFile max offset "
                            + maxOffsetInLogicQueue + ", delete it");
                    }
                } 
                // 删除ConsumeQueue的MappedFile
                if (destroy && mappedFile.destroy(1000 * 60)) {
                    files.add(mappedFile);
                    deleteCount++;
                } else {
                    break;
                }
            }
        }
        // 删除MappedFileQueue的mappedFiles列表中已经删除的MappedFile
        deleteExpiredFile(files);
        return deleteCount;
    }

    IndexFile清理

    IndexFile清理逻辑与ConsumeQueue类似,都是删除文件中关联的CommitLog消息全部被删除的文件。核心逻辑包括下面两个

    • 获取IndexFileList中所有最大offset小于CommitLog最小offset的IndexFile
    • 删除过期的IndexFile
    // org.apache.rocketmq.store.index.IndexService#deleteExpiredFile(long)
    public void deleteExpiredFile(long offset) {
        Object[] files = null;
        try {
            // indexFileList的第一个索引文件的最后一个offset
            long endPhyOffset = this.indexFileList.get(0).getEndPhyOffset();
            if (endPhyOffset < offset) {
                files = this.indexFileList.toArray();
            }
        } 
        if (files != null) {
            List<IndexFile> fileList = new ArrayList<IndexFile>();
            for (int i = 0; i < (files.length - 1); i++) {
                IndexFile f = (IndexFile) files[i];
              	// IndexFile中最大的offset小于CommitLog最小offset,说明文件可以被删除
                if (f.getEndPhyOffset() < offset) {
                    fileList.add(f);
                } else {
                    break;
                }
            }
            // 删除过期的IndexFile,并将其从indexFileList缓存中删除
            this.deleteExpiredFile(fileList);
        }
    }
    

    总结

    Broker消息清理机制由DefaultMessageStore负责,CommitLog、ConsumeQueue和IndexFile的清理都是按照文件颗粒度进行。

    每10s检查一次,通常情况下每天凌晨4点删除超过72小时的CommitLog;如果CommitLog所在磁盘分区的磁盘占用率超过75%,则会触发CommitLog文件清理;如果CommitLog所在磁盘分区的磁盘占用率超过85%,则会强制删除CommitLog文件;

    如果ConsumeQueue和IndexFile关联CommitLog都被删除,ConsumeQueue文件和IndexFile也会被清理。

    以上就是RocketMQ | 源码分析】Broker过期消息清理机制的详细内容,更多关于RocketMQ | 源码分析】Broker过期消息清理机制的资料请关注我们其它相关文章!

    0

    精彩评论

    暂无评论...
    验证码 换一张
    取 消

    关注公众号