RocketMQ（五）——工作原理

消息的生产

消息的生产过程

Producer可以将消息写入到某Broker中的某Queue中，经历了如下过程：

Producer发送消息之前，会先向Name Server发出获取Topic的路由信息的请求
Name Server返回该Topic的路由表及Broker列表
Producer根据代码中指定的Queue选择策略，从Queue列表中选出一个队列，用于后续存储消息
Producer对消息做一些特殊处理，例如：消息超过4M，会对其进行压缩
Producer向选出的Queue所在Broker发出消息RPC请求，将消息发送到选择出的Queue

路由表：实际是一个Map，key是topic名称，value是一个Queue Data实例列表。Queue Data并不是一个Queue对应一个Queue Data，而是一个Broker中该Topic所有的Queue对应一个Queue Data。即只要是涉及到该Topic的Broker，一个Broker中该Topic的所有Queue对应一个Queue Data。Queue Data中包含Broker Name。简单来说，路由表的Key为Topic名称，value则为所有涉及该Topic的Broker Name 列表。

Broker 列表：实际也是Map。Key为broker name，value为Broker Data 。一套Broker Name 名称相同的Master-Slave集群对应一个Broker Data，Broker Data中包含Broker Name及一个map。该map的key为BrokerID，value为该Broker对应的地址。brokerID为0时表示该Broker 为Master。非0表示Slave

Queue算法

对于无序消息，其 Queue选择算法（消息投递算法），有两种较为常见方式：

轮询算法

默认选择该算法，该算法保证了每个Queue中都可以均匀获得消息

存在问题：

某些Broker上的Queue可能投递延迟较严重。从而导致 Producer的缓存队列中出现较大的消息积压，影响消息的投递性能。

最小投递延迟算法

该算法会统计每次消息投递的时间延迟，然后根据统计出的结果将消息投递到时间延迟最小的Queue。如果延迟相同，则采用轮询算法投递。该算法可以有效提升消息的投递性能。

存在问题：

消息在Queue上的分配不均匀。投递延迟小的Queue其可能会存在大量的消息。而对该Queue的消费者压力会增大，降低消息的消费能力，可能会导致MQ中消息的堆积

消息的存储

RocketMQ中的消息存储在本地文件系统中，这些相关文件默认在当前用户主目录下的store目录中。

文件说明：

abort：该文件在Broker启动后会自动创建，正常关闭Broker，该文件会自动消失。若在没有启动
Broker的情况下，发现这个文件是存在的，则说明之前Broker的关闭是非正常关闭。
checkpoint：其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳
commitlog：其中存放着commitlog文件，而消息是写在commitlog文件中的。注意：在源码中，该文件被命名为mappedFile
config：存放着Broker运行期间的一些配置数据
consumequeue：其中存放着consumequeue文件，队列就存放在这个目录中
index：其中存放着消息索引文件indexFile
lock：运行期间使用到的全局资源锁

Commitlog

该目录中存放这很多mappedFile 文件，当前Broker中所有消息都是落盘到这些mappedFile文件中的。mappedFile文件大小为1G，文件名有20位十进制数构成，表示当前文件的第一条消息的起始位置偏移量。
第一个文件名一定是20个0，即第一个文件的第一条消息的偏移量commitlog offset为0
当第一个文件放满时，则会自动生成第二个文件继续存放消息。
一个Broker中所有的mappedFile文件的commitlog offset是连续的
一个Broker仅包含一个commit log目录，所有的mappedFile文件都是存放在该目录中的。也就是说，无论当前Broker中存放这多少Topic的消息，这些消息都被顺序写入到了mappedFile文件中的，这些消息在Broker中存放时并没有按照Topic进行分类存放。

消息单元

mappedFile文件内容由一个个消息单元构成，每个消息单元中包含 MsgLen（总长度），Physical Offset（物理位置），Body（消息体），BodyLength（消息体长度），Topic（主题），TopicLength（主题长度），Born Host（消息生产者），BornTimestamp（消息发送时间戳），QueueID（消息所在队列ID），QueueOffset（消息在Queue中存储的偏移量）等消息

consume queue

为了提高效率，会为每个Topic在~/store/consumequeue中创建一个目录，目录名为Topic名称。在该Topic目录下，会再为每个该Topic的Queue建立一个目录，目录名为queueId。每个目录中存放着若干consumequeue文件，consumequeue文件是commitlog的索引文件，可以根据consumequeue定位到具体的消息。
consumequeue文件名也由20位数字构成，表示当前文件的第一个索引条目的起始位移偏移量。与mappedFile文件名不同的是，其后续文件名是固定的。因为consumequeue文件大小是固定不变的。

索引条目

每个consumequeue文件可以包含30万个索引条目，每个索引条目包含了三个消息重要属性：消息在mappedFile文件中的偏移量CommitLog Offset、消息长度、消息Tag的hashcode值。这三个属性占20个字节，所以每个文件的大小是固定的30万 * 20字节。
一个consumequeue文件中所有消息的Topic一定是相同的。但每条消息的Tag可能是不同的。

对文件的读写

消息写入

一条消息进入到Broker后经历了以下几个过程才最终被持久化。

Broker根据queueId，获取到该消息对应索引条目要在consumequeue目录中的写入偏移量，即QueueOffset
将queueId、queueOffset等数据，与消息一起封装为消息单元
将消息单元写入到commitlog，同时，形成消息索引条目
将消息索引条目分发到相应的consumequeue

消息获取

当Consumer来拉取消息时会经历以下几个步骤

Consumer获取到其要消费消息所在Queue的消费偏移量offset，计算出其要消费消息的消息offset
Consumer向Broker发送拉取请求，其中会包含其要拉取消息的Queue、消息offset及消息Tag。
Broker计算在该consumequeue中的queueOffset。
从该queueOffset处开始向后查找第一个指定Tag的索引条目
解析该索引条目的前8个字节，即可定位到该消息在commitlog中的commitlog offset
从对应commitlog offset中读取消息单元，并发送给Consumer

性能提升

RocketMQ对文件的操作是通过mmap零拷贝实现的，对文件的操作转化为对内存地址的直接操作，极大提高了读写效率
consumequeue中的文件是顺序存放的，硬盘顺序读写速度要高于随机读写。并且还引入了PageCache页缓存机制，使得对consumequeue的读取几乎接近内存读取速度。