RocketMQ-04、消息存储(6)

4.8.2.Broker异步刷盘

    public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) { // Synchronization flush if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) { final GroupCommitService service = (GroupCommitService) this.flushCommitLogService; if (messageExt.isWaitStoreMsgOK()) { GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes()); service.putRequest(request); boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()); if (!flushOK) { log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags() + " client address: " + messageExt.getBornHostString()); putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT); } } else { service.wakeup(); } } // Asynchronous flush else { if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) { flushCommitLogService.wakeup(); } else { commitLogService.wakeup(); } } } 

// Asynchronous flush else { if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) { flushCommitLogService.wakeup(); } else { commitLogService.wakeup(); } } 

异步刷盘根据是否开启transientStorePoolEnable机制，刷盘实现会有细微差别。如果transientStorePoolEnable 为true , RocketMQ 会单独申请一个与目标物理文件（ commitlog)同样大小的堆外内存， 该堆外内存将使用内存锁定，确保不会被置换到虚拟内存中去，消息首先追加到堆外内存，然后提交到与物理文件的内存映射内存中，再flush 到磁盘。如果transientStorePoolEnable 为flalse ，消息直接追加到与物理文件直接映射的内存中，然后刷写到磁盘中。transientStorePoolEnable 为true 的磁盘刷写流程如图4-20 所示。

