RocketMQ源码以及关键流程分析

OffsetStore接口核心方法

定时持久化位点org.apache.rocketmq.client.impl.factory.MQClientInstance#startScheduledTask,
定时持久化位点逻辑是通过定时任务来实现的，在启动程序10s后，会定时调用持久化方法，
MQClientInstance.this.persistAllConsumerOffset(),持久化每一个消费者消费的每一个MessageQueue的进度

不定时持久化也叫Pull-And-Commit，也就是在执行Pull方法同时，把队列最新消费位点信息发给Broker，
org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage(),该方法有两处持久化位点消息

理论上位点信息越是及时上报Broker，越能减少消息重复的可能性，RocketMQ再设计时并不完全支持Exactly-Once类型的幂等语义，
因为实现该语义的代价颇大，并且使用该场景极少，再加上用户侧实现幂等的代价更小，故而RocketMQ在设计时将幂等操作交由用户处理

ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable:记录MessageQueue和ProcessQueue的关系，
MessageQueue可以简单地理解为ConsumeQueue的客户端实现；ProcessQueue是保存Pull消息的本地容器

ConcurrentMap<String, Set<MessageQueue>> topicSubscribeInfoTable:Topic路由信息。保存Topic和MessageQueue的关系

ConcurrentMap<String(Topic), SubscriptionData> subscriptionInner:真正的订阅关系，保存当前消费者组订阅了哪些Topic的哪些Tag

AllocateMessageQueueStrategy allocateMessageQueueStrategy:MessageQUeue消息分配策略的实现

MQClientInstance mQClientFactory:client实例对象

boolean lock():为MessageQueue加锁

void doRebalance():执行Rebalance操作

void messageQueueChanged():通知Message发生变化，这个方法在Push和Pull两个类中被重写

boolean removeUnnecessaryMessageQueue():去掉不再需要的MessageQueue

void dispatchPullRequest():执行消息拉取请求

boolean updateProcessQueueTableInRebalance():在Rebalance中更新processQueue

消费者实例在收到Broker通知后是怎么执行Reblance的？这个操作是通过调用
MQClientInstance.rebalanceImmediately()来实现的

这种设计是RocketMQ种典型的锁方式，执行wakeup命令后，this.waitForRunning()就会暂停，
再执行this.mqClientFactory.doRebalance()

doRebalance()方法主要有以下几个步骤
1.查找当前clientId对应的全部的消费者组，全部执行一次Rebalance.
虽然消费者实现分别为Pull消费和Push消费两种默认实现，调用的是不同实现类的Rebalance方法,
但是实现逻辑都差不多

2.判断Rebalance开关，如果没有被暂停，则调用RebalancePushImpl.rebalance()方法

3.在RebalancePushImpl.rebalance()方法中,获取当前消费者全部订阅关系中的Topic,
循环对每个Topic进行Rebalance.待全部的Rebalance都执行完之后，将不属于当前
消费者的队列删除

4.Topic队列重新分配，这里也就是客户端Rebalance的核心逻辑之处，根据是集群消费还是广播消费分别执行
MessageQueue重新分配的逻辑，以集群消费为例分析

第一步:用户发送一个带Tag的消息

第二步:用户订阅一个Topic的Tag,RocketMQ Broker会保存订阅关系

第三步:在Broker端做Tag过滤。消费者在Pull消息时，RocketMQ Broker会根据Tag的HasCode进行对比，
如果不满足条件，消息不会返回给消费者，以节约带宽
也许你们会问，为什么不直接用字符串进行对比和过滤呢?原因是HashCode对比存在Hash碰撞而导致过滤失败，
字符串比较的速度相较HashCode慢。HashCode对比是数字比较，Java底层可以直接通过位运算进行对比，
而字符串对比需要按照字符顺序比较，相比位运算更加耗时。由于HashCode对比有Hash碰撞的危险，所以才
引出第四步

第四步:客户端Tag过滤。Hash碰撞相信大家都有所了解，就是不同的Tag计算出来的Hash值可能是一样的，
在这种情况下过滤的消息是错误的，所以RocketMQ设计了客户端字符串对比功能，用来做第二次Tag过滤

