public class StoreCheckpoint {
    // commitLog最后一條信息的刷盤時(shí)間戳
    private volatile long physicMsgTimestamp = 0;
    // consumeQueue最后一個(gè)存儲(chǔ)單元刷盤時(shí)間戳
    private volatile long logicsMsgTimestamp = 0;
    // 最近一個(gè)已經(jīng)寫完IndexFile的最后一條記錄刷盤時(shí)間戳
    private volatile long indexMsgTimestamp = 0;
}

// org.apache.rocketmq.store.CommitLog.FlushRealTimeService#run
public void run() {
  // ...
  // 更新CommitLog刷盤時(shí)間戳
  if (storeTimestamp > 0) { 					       CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
  }
}

// org.apache.rocketmq.store.CommitLog.GroupCommitService#doCommit
private void doCommit() {
  // ...
  // 更新CommitLog刷盤時(shí)間戳
  if (storeTimestamp > 0) {
CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
  }
}

// org.apache.rocketmq.store.ConsumeQueue#putMessagePositionInfoWrapper
public void putMessagePositionInfoWrapper(DispatchRequest request, boolean multiQueue) {
  // ...
  // 如果consumeQueue保存成功，則更新ConsumeQueue存儲(chǔ)點(diǎn)信息
  if (result) {
this.defaultMessageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
  }
}

// org.apache.rocketmq.store.DefaultMessageStore.FlushConsumeQueueService#doFlush
private void doFlush(int retryTimes) {
  // ...
  // 更新ConsumeQueue存儲(chǔ)時(shí)間戳，并刷盤
  if (0 == flushConsumeQueueLeastPages) {
    if (logicsMsgTimestamp > 0) {
  DefaultMessageStore.this.getStoreCheckpoint().setLogicsMsgTimestamp(logicsMsgTimestamp);
    }
    // 更新存儲(chǔ)點(diǎn)
    DefaultMessageStore.this.getStoreCheckpoint().flush();
  }
}

// org.apache.rocketmq.store.index.IndexService#getAndCreateLastIndexFile
public IndexFile getAndCreateLastIndexFile() {
  // 獲取最新IndexFile，如果IndexFile已經(jīng)滿了，需要?jiǎng)?chuàng)建一個(gè)新的IndexFile
  if (indexFile == null) {
          indexFile =
              new IndexFile(fileName, this.hashSlotNum, this.indexNum, lastUpdateEndPhyOffset,
                  lastUpdateIndexTimestamp);
			// 如果創(chuàng)建新的IndexFile成功，原IndexFile刷盤
      if (indexFile != null) {
          final IndexFile flushThisFile = prevIndexFile;
          Thread flushThread = new Thread(new Runnable() {
              @Override
              public void run() {
                	// indexFile刷盤
                  IndexService.this.flush(flushThisFile);
              }
          }, "FlushIndexFileThread");
          flushThread.setDaemon(true);
          flushThread.start();
      }
  }
  return indexFile;
}
// org.apache.rocketmq.store.index.IndexService#flush
public void flush(final IndexFile f) {
    if (null == f)
        return;
    long indexMsgTimestamp = 0;
    if (f.isWriteFull()) {
        indexMsgTimestamp = f.getEndTimestamp();
    }
    f.flush();
    if (indexMsgTimestamp > 0) {
        // 更新checkPoint的indexMsgTimestamp并觸發(fā)刷盤
        this.defaultMessageStore.getStoreCheckpoint().setIndexMsgTimestamp(indexMsgTimestamp);
        this.defaultMessageStore.getStoreCheckpoint().flush();
    }
}

// org.apache.rocketmq.store.StoreCheckpoint#flush
public void flush() {
    this.mappedByteBuffer.putLong(0, this.physicMsgTimestamp);
    this.mappedByteBuffer.putLong(8, this.logicsMsgTimestamp);
    this.mappedByteBuffer.putLong(16, this.indexMsgTimestamp);
    this.mappedByteBuffer.force();
}