• 首先将消息直接追加到ByteBuffer （堆外内存DirectByteBuffer ), wrotePosition 随着消息的不断追加向后移动。
• CommitRealTimeService 线程默认每200ms 将ByteBuffer 新追加的内容（ wrotePosihon减去commitedPosition ）的数据提交到Ma ppedB yte Buff1巳r 中。
• MappedByt巳Buffer 在内存中追加提交的内容， wrotePosition 指针向前后移动， 然后返回。
• commit 操作成功返回，将commitedPosition 向前后移动本次提交的内容长度，此时wrotePosition 指针依然可以向前推进。
• FlushRealTimeService 线程默认每500ms 将MappedByteBuffer 中新追加的内存( wrotePosition 减去上一次刷写位置flushedPositiont ）通过调用MappedByteBuffer#force()方法将数据刷写到磁盘。

 

image

CommitRealTimeService 提交线程工作机制

@Override public void run() { CommitLog.log.info(this.getServiceName() + " service started"); while (!this.isStopped()) { int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog(); int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages(); int commitDataThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval(); long begin = System.currentTimeMillis(); if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) { this.lastCommitTimestamp = begin; commitDataLeastPages = 0; } try { boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages); long end = System.currentTimeMillis(); if (!result) { this.lastCommitTimestamp = end; // result = false means some data committed. //now wake up flush thread. flushCommitLogService.wakeup(); } if (end - begin > 500) { log.info("Commit data to file costs {} ms", end - begin); } this.waitForRunning(interval); } catch (Throwable e) { CommitLog.log.error(this.getServiceName() + " service has exception. ", e); } } boolean result = false; for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) { result = CommitLog.this.mappedFileQueue.commit(0); CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK")); } CommitLog.log.info(this.getServiceName() + " service end"); } 

int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog(); int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages(); int commitDataThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval(); 

Step1： 首先解释三个配置参数的含义。

• commitlnterva!CommitLog: CommitRea!TimeService 线程间隔时间，默认200ms 。
• commitCommitLogLeastPages ： 一次提交任务至少包含页数， 如果待提交数据不足，小于该参数配置的值，将忽略本次提交任务，默认4 页。
• commitDataThoroughinterval ：两次真实提交最大间隔，默认200ms 。

    long begin = System.currentTimeMillis(); if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) { this.lastCommitTimestamp = begin; commitDataLeastPages = 0; } 

Step2 ：如果距上次提交间隔超过commitDataThoroughInterval ， 则本次提交忽略commitCommitLogLeastPages参数， 也就是如果待提交数据小于指定页数， 也执行提交操作。

    boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages); long end = System.currentTimeMillis(); if (!result) { this.lastCommitTimestamp = end; // result = false means some data committed. //now wake up flush thread. flushCommitLogService.wakeup(); } if (end - begin > 500) { log.info("Commit data to file costs {} ms", end - begin); } this.waitForRunning(interval); 

Step3 ：执行提交操作，将待提交数据提交到物理文件的内存映射内存区，如果返回false ，并不是代表提交失败，而是只提交了一部分数据，唤醒刷盘线程执行刷盘操作。该线程每完成一次提交动作，将等待2 00ms 再继续执行下一次提交任务。

FlushRealTimeService 刷盘线程工作机制

        public void run() { CommitLog.log.info(this.getServiceName() + " service started"); while (!this.isStopped()) { boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed(); int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog(); int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages(); int flushPhysicQueueThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval(); boolean printFlushProgress = false; // Print flush progress long currentTimeMillis = System.currentTimeMillis(); if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) { this.lastFlushTimestamp = currentTimeMillis; flushPhysicQueueLeastPages = 0; printFlushProgress = (printTimes++ % 10) == 0; } try { if (flushCommitLogTimed) { Thread.sleep(interval); } else { this.waitForRunning(interval); } if (printFlushProgress) { this.printFlushProgress(); } long begin = System.currentTimeMillis(); CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages); long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp(); if (storeTimestamp > 0) { CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp); } long past = System.currentTimeMillis() - begin; if (past > 500) { log.info("Flush data to disk costs {} ms", past); } } catch (Throwable e) { CommitLog.log.warn(this.getServiceName() + " service has exception. ", e); this.printFlushProgress(); } } // Normal shutdown, to ensure that all the flush before exit boolean result = false; for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) { result = CommitLog.this.mappedFileQueue.flush(0); CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK")); } this.printFlushProgress(); CommitLog.log.info(this.getServiceName() + " service end"); } 

boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed(); int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog(); int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages(); int flushPhysicQueueThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval(); 

Step1： 首先解释四个配置参数的含义。

• flushCommitLogTimed ： 默认为false ， 表示await 方法等待；如果为true ，表示使用Thread.sleep 方法等待。
• flushIntervalCommitLog: FlushRealTimeService 线程任务运行间隔。
• flushPhysicQueueLeastPages ： 一次刷写任务至少包含页数， 如果待刷写数据不足，小于该参数配置的值，将忽略本次刷写任务，默认4 页。
• flushPhysicQueueThoroughInterval ：两次真实刷写任务最大间隔， 默认10s 。

// Print flush progress long currentTimeMillis = System.currentTimeMillis(); if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) { this.lastFlushTimestamp = currentTimeMillis; flushPhysicQueueLeastPages = 0; printFlushProgress = (printTimes++ % 10) == 0; } 

Step2 ：如果距上次提交间隔超过flushPhysicQueueThoroughInterval ，则本次刷盘任务将忽略flushPhysicQueueLeastPages ， 也就是如果待刷写数据小于指定页数也执行刷写磁盘操作。

    if (flushCommitLogTimed) { Thread.sleep(interval); } else { this.waitForRunning(interval); } 

Step3 ：执行一次刷盘任务前先等待指定时间间隔， 然后再执行刷盘任务。

    long begin = System.currentTimeMillis(); CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages); long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp(); if (storeTimestamp > 0) { CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp); } 

Step4 ：调用flush 方法将内存中数据刷写到磁盘，并且更新存储检测点文件的comm1tlog 文件的更新时间戳，文件检测点文件（ checkpoint 文件）的刷盘动作在刷盘消息消费队列线程中执行， 其入口为DefaultMessageStore# FlushConsumeQueueS 巳rvice 。由于消息消费队列、索引文件的刷盘实现原理与Comm itlog 文件的刷盘机制类同，故本书不再做重复分析。

4.9.过期文件删除机制

由于RocketMQ 操作CommitLog 、ConsumeQueue 文件是基于内存映射机制并在启动的时候会加载commitlog 、ConsumeQueue 目录下的所有文件，为了避免内存与磁盘的浪费，不可能将消息永久存储在消息服务器上，所以需要引人一种机制来删除己过期的文件。RocketMQ 顺序写Commitlog 文件、Cons umeQueue 文件，所有写操作全部落在最后一个CommitLog 或Cons umeQueu e 文件上，之前的文件在下一个文件创建后将不会再被更新。RocketMQ 清除过期文件的方法是：如果非当前写文件在一定时间间隔内没有再次被更新，则认为是过期文件，可以被删除， RocketMQ 不会关注这个文件上的消息是否全部被消费。默认每个文件的过期时间为72 小时，通过在Broker 配置文件中设置fi leReservedTime 来改变过期时间，单位为小时· 。接下来详细分析RocketMQ 是如何设计与实现上述机制的。

    private void addScheduleTask() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { DefaultMessageStore.this.cleanFilesPeriodically(); } }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { DefaultMessageStore.this.checkSelf(); } }, 1, 10, TimeUnit.MINUTES); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { if (DefaultMessageStore.this.getMessageStoreConfig().isDebugLockEnable()) { try { if (DefaultMessageStore.this.commitLog.getBeginTimeInLock() != 0) { long lockTime = System.currentTimeMillis() - DefaultMessageStore.this.commitLog.getBeginTimeInLock(); if (lockTime > 1000 && lockTime < 10000000) { String stack = UtilAll.jstack(); final String fileName = System.getProperty("user.home") + File.separator + "debug/lock/stack-" + DefaultMessageStore.this.commitLog.getBeginTimeInLock() + "-" + lockTime; MixAll.string2FileNotSafe(stack, fileName); } } } catch (Exception e) { } } } }, 1, 1, TimeUnit.SECONDS); // this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { // @Override // public void run() { // DefaultMessageStore.this.cleanExpiredConsumerQueue(); // } // }, 1, 1, TimeUnit.HOURS); } 