Tag过滤为什么设计成Broker端使用Hash过滤，而客户端使用Tag字符串进行对比过滤呢？
Broker端使用Hash过滤可以快速过滤海量消息，即使偶尔有"漏网之鱼"，在客户端字符串
过滤后也能被成功过滤。这种层次设计 的过滤方式在做系统时可以参考

第一步:消费订阅Topic,上传过滤SQL语句，RocketMQ Broker编译SQL保存

第二步:消费者Pull消息

第一次过滤:使用Bloom过滤器的isHit()方法做第一次过滤。Bloom过滤器效率高，但是也存在
缺陷，即只能判断不需要的消息，过滤后的消息也不保证都是需要消费的。

第二次过滤:执行编译后的SQL方法evaluate()即可过滤出最终的结果
在使用SQL过滤前，需要在启动Broker时配置如下几个参数:
enableConsumeQueueExt=true
filterSupportRetry=true
enablePropertyFilter=true
enableCalcFilterBitMap=true

这是一种不常用但是非常灵活的过滤方式，要使用Filter Server过滤必须在启动Broker时，添加如下配置:
filterServerNums=大于0的数字.这样就可以启动一个或多个过滤服务器，每个过滤服务在启动时会自动
注册到Namesrv中

第一步:用户消费者从Namesrv获取Topic路由信息，同时上传自定义的过滤器实现类源代码到FilterServer中，
FilterServer编译并实例化过滤器类

第二步:用户发送拉取消息请求到FilterServer，FilterServer通过Pull consumer从Broker拉取消息，
执行过滤类中的过滤方法，返回过滤后的消息

生产者发送事务消息和执行本地事务。
发送过程分为两个阶段:
第一阶段,发送事务消息
第二阶段,发送endTransaction消息
事务消息发送过程的实现类TransactionMQProducer，该类继承鱼DefaultMQProducer,
不仅能发送事务消息，还能发送其他消息。虽然4.2.0版本有事务消息代码，但实际是4.3.0
版本才全面支持事务消息。

Broker存储事务消息。
在Broker中，事务消息的初始化是通过BrokerController.initialTransaction()方法执行的。
3个核心的初始化变量

Broker回查事务消息。
如果用户由于某种原因，在第二阶段中没有将endTransaction消息发送给Broker,
Broker的Half消息又将如何处理。
RocketMQ在设计时已经考虑到这个问题，通过"回查机制"处理第二阶段既未发送Commit
也没有发送Rollback的消息。回查是Broker发起的，Broker认为在接收Half消息后的一段时间内，
如果生产者都没有发送Commit或Rollback消息给Broker，那么Broker会主动"询问"生产者该
事务消息对应的本地事务执行结果，以此来决定事务是否要Commit.
TransactionalMessageCheckService是回查服务的实现类

Broker提交或回滚事务消息。
当生产者本地事务处理完成并且Broker回查事务消息后，不管执行Commit还是Rollback,
都会根据用户本地事务的执行结果发送一个End_transaction的RPC请求给Broker，Broker
端处理该请求的类是EndTransactionProcessor

DefualtMQPullConsumer的继承关系

核心属性

核心方法

Pull消费启动流程

核心属性和方法

Push消费启动流程

Pull方式。
用户主动Pull消息，自主管理位点，可以灵活地掌控消费进度和消费速度，适合流计算、消费特别耗时等特殊的消费场景。
缺点也显而易见，需要从代码层面精准地控制消费，对开发人员有一定要求，再RocketMQ中
org.apache.rocketmq.client.consumer.DefaultMQPullConsume是默认的Pull消费者实现类

Push方式。
代码介入非常简单，适合大部分业务场景。缺点灵活度差，在了解消费原理后，排查消费问题可简单快捷.
在RocketMQ中org.apache.rocketmq.client.consumer.DefaultMQPushConsumer是默认的Push消费者实现类

Pull和Push方式的比较

RocketMQ是一个消息队列，FIFO先进先出规则如何再消费失败时保证消息的顺序呢?
可以从消费任务实现类ConsumeRequest和本地缓存队列ProcessQueue的设计来看主要差异