// org.apache.rocketmq.store.DefaultMessageStore#load
public boolean load() {
    boolean result = true;
    try {
      	// 1. Broker上次是否正常退出	
        boolean lastExitOK = !this.isTempFileExist();
        log.info("last shutdown {}", lastExitOK ? "normally" : "abnormally");
        // 2. 加載commitLog
        result = result && this.commitLog.load();
				// 3. 加載consumeQueue
        result = result && this.loadConsumeQueue();
        if (result) {
            // 4. 加載StoreCheckPoint
            this.storeCheckpoint =
                new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
            // 5. 加載IndexFile
            this.indexService.load(lastExitOK);
            // 6. 恢復(fù)ConsumeQueue與IndexFile
            this.recover(lastExitOK);
						// 7. 延遲隊(duì)列服務(wù)加載
            if (null != scheduleMessageService) {
                result =  this.scheduleMessageService.load();
            }
        }
    } 
    return result;
}

// org.apache.rocketmq.store.MappedFileQueue#load
public boolean load() {
    File dir = new File(this.storePath/*${user.home}/store/commitlog/*/);
    File[] ls = dir.listFiles();
    if (ls != null) {
        return doLoad(Arrays.asList(ls));
    }
    return true;
}

public boolean doLoad(List<File> files) {
    // 按照文件名稱排序
    files.sort(Comparator.comparing(File::getName));
    for (File file : files) {
      	// 如果commitLog文件大小與mappedFileSize不一致，則說明配置被改了，或者CommitLog文件被修改
        if (file.length() != this.mappedFileSize) {
            return false;
        }
        try {
          	// 創(chuàng)建MappedFile
            MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);
            mappedFile.setWrotePosition(this.mappedFileSize);
            mappedFile.setFlushedPosition(this.mappedFileSize);
            mappedFile.setCommittedPosition(this.mappedFileSize);
            this.mappedFiles.add(mappedFile);
        } 
    }
    return true;
}

// org.apache.rocketmq.store.DefaultMessageStore#loadConsumeQueue
private boolean loadConsumeQueue() {
  // 獲取consumeQueue目錄
  File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()/*${user.home}/store */));
  // topic文件夾數(shù)組
  File[] fileTopicList = dirLogic.listFiles();
  if (fileTopicList != null) {
      // 遍歷topic	
      for (File fileTopic : fileTopicList) {
          // 獲取topic名稱
          String topic = fileTopic.getName();
          // 獲取queueId文件夾數(shù)組
          File[] fileQueueIdList = fileTopic.listFiles();
          // 遍歷queueId
          if (fileQueueIdList != null) {
              for (File fileQueueId : fileQueueIdList) {
                  int queueId;
                  // 文件夾名稱就是queueId
                  queueId = Integer.parseInt(fileQueueId.getName());
                  // 構(gòu)建consumeQueue
                  ConsumeQueue logic = new ConsumeQueue(/* ... */);
                  this.putConsumeQueue(topic, queueId, logic);
                  // ConsumeQueue加載
                  if (!logic.load()) {
                      return false;
                  }
              }
          }
      }
  }
  return true;
}

// org.apache.rocketmq.store.index.IndexService#load
public boolean load(final boolean lastExitOK) {
    // indexFile文件目錄
    File dir = new File(this.storePath);
    // indexFile文件列表
    File[] files = dir.listFiles();
    if (files != null) {
        // 文件排序
        Arrays.sort(files);
        for (File file : files) {
            try {
                IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
                f.load();
                if (!lastExitOK) {
                    // 文件最后存儲(chǔ)時(shí)間戳大于刷盤點(diǎn)，則摧毀indexFile，重建
                    if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint()/*存儲(chǔ)點(diǎn)時(shí)間*/
                        .getIndexMsgTimestamp()) {
                        f.destroy(0);
                        continue;
                    }
                }
                this.indexFileList.add(f);
            } 
        }
    }
    return true;
}