RocketMQ 会每隔10 s 调度一次cleanFilesPeriodically ， 检测是否需要清除过期文件。执行频率可以通过设置cleanResourceInterval ，默认为10 s 。

    private void cleanFilesPeriodically() { this.cleanCommitLogService.run(); this.cleanConsumeQueueService.run(); } 

分别执行清除消息存储文件（ Commitlog 文件）与消息消费队列文件（ ConsumeQueue文件） 。由于消息消费队列文件与消息存储文件（ Commitlo g ）共用一套过期文件删除机制，本书将重点讲解消息存储过期文件删除。实现方法： DefaultMessage Store$Clean CommitLogService#deleteExpiredFiles 。

        private void deleteExpiredFiles() { int deleteCount = 0; long fileReservedTime = DefaultMessageStore.this.getMessageStoreConfig().getFileReservedTime(); int deletePhysicFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteCommitLogFilesInterval(); int destroyMapedFileIntervalForcibly = DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly(); boolean timeup = this.isTimeToDelete(); boolean spacefull = this.isSpaceToDelete(); 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: {}", fileReservedTime, timeup, spacefull, manualDeleteFileSeveralTimes, cleanAtOnce); fileReservedTime *= 60 * 60 * 1000; 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."); } } } 

Step1： 解释一下这个三个配置属性的含义。

int deleteCount = 0; long fileReservedTime = DefaultMessageStore.this.getMessageStoreConfig().getFileReservedTime(); int deletePhysicFilesInterval = DefaultMessageStore.this.getMessageStoreConfig().getDeleteCommitLogFilesInterval(); int destroyMapedFileIntervalForcibly = DefaultMessageStore.this.getMessageStoreConfig().getDestroyMapedFileIntervalForcibly(); 

• fileReservedTime ： 文件保留时间， 也就是从最后一次更新时间到现在， 如果超过了该时间， 则认为是过期文件， 可以被删除。
• deletePhysicFilesInterval ：删除物理文件的间隔，因为在一次清除过程中， 可能需要被删除的文件不止一个，该值指定两次删除文件的问隔时间。
• destroyMapedFilelntervalForcibly ： 在清除过期文件时， 如果该文件被其他线程所占用（引用次数大于0 ，比如读取消息）， 此时会阻止此次删除任务， 同时在第一次试图删除该文件时记录当前时间戳， destroyMapedFile lntervalForcibly 表示第一次拒绝删除之后能保留的最大时间，在此时间内， 同样可以被拒绝删除， 同时会将引用减少1000 个，超过该时间间隔后，文件将被强制删除。