public class Message implements Serializable {
    private static final long serialVersionUID = 8445773977080406428L;
    private String topic;
    private int flag;
    private Map<String, String> properties;
    private byte[] body;
    public void setKeys(String keys);
    public void setKeys(Collection<String> keyCollection);
    public void setTags(String tags);
    public void setDelayTimeLevel(int level);
    public void setTopic(String topic);
    public void putUserProperty(final String name, final String value);
}

putUserProperty():如果还有其他扩展信息，可以存放在这里，内部是一个Map,重估调用会覆盖旧值

setDelayTimeLevel():设置延迟级别，延迟多久消费者可以消费

setTags():消息过滤的标记，用户可以订阅某个Topic的某些Tag，这样Broker只会把订阅了topic-tag的消息发送给消费者

setKeys():设置消息的key，多个key可以用MessageConst.KEY_SEPARATOR(空格)分隔或者直接用另一个重载方法。
如果Broker中的messageIndexEnable=true则会根据key创建消息的hash索引，帮助用户快速过滤

Body:消息体，字节数组，需要注意生产者使用什么编码，消费者也必须使用相同编码节码，否则会产生乱码

Properties:消息扩展信息，Tag、Keys、延迟级别都保存在这里

Flag：目前没用

Topic:主题名字，可以通过RocketMQ Console创建

生产者启动的流程比消费者启动的流程更加简单一般用户使用DefaultMQProducer的
构造函数构造一个生产者实例，并设置各种参数，比如，Namesrv地址、生产者组名等，
调用start()方法启动生产者实例，start()方法调用了生产者默认实现类的start()方法启动，
这里我们主要分析start()方法内部是怎么实现的 

1.构造函数初始化实例

2.启动生产者实例start()

3.CREATE_JUST。通过switch-case判断当前生产者的服务状态，创建时默认是CREATE_JUST,
设置默认启动状态为启动失败

4.将生产者置为启动失败

5.生产者参数校验，执行checkConfig()方法，校验生产者实例设置的各种参数，
比如生产者组名是否为空，是否满足命名规则，长度是否满足等等

6.默认生产者组名转换为进行id,执行changeInstanceNameToPID()方法，
校验instance name，如果是默认名字则将其修改为进程id.接着调用getOrCreateMQClientInstance()
创建一个MQClientInstance实例，它与ClientId是一一对应，clientId是由ClientIP、instance及unitName
构成的，一般来讲，为了减少客户端的使用资源，如果将所有的instanceName和unitName设置为同样的值，
就只会创建一个MQClientInstance实例，而MQClientInstance实例的功能是管理本实例中全部的生产者与消费者的生产和消费行为。
同时将调用registerProducer()将生产者放入组中

7.注册本地路由信息

8.启动MQClientInstance，置为服务启动失败状态

9.如果没有配置Namesrv,则调用fetchNameServerAddr()远程获取

10.启动通信模块.this.mQClientAPIImpl.start();

11..启动各种定时任务。this.startScheduledTask();

12.启动消息拉取服务。this.pullMessageService.start();

13.启动负载均衡服务。this.rebalanceService.start();

14.启动push服务，this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

15.置为服务启动状态。this.serviceState = ServiceState.RUNNING;

业务层:通常指直接调用RocketMQClient发送API的业务代码

消息处理层:指RocketMQ Client获取业务发送的消息对象后，一系列的参数检查、消息发送准备、参数包装等操作

通信层：指RocketMQ基于Netty封装的一个RPC通信服务，RocketMQ的各个组件之间的通信全部使用该通信层

首先RocketMQ客户端接收业务层消息，然后通过DefaultMQProducerImpl发送一个RPC请求给Broker,
再由Broker处理请求并保存消息,以DefualtMQProducerImpl.send()接口为例，消息发送流程具体分为3步

Namesrv核心数据结构和API.
Namesrv中保存的数据被称为Topic路由信息，Topic路由决定了Topic消息发送到哪些Broker,
消费者从哪些Broker消费消息，那么路由信息都包含哪些数据呢?
路由数据结构的实现代码都在RouteInfoManager类中

Namesrv支持的全部API在DefaultRequestProcessor类中

