# 文件自動(dòng)清理時(shí)間，單位H，默認(rèn)72
fileReservedTime=72
# CommitLog物理文件刪除間隔，但是ms，默認(rèn)100
deleteCommitLogFilesInterval = 100
# 文件自動(dòng)清理時(shí)間，默認(rèn)04，即凌晨4點(diǎn)
deleteWhen = "04"
# 硬盤(pán)占用率所在分區(qū)過(guò)期清理警戒線，超過(guò)這個(gè)值，無(wú)論是否到達(dá)清理時(shí)間，都會(huì)自動(dòng)清理過(guò)期文件
diskMaxUsedSpaceRatio = 75

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

// org.apache.rocketmq.store.DefaultMessageStore#cleanFilesPeriodically
private void cleanFilesPeriodically() {
    // 清理CommitLog
    this.cleanCommitLogService.run();
    // 清理ConsumeQueue和IndexFile
    this.cleanConsumeQueueService.run();
}

// org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles
private void deleteExpiredFiles() {
    // 是否是凌晨4點(diǎn)，用小時(shí)匹配[04:00,05:00)
    boolean timeup = this.isTimeToDelete();
    // >75%就會(huì)返回true，如果大于85%，則觸發(fā)強(qiáng)制刪除
    boolean spacefull = this.isSpaceToDelete();
    // 手動(dòng)刪除次數(shù)是否>0
    boolean manualDelete = this.manualDeleteFileSeveralTimes > 0;
    if (timeup/*凌晨4點(diǎn)*/ || spacefull/*空間滿了*/ || manualDelete/*手動(dòng)刪除*/) {
        boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable()/*默認(rèn)true*/ && this.cleanImmediately/*空間占用超過(guò)85%，觸發(fā)強(qiáng)制刪除*/;
				// 刪除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.");
        }
    }
}

// org.apache.rocketmq.store.MappedFileQueue#deleteExpiredFileByTime
public int deleteExpiredFileByTime(final long expiredTime,final int deleteFilesInterval,final long intervalForcibly,final boolean cleanImmediately) {
    // 復(fù)制一份當(dāng)前mappedFile
    Object[] mfs = this.copyMappedFiles(0);
		// 會(huì)保留最后一個(gè)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];
            // 最后修改時(shí)間+過(guò)期時(shí)間
            long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
            // 如果commitLog所在磁盤(pán)分區(qū)總?cè)萘砍^(guò)85%，觸發(fā)立即刪除，或者超過(guò)了72小時(shí)的mappedFile
            if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) {
                // 刪除mappedFile
                if (mappedFile.destroy(intervalForcibly)) {
                    files.add(mappedFile);
                    deleteCount++;
                    // 一次最多刪除10個(gè)mappedFile
                    if (files.size() >= DELETE_FILES_BATCH_MAX/*10*/) {
                        break;
                    }
                    if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
                        try {
                            // 刪除文件時(shí)間間隔，默認(rèn)100ms
                            Thread.sleep(deleteFilesInterval);
                        } catch (InterruptedException e) {
                        }
                    }
                } else {
                    break;
                }
            } else {
                //avoid deleting files in the middle
                break;
            }
        }
    }
    // 從MappedFileQueue的mappedFiles中刪除這個(gè)mappedFile
    deleteExpiredFile(files);
    return deleteCount;
}

// 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);
    }
}

// org.apache.rocketmq.store.MappedFileQueue#deleteExpiredFileByOffset
public int deleteExpiredFileByOffset(long offset, int unitSize) {
    // 復(fù)制一份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最后一個(gè)存儲(chǔ)單元消息在commitLog的偏移量
                long maxOffsetInLogicQueue = result.getByteBuffer().getLong();
                result.release();
                // 如果consumeQueue最后一條消息已經(jīng)小于commitLog的最小offset，則說(shuō)明要?jiǎng)h除了
                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列表中已經(jīng)刪除的MappedFile
    deleteExpiredFile(files);
    return deleteCount;
}

// org.apache.rocketmq.store.index.IndexService#deleteExpiredFile(long)
public void deleteExpiredFile(long offset) {
    Object[] files = null;
    try {
        // indexFileList的第一個(gè)索引文件的最后一個(gè)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，說(shuō)明文件可以被刪除
            if (f.getEndPhyOffset() < offset) {
                fileList.add(f);
            } else {
                break;
            }
        }
        // 刪除過(guò)期的IndexFile，并將其從indexFileList緩存中刪除
        this.deleteExpiredFile(fileList);
    }
}