boolean timeup = this.isTimeToDelete(); boolean spacefull = this.isSpaceToDelete(); 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: {}", fileReservedTime, timeup, spacefull, manualDeleteFileSeveralTimes, cleanAtOnce); fileReservedTime *= 60 * 60 * 1000; 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."); } } 

Step2: RocketMQ 在如下三种情况任意之一满足的情况下将继续执行删除文件操作。

• 指定删除文件的时间点， RocketMQ 通过delete When 设置一天的固定时间执行一次删除过期文件操作， 默认为凌晨4 点。
• 磁盘空间是否充足，如果磁盘空间不充足，则返回true ，表示应该触发过期文件删除操作。
• 预留，手工触发，可以通过调用excuteDeleteFilesManualy 方法手工触发过期文件删除，目前RocketMQ 暂未封装手工触发文件删除的命令。

本节重点分析一下磁盘空间是否充足的实现逻辑。

        private boolean isSpaceToDelete() { double ratio = DefaultMessageStore.this.getMessageStoreConfig().getDiskMaxUsedSpaceRatio() / 100.0; cleanImmediately = false; { String storePathPhysic = DefaultMessageStore.this.getMessageStoreConfig().getStorePathCommitLog(); double physicRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathPhysic); if (physicRatio > diskSpaceWarningLevelRatio) { boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskFull(); if (diskok) { DefaultMessageStore.log.error("physic disk maybe full soon " + physicRatio + ", so mark disk full"); } cleanImmediately = true; } else if (physicRatio > diskSpaceCleanForciblyRatio) { cleanImmediately = true; } else { boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskOK(); if (!diskok) { DefaultMessageStore.log.info("physic disk space OK " + physicRatio + ", so mark disk ok"); } } if (physicRatio < 0 || physicRatio > ratio) { DefaultMessageStore.log.info("physic disk maybe full soon, so reclaim space, " + physicRatio); return true; } } { String storePathLogics = StorePathConfigHelper .getStorePathConsumeQueue(DefaultMessageStore.this.getMessageStoreConfig().getStorePathRootDir()); double logicsRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathLogics); if (logicsRatio > diskSpaceWarningLevelRatio) { boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskFull(); if (diskok) { DefaultMessageStore.log.error("logics disk maybe full soon " + logicsRatio + ", so mark disk full"); } cleanImmediately = true; } else if (logicsRatio > diskSpaceCleanForciblyRatio) { cleanImmediately = true; } else { boolean diskok = DefaultMessageStore.this.runningFlags.getAndMakeDiskOK(); if (!diskok) { DefaultMessageStore.log.info("logics disk space OK " + logicsRatio + ", so mark disk ok"); } } if (logicsRatio < 0 || logicsRatio > ratio) { DefaultMessageStore.log.info("logics disk maybe full soon, so reclaim space, " + logicsRatio); return true; } } return false; } 

1 ）首先解释一下几个参数的含义。

• diskMaxUsedSpaceRatio ： 表示commitlog 、consumequeue 文件所在磁盘分区的最大使用量，如果超过该值， 则需要立即清除过期文件。
• cleanImmediately ： 表示是否需要立即执行清除过期文件操作。
• physicRatio ： 当前commitlog 目录所在的磁盘分区的磁盘使用率，通过File#getTotalSpace（）获取文件所在磁盘分区的总容量，通过File#getFreeSpace() 获取文件所在磁盘分区剩余容量。
• diskSpaceWarningLevelRatio ： 通过系统参数－Drocketmq.broker.diskSpace WamingLevelRatio设置，默认0.90 。如果磁盘分区使用率超过该阔值，将设置磁盘不可写，此时会拒绝新消息的写人。
• diskSpaceCleanForciblyRatio ：通过系统参数－Drocketmq.broker.diskSpaceCleanF orciblyRatio设置， 默认0.85 。如果磁盘分区使用超过该阔值，建议立即执行过期文件清除，但不会拒绝新消息的写入。