曾几何时，RocketMQ也采用Zookeeper作为协调者，但是繁杂的运行机制和过多的依赖导致RocketMQ
最终完全重新开发了一个零依赖、更简洁的Namesrv来替换

第一步:脚本和启动参数配置。
启动命令 nohup ./bin/mqnamesrv -c ./conf/namesrv.conf > dev/null 2>&1 &
通过脚本配置启动基本参数，比如配置文件路径、JVM参数，调用NamesrvStartup.main()方法，解析命令行的参数，
将处理好的参数转化为Java实例，传递给NamesrvController实例

第二步:new 一个NamesrvController，加载命令行传递的配置参数，调用controller.initialize()方法初始化NamesrvController。
Namesrv启动的主要初始化过程也在这个方法中，代码如图
1.加载KV配置。主要是从本地文件中加载KV配置到内存中
2.初始化Netty通信层实例。RocketMQ基于Netty实现了一个RPC服务端，即NettyRemotingServer.通过参数nettyServerConfig,
会启动9876端口监听
3.Namesrv主动检测Broker是否可用，如果不可用就剔除。生产者、消费者也能通过心跳发现被踢出的路由，从而感知Broker下线
4.Namesrv定时打印配置信息到日志中。

第三步:NamesrvController在初始化后添加JVM Hook.Hook中会调用NamesrvController.shutdown()方法来关闭整个Namesrv服务

第四步:调用NamesrvController.start()方法，启动整个Namesrv。其实start()方法只启动了Namesrv接口处理线程池

为什么需要了解停止流程呢?RocketMQ在设计之初已经考虑了很多异常情况，比如Namesrv异常退出、
突然断电、内存被打满等等，只有了解了正常的停止流程才能对异常退出导致的问题进行精确的分析和排障。

通常Namesrv的停止是通过关闭命令./mqshutdown namesrv来实现的。这个命令通过调用kill命令将关闭进程
通知发给JVM,JVM调用观级Hook执行停止逻辑。具体实现如下

路由注册。
registerBrokerWithFilterServer()方法中的this.namesrvController.getRouteInfoManager.registerBroker()方法，
该方法的主要功能是将request解码为路由对象，保存在Namesrv中。
在路由信息注册完成后，Broker会每隔30s发送一个注册请求给集群中全部的Namesrv，俗称心跳信，会把最新的
Topic路由信息注册到Namesrv中

路由剔除
如果Broker长久没有心跳或者宕机，那么Namesrv会将这些不提供服务的Broker剔除，同时生产者和消费者与Namesrv
心跳后也会感知被踢掉的Broker,如此Broker扩容或者宕机对生产、消费无感知的情况就处理完了
Namesrv有两种剔除Broker的方式

第一步：初始化启动环境。
这是由./bin/mqbroker和./bin/runbroker.sh两个脚本来完成的，/bin/mqbroker脚本主要用于设置RocketMQ根目录环境变量
if [ -z "$ROCKETMQ_HOME" ] ; then
 ....
fi
export ROCKETMQ_HOME
sh ${ROCKETMQ_HOME}/bin/runbroker.sh org.apache.rocketmq.broker.BrokerStartup $@
./bin/runbroker.sh脚本的主要功能是检测JDK的环境配置和JVM的参数配置。JDK的环境配置的检查逻辑的实现代码如下:
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"

