RocketMQ原理解析-Broker

1. 部署方式
    Broker可部署为2m-2s-async, 2m-2s-sync, 2m-noslave。推荐的部署方式为1.

2. 文件结构

    abort：broker启动后会创建此空文件，当正常关闭broker后，将其删除。该文件的作用是，当broker启动后检测到该文件存在，说明上次broker是异常关闭，那么broker会根据checkpoint文件进行异常恢复启动。
    checkpoint：只保存三个字段信息，physicMsgTimestamp，logicsMsgTimestamp，indexMsgTimestamp，当消息存储到commitlog文件后，都会相应的更新这三个字段的值。
    config： 该文件夹下保存四中文件，consumerOffset.json保存消息消费的逻辑offset，delayOffset.json保存延迟消息的offset，subscriptionGroup.json保存订阅关系，topics.json保存topic队列数量，权限等信息，并且每个文件都有一个备份。当broker启动时，先尝试加载*.json文件，不存在的情况下，则尝试加载*.json.bak文件。
    commitlog： 该文件夹下是存放所有消息的物理文件，默认每个文件1G，文件名长20位，即为消息起始的offset，第一个文件为0000000000000000000，第二个文件为00000000001073741824(1024 * 1024 * 1024)，依次类推。
    consumequeue：该文件夹下保存消息的逻辑队列文件，所谓的逻辑队列，是相对于物理文件commitlog而言，逻辑队列文件内保存物理消息offset，size，tagcode三个值。逻辑队列分为以%DLQ%开头的死信队列，以%RETRY%开头的重试队列，以SCHEDULE_开头的定时消息队列，和普通消息队列，每个队列下面按照队列数量(默认8个队列)，又有若干的队列文件，且每个文件存储30w数据。
    index： 该文件夹下保存消息的索引文件，命名方式以时间戳。

3. commitlog文件
    commitlog文件为消息的物理文件，由于消息的长度不定，所以该文件为顺序写随机读。一条消息由以下属性组成：


4. consumequeue文件

    逻辑队列存储了物理文件的三个值，即消息的offset，消息的大小size，消息tag的hashcode值，固定长度20字节，顺序写顺序读。根据topic区分，不同的消息存储到不同的逻辑队列下。broker先把消息保存到commitlog文件，然后根据消息属性，异步构建逻辑队列信息，如果当前消息为定时消息，则放到SCHEDULE_下，如果消息消费失败，保存到%RERTY%下，消费失败16次后，保存到%DLQ%下。

5. phyOffset和logicOffset对应关系
    在第二节介绍的topics.json文件中，强调该文件保存为消息的逻辑消费offset，即logicOffset，那么它与commitlog文件中实际的物理消息offset，即phyOffset的对应关系是怎么建立的呢？当broker保存消息到commitlog文件前，首先到TopicQueueTable(map结构)以topic_queueId为key查找逻辑offset，不存在，就设置key的逻辑offset为0，然后offset++。
假如，某条消息的物理offset可能为1032456，但是其逻辑offset可能为13，这样设计的一个好处就是减少网络传输大小。


6. 数据存储结构

    消息先保存到commitlog文件，通过ReputMessageService异步构建逻辑队列，IndexService构建索引文件等信息。

7. 刷盘
    broker通过FlushRealTimeService每隔1s进行commitlog文件刷盘检查，如果超过4页数据，即16k，就刷盘。每隔10s刷0页数据，即内存中有多少数据，就刷多少。所以，依此衡量broker丢失消息情形。同步和异步刷盘的区别是，异步刷盘写完PAGECACHE直接返回，而同步需要等待刷盘完成后才返回。不论同步还是异步刷盘，刷盘后都会保存消息时间到checkpoint的PhysicMsgTimestamp。
    通过FlushConsumeQueueService每隔1s进行consumequeue文件刷盘检查，如果超过2页数据，即8k，就刷盘。每隔1分钟刷0页数据，并保存消息时间到checkpoint的LogicsMsgTimestamp。


8. 清理文件
    A. commitlog文件删除
      源源不断的发送消息，消费消息，会产生大量的commitlog文件，MQ会定期清理过期的文件。由于消息存放在commitlog文件中大小不确定，如果按照消息进行删除的话，逻辑将异常的复杂，所以RocketMQ对于commitlog的删除是以文件为单位。通过deleteWhen属性进行配置删除文件时间，默认为4am。MQ会启动线程CleanCommitLogService进行commitlog文件的清理，执行删除动作判断条件：
     

• 到删除时间，磁盘超过指定使用率。
• 对于磁盘使用率，当超过75%或配置的rocketmq.broker.diskSpaceWarningLevelRatio(默认值为90%)或rocketmq.broker.diskSpaceCleanForciblyRatio(默认为85%)即满足条件。

    删除的文件对象：
     

• 超过fileReservedTime属性指定的时间，默认为48，即文件的修改时间超过48小时的commitlog文件。

    由于最大文件名的commitlog文件一直处于写状态，所以最后一个commitlog文件不在清除范围内，且每次批量删除对象不超过10个。

    B. consumequeue文件删除
      由于ConsumeQueue文件是基于commitlog文件进行构建的，当清理commitlog时，consumequeue也会跟着删除。由于commitlog的minOffset即为未删除文件中的最小文件名，所以consumequeue的删除对象即为小于minOffset的那些文件。Consumequeue的删除原理：
从后加载consumequeue文件，获取文件中最后一条消息的offset值，与minOffset相比，小于即表示该文件将被删除。

    C. index文件删除
      同样是根据commitlog的minOffset进行判断索引文件头中的endPhyOffset值，小于minOffset，即判断为需要删除的文件（如果当前仅有一个索引文件，不进行删除）。