2 ） 如果当前磁盘分区使用率大于diskSpace WarningLeve!Ratio ，设置磁盘不可写，应该立即启动过期文件删除操作；如果当前磁盘分区使用率大于diskSpaceCleanForciblyRatio,建议立即执行过期文件清除；如果磁盘使用率低于diskSpaceCl eanF orcibly Ratio 将恢复磁盘可写；如果当前磁盘使用率小于diskMax U sedSpaceRatio 则返回false ，表示磁盘使用率正常，否则返回true ， 需要执行清除过期文件。

for (int i = 0; i < mfsLength; i++) { MappedFile mappedFile = (MappedFile) mfs[i]; long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime; if (System.currentTimeMillis() >= liveMaxTimestamp || cleanImmediately) { if (mappedFile.destroy(intervalForcibly)) { files.add(mappedFile); deleteCount++; if (files.size() >= DELETE_FILES_BATCH_MAX) { break; } if (deleteFilesInterval > 0 && (i + 1) < mfsLength) { try { Thread.sleep(deleteFilesInterval); } catch (InterruptedException e) { } } } else { break; } } else { //avoid deleting files in the middle break; } } 

执行文件销毁与删除。从倒数第二个文件开始遍历，计算文件的最大存活时间（ ＝ 文件的最后一次更新时间＋文件存活时间（默认72 小时）） ， 如果当前时间大于文件的最大存活时间或需要强制删除文件（当磁盘使用超过设定的阔值）时则执行MappedFile#destory 方法，清除MappedFile 占有的相关资源，如果执行成功，将该文件加入到待删除文件列表中，然后统一执行File#delete 方法将文件从物理磁盘中删除。

4.10.本章小节

RocketMQ 主要存储文件包含消息文件（ commitlog ）、消息消费队列文件（ConsumeQueue）、Hash 索引文件(indexFile)、检测点文件（ checkpoint ） 、abort （关闭异常文件） 。单个消息存储文件、消息消费队列文件、Hash 索引文件长度固定以便使用内存映射机制进行文件的读写操作。RocketMQ 组织文件以文件的起始偏移量来命名文件，这样根据偏移量能快速定位到真实的物理文件。RocketMQ 基于内存映射文件机制提供了同步刷盘与异步刷盘两种机制，异步刷盘是指在消息存储时先追加到内存映射文件，然后启动专门的刷盘线程定时将内存中的数据刷写到磁盘。

Commitlog，消息存储文件， RocketMQ 为了保证消息发送的高吞吐量，采用单一文件存储所有主题的消息，保证消息存储是完全的顺序写，但这样给文件读取同样带来了不便，为此RocketMQ 为了方便消息消费构建了消息消费队列文件，基于主题与队列进行组织， 同时RocketMQ 为消息实现了Hash 索引，可以为消息设置索引键，根据索引能够快速从Commitog 文件中检索消息。

当消息到达Commitlog 文件后，会通过ReputMessageService 线程接近实时地将消息转发给消息消费队列文件与索引文件。为了安全起见， RocketMQ 引人abort 文件，记录Broker 的停机是正常关闭还是异常关闭，在重启Broker 时为了保证Commitlog 文件、消息消费队列文件与Hash 索引文件的正确性，分别采取不同的策略来恢复文件。

RocketMQ 不会永久存储消息文件、消息消费队列文件，而是启用文件过期机制并在磁盘空间不足或默认在凌晨4 点删除过期文件，文件默认保存72 小时并且在删除文件时并不会判断该消息文件上的消息是否被消费。下面一章我们将重点分析有关消息、消费的实现机制。

转https://www.jianshu.com/p/8ee482ef6911

https://itzones.cn/2019/07/24/RocketMQ%E5%88%B7%E7%9B%98%E7%AD%96%E7%95%A5/