export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export CLASSPATH=.:${BASE_DIR}/conf:${BASE_DIR}/lib/*:${CLASSPATH}
下面是JVM的参数配置，通常-Xms -Xmx -Xmn -XX:MaxDirectMemorySize这四个参数会随着部署RocketMQ服务器的物理内存大小的变化
而进行相应的改变
choose_gc_options()
{
    JAVA_MAJOR_VERSION=$("$JAVA" -version 2>&1 | head -1 | cut -d'"' -f2 | sed 's/^1\.//' | cut -d'.' -f1)
    if [ -z "$JAVA_MAJOR_VERSION" ] || [ "$JAVA_MAJOR_VERSION" -lt "8" ] ; then
      JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
    else
      JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0"
    fi

    if [ -z "$JAVA_MAJOR_VERSION" ] || [ "$JAVA_MAJOR_VERSION" -lt "9" ] ; then
      JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
      JAVA_OPT="${JAVA_OPT} -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
    else
      JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0"
      JAVA_OPT="${JAVA_OPT} -Xlog:gc*:file=${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log:time,tags:filecount=5,filesize=30M"
    fi
}

choose_gc_log_directory

JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g"
choose_gc_options
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -XX:+AlwaysPreTouch"
JAVA_OPT="${JAVA_OPT} -XX:MaxDirectMemorySize=15g"
JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages -XX:-UseBiasedLocking"
#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"

第二步:初始化BrokerController
该初始化主要包含ROcketMQ启动命令行参数解析、Broker各个模块配置参数解析、Broker各个模块初始化、进程关机Hook初始化等过程.

RocketMQ启动命令行参数解析。其代码在BrokerStartup.createBrokerController()方法中，RocketMQ的启动参数支持启动命令指定，
也可以在配置文件中进行配置。通常，命令行参数的优先级大于配置文件。
通过第三方库将命令行输入参数解析为commandLine对象，再获取输入参数值。命令行参数的启动比较简单，如果大量的RocketMQ配置
项放在启动命令中，就会导致启动命令较长，难以维护，一般推荐启动RocketMQ使用配置文件的方式。配置文件在createBrokerController()
方法中被解析的代码如图所示
在brokerConfig、nettyServerConfig、nettyClientConfig、messageStoreConfig这些基本配置对象初始化完毕后，还有后续代码依据
各种启动条件重新调整部分参数。在各个配置对象初始化完毕后，程序会调用BrokerController.initialize()方法对Broker的各个模块进行
初始化

xxxConfigManager.load()方法的功能是加载Broker基础数据配置，包含Broker中的Topic、消费位点、订阅关系、消费过滤(无实际数据需要加载).
这些配置加载成功后，初始化存储层服务对象messageStore和Broker监控统计对象brokerStats.
然后，Broker会初始化通信层服务和一些列定时任务，通信层服务主要初始化正常通信通道、VIP通信通道和通信线程池。
这里以VIP通道为例，分析通信层服务初始化，以消费进度定时持久化为例，分析定时任务初始化。
fastConfig就是VIP通信层的配置，其配置对象"克隆"自正常通信的配置对象，唯独通信端口是nettyServerConfig.getListenPort()-2,
也就是10911-2.利用fastConfig初始化fastRemotingServer的结果也就是我们常用的VIP通道.
从fastConfig和fastRemotingServer的实现类命名来看，RocketMQ的通信层实现本质上是基于Netty的，那么通信层又是如何处理客户端
发送的Netty请求的呢?
通信层对象初始化完成后，会调用this.registerProcessor()方法，这里将正常的通信层对象和VIP通道的通信层对象与各个请求处理器进行
关联，比如将发送消息的请求交给接收消息的请求处理器进行处理

消费进度定时持久化。
Broker在接收到消费者上报的消费进度后，会定期持久化到物理文件中，当消费者因为重新发布或者宕机而重启时，能从消费进度中得知恢复，
不至于重复消费，持久化周期可以通过参数flushConsumerOffsetInterval(以ms为单位)进行配置

第三步，启动RocketMQ的各个组件
组件启动在BrokerController.start()方法中

存储概述

1.Broker存储流程
RocketMQ首先将消息数据写入操作系统PageCache，然后定时将数据刷入磁盘。
接下来主要分析RocketMQ是如何接收发送消息请求并将消息写入PageCache的，
整个过程如图

2.内存映射机制与高效磁盘。
RocketMQ在存储涉及中通过内存映射、顺序写文件等方式实现了高吞吐。
RocketMQ的基本数据结构:
CommitLog:RocketMQ对存储消息的物理文件的抽象实现，也就是对物理CommitLog文件的具体实现。
MappedFile:CommitLog文件在内存中的映射文件，映射文件同时具有内存的写入速度和与磁盘一样可靠的
持久化方式.
MappedFileQueue:映射文件队列中有全部的CommitLog映射文件，第一个映射文件为最先过期的文件，
最后一个文件是最后过期的文件，最新的消息总是写入最后一个映射文件。

3.文件刷盘机制
消息存储完成后，会被操作系统持久化到磁盘，也就是刷盘。
RocketMQ支持两种刷盘方式，在Broker启动时配置flushDiskType=SYNC_FLUSH表示同步刷盘.
配置flushDiskType=ASYNC_FLUSH表示异步刷盘
刷盘涉及以下3个线程服务。如图所示

3.1同步刷盘和异步刷盘

Master-Slave读写分离机制。通常Master提供读写处理，如果Master负载较高就从Slave读取

Direct Memory-Page Cache的读写分离机制
以上逻辑通过MappedFile.appendMessagesInner()方法来实现，核心代码如图

索引的数据结构

索引的构建过程

索引如何使用？

1.CommitLog文件的删除过程

ConsumeQueue、IndexFile文件的删除过程。
ConsumeQueue和IndexFile都是索引文件，在CommitLog文件被删除后，对应的
索引文件起始没有存在的意义，并且占用磁盘空间，所以这些文件应该被删除。
RocketMQ的删除策略是定时检查，满足删除条件时会删除过期或者无意义的文件。
最终程序调用CleanConsumeQueueService.deleteExpiredFiles()方法来删除索引文件

主从同步概述
Broker有两种角色Master和Slave.Master主要用于处理生产者、消费者的请求和存储数据。
Slave从Master同步所有数据到本地，具体作用体现在两个方面。
第一，Broker服务高可用。一般生产环境会部署两个主Broker节点和两个从Broker节点(也叫2m2s),
一个Master宕机后，另一个Master可以接管工作；如果两个Master都宕机，消费者可以通过连接
Slave继续消费。这样可以保证服务的高可用
第二，提高服务性能。如果消费从Master Broker拉取消息时，发现拉取消息的offset和commitLog的
物理offset相差太多，会转向Slave拉取消息，这样可以减轻Master的压力，从而提高性能
Broker同步数据的方式有两种:同步复制、异步复制。

主从同步流程

概述。
Broker关机恢复是指恢复CommitLog、Consume Queue、Index File等数据文件。
Broker关机分为正常调用命令关机和异常被迫进程终止关机两种情况。恢复过程的设计目标是
使正常停止的进程实现零数据丢失，异常停止的进程实现最少量的数据丢失，与关机恢复相关
的主要文件有两个:abort和checkpoint.

Broker关机恢复流程。
Broker在启动时会初始化abort、checkpoint两个文件。正常关闭进程时会删除abort文件，
将checkpoint文件刷盘；异常关闭时，通常来不及删除abort文件。由此，在重新启动Broker时
会根据abort判断是否需要异常停止进程，而后恢复数据。
Broker启动时，会启动存储服务DefaultMessageStore.存储服务在初始化时执行load方法加载
全部数据，这里主要分析数据加载流程。Broker关机的恢复过程可以分为以下几步.

核心属性

核心方法

msg.getDelayTimeLevel()是发送消息时可以设置的延迟级别，如果该值大于0，则表示
当前处理的消息是一个延迟消息，将对该消息做如下修改:

1.将原始Topic、queueId备份在消息的扩展字段中，全部的延迟消息都保存在
SCHEDULE_TOPIC的Topic中
2.备份原始Topic、queueId为延迟消息的Topic、queueId。备份的目的是当消息到达
投递时间时会恢复原始的Topic和queueId，继而被消费者拉取并消费

经过处理后，该消息会被正常保存到CommitLog中，然后创建ConsumeQueue和IndexFile两个索引。
在创建ConsumeQueue时，从CommitLog中获取的消息内容会单独进行处理，单独处理的逻辑方法是
CommitLog.checkMessageAndReturnSize().

有一个很精巧的设计:在CommitLog中查询出消息后，调用computeDeliverTimestamp()方法计算消息
具体的投递时间，再将该时间保存在ConsumeQueue的tagCode中。

这样设计的好处是，不需要检查CommitLog大文件，在定时任务检查消息是否需要投递时，只需要检查
ConsumeQueue中的tagCode(不再是Tag的Hash值，而是消息可以投递的时间，单位是ms)，如果满足
条件再通过查询CommitLog将消息投递出去即可，如果每次都查询CommitLog，那么可想而知，效率会很低

核心字段和方法