RocketMQ源码分析之Broker过期消息清理机制
作者：林师傅 发布时间：2023-06-10 15:47:43
前言
前面文章讲了消息是如何保存的以及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、consumeQueue和IndexFile定时删除
DefaultMessageStore.this.cleanFilesPeriodically();
}
}, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval()/*10s*/, TimeUnit.MILLISECONDS);
// ...
}
在cleanFilesPeriodically()中有两个方法，cleanCommitLogService.run()负责清理CommitLog，cleanConsumeQueueService.run()负责清理ConsumeQueue和IndexFile。
// org.apache.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 liveMaxTimestamp = 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) {
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);
}
}
ConsumeQueue文件清理
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 = maxOffsetInLogicQueue < 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);
}
}
来源：https://juejin.cn/post/7229799197096149049
0
投稿猜你喜欢
- 本文实例讲述了Java泛型的使用限制。分享给大家供大家参考，具体如下：一 什么情况下不能使用泛型1 不能使用泛型的形参创建对象。T o=ne
- 1.引言在开发中，拖放是一种比较常见的手势操作，使用它能够让应用的交互更加地便捷和友好，本文将简要介绍如何为Android中的View添加拖
- 首先给出一段代码：public class AslistMethod { public static void main(String[]
一、代码@Componentpublic class BService { @Autowired &- 一、ArrayList简介在集合框架中，ArrayList是一个普通的类，实现了List接口，具体框架图如下：ArrayList底层是一段连
- 异常处理是每个项目中都绕不开的话题，那么如何优雅的处理异常，是本文的话题。本文将结合SpringBoot框架一起和大家探讨下。要思考的问题在
- java如何实现ftp上传？如何创建文件夹？最佳答案：准备条件：java实现ftp上传用到了commons-net-3.3.jar包首先建立
- java中的LIST在删除时，一般会用list.remove(o); 但这样往往会出现问题，先来看下面的这段代码：package com.d
- SessionFactory在Hibernate中实际上起到了一个缓冲区的作用 他缓冲了HI
- 一.话题引入在做项目过程中，我们一般都是最先编写登录注册功能，登录功能最重要的是登录成功后，系统还会保存该登录用户信息，这种保存用户信息的逻
- 概述MerkleTree被广泛的应用在比特币技术中，本文旨在通过代码实现一个简单的MerkleTree，并计算出Merkle tree的 T
- 前言在框架中经常会会用到method.invoke()方法，用来执行某个的对象的目标方法。以前写代码用到反射时，总是获取先获取Method，
- 本文实例讲述了C#编程获取IP地址的方法。分享给大家供大家参考，具体如下：1、获取客户端IP/// <summary>/// 获
- 实体类package com.whty.entity;public class User {private int id;private S
- 1.服务配置中心1.1 服务配置中心介绍首先我们来看一下,微服务架构下关于配置文件的一些问题:1.配置文件相对分散。在一个微服务架构下，配置
- 目录1、this代表了()的对象引用,super表示的是当前对象的()对象?2、输出内容是：3、下面程序的输出是:()4、执行下列代码的输出
- 有时候我们需要在一个ArrayList的for循环中动态删除元素的需求, 废话不多说看代码List<Integer> list
- 实践过程效果代码public partial class Form1 : Form{ public Form1()
- 前言最近写一个东东,可能会考虑到字符串拼接,想了几种方法,但对性能未知，所以下面就来测试下面，话不多说了，来一起看看详细的介绍吧。示例代码p
- 封面图下个季度的目标是把前端监控相关的内容梳理出来，梳理出来之后可能会在公司内部做个分享～Flutter应用程序既括代码也包括一些其他的资产