// org.apache.rocketmq.store.DefaultMessageStore#recover
private void recover(final boolean lastExitOK) {
    // consumeQueue中最大物理偏移量
    long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();
    if (lastExitOK) {
        // 正常退出文件恢復(fù)
        this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue);
    } else {
        // 異常退出文件恢復(fù)
        this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue);
    }
    // 恢復(fù)topicQueueTable
    this.recoverTopicQueueTable();
}

    public void recoverNormally(long maxPhyOffsetOfConsumeQueue) {
        // 確認(rèn)消息是否完整，默認(rèn)是true
        boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
        final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            // 默認(rèn)從最后3個(gè)mappedFile開始恢復(fù)
            int index = mappedFiles.size() - 3;
            // 如果commitLog不足三個(gè)，則從第一個(gè)文件開始恢復(fù)
            if (index < 0)
                index = 0;
            MappedFile mappedFile = mappedFiles.get(index);
            ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
            // 最后一個(gè)MappedFile的文件起始偏移量
            long processOffset = mappedFile.getFileFromOffset();
            // mappedFileOffset偏移量
            long mappedFileOffset = 0;
            // 遍歷CommitLog文件
            while (true) {
                // 校驗(yàn)消息完整性
                DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                // 獲取消息size
                int size = dispatchRequest.getMsgSize();
                // 返回結(jié)果為true并且消息size>0，說明消息是完整的
                if (dispatchRequest.isSuccess() && size > 0) {
                    mappedFileOffset += size;
                }
            }
            // 最大物理偏移量
            processOffset += mappedFileOffset;
            // 更新flushedWhere和committedPosition指針
            this.mappedFileQueue.setFlushedWhere(processOffset);
            this.mappedFileQueue.setCommittedWhere(processOffset);
            this.mappedFileQueue.truncateDirtyFiles(processOffset);
            // 清除ConsumeQueue冗余數(shù)據(jù)
            if (maxPhyOffsetOfConsumeQueue >= processOffset) {
                this.defaultMessageStore.truncateDirtyLogicFiles(processOffset/*CommitLog最大物理偏移量*/);
            }
        } 
    }

// org.apache.rocketmq.store.CommitLog#recoverAbnormally
public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
    // 是否CRC校驗(yàn)
    boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
    final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
    if (!mappedFiles.isEmpty()) {
        // 最后一個(gè)mappedFile的index
        int index = mappedFiles.size() - 1;
        MappedFile mappedFile = null;
        // 倒序遍歷mappedFile數(shù)組，
        for (; index >= 0; index--) {
            mappedFile = mappedFiles.get(index);
            // 1. 如果第一條消息的時(shí)間戳小于存儲(chǔ)點(diǎn)時(shí)間戳
            if (this.isMappedFileMatchedRecover(mappedFile)) {
                break;
            }
        }
        long processOffset = mappedFile.getFileFromOffset();
        long mappedFileOffset = 0;
        while (true) {
            DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
            int size = dispatchRequest.getMsgSize();
            if (dispatchRequest.isSuccess()) {
                if (size > 0) {
                    mappedFileOffset += size;
                    // 2. 轉(zhuǎn)發(fā)消息
                    if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()/*消息是否可以重復(fù)，默認(rèn)是false*/) {
                        if (dispatchRequest.getCommitLogOffset() < this.defaultMessageStore.getConfirmOffset()) {
                            this.defaultMessageStore.doDispatch(dispatchRequest);
                        }
                    } else {
                        this.defaultMessageStore.doDispatch(dispatchRequest);
                    }
                }
        }
				// 3. 更新MappedFileQueue中的刷盤位置和提交位置
        processOffset += mappedFileOffset;
        this.mappedFileQueue.setFlushedWhere(processOffset);
        this.mappedFileQueue.setCommittedWhere(processOffset);
        this.mappedFileQueue.truncateDirtyFiles(processOffset);
        // 清除ConsumeQueue中的冗余數(shù)據(jù)
        if (maxPhyOffsetOfConsumeQueue >= processOffset) {
            this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
        }
    }
}