RocketMQ观后感--Broker

小鱼儿 2022-05-27 04:43 235阅读 0赞

![这里写图片描述][SouthEast]

1.  BrokerStartup： RocketMQ在启动Broker时，通过”-c”来指定配置文件的位置，BrokerStartup的main方法中将properties类型的配置文件中的键值对提取出来，然后按键的名称和配置类的属性名称一致性原则注入到相应的配置类中，这些配置类有：BrokerConfig，NettyServerConfig，NettyClientConfig，MessageStoreConfig。这些Config中大部分属性都有默认值，我们的配置是去覆盖这些默认值，所以需要配置那些可以查看其源码，同时不要重复配置，尽量精简配置项。
2.  RocketMQ生成的配置文件或目录都是有统一的大小，名称格式，目录结构的，比如“\\store\\consumequeue” 目录代表着消费队列，其下一级别的目录名称代表着topic名称，再下一级代表着队列ID，其内部的文件才是此ConsumeQueue的数据文件，每个文件大小为 20 \* 300，000 字节，文件名称以 fileIndex(从0开始) \* fileSize 表示。
3.  RocketMQ生成的各种数据也都是有统一的格式，大小的。以ConsumeQueue里的为例，每个PositionInfo大小为20字节，前8个字节表示CommitLog Offset，中间4字节表示msg size，后8字节表示消息的tag的hashCode，每个ConsumeQueue文件能存放300,000个消息位置信息。CommitLog里存储消息也是严格遵循格式的，具体格式看上图。正是基于这种严格的数据写入机制，数据的读取才会变得很方便，知晓ConsumeQueue的Offset就能从CommitLog里查询出完整的消息数据。
4.  ScheduleMessageService： 消费者发回消息时，可以指定延迟级别，默认级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也就是说delayLevel = 3代表延迟10秒后重投递，最大重试次数16对应着2h后投递，每多消费一次投递时间就增长到下个阶段。当延迟级别delayLevel < 0 或超过最大消费次数时，放入Dead Letter Queue，topic名称格式为：%DLQ%+consumeGroup，默认queueId=0，死信队列只能写入，不能消费，这在创建topic时就指定的。
5.  ScheduleMessageService：**按顺序投递延时消息**。  
    Consumer消费消息时，如果返回RECONSUME\_LATER，或者主动的sendMessageBack(…，int delayLevel)时，会将消息发回给Broker，Broker对消息做个封装，指定topic为SCHEDULE\_TOPIC\_XXXX，QueudId=delayLevel-1，若未指定delayLevel，默认是ReConsumeTimes + 3，将封装后的消息存入CommitLog，ReputMessageService为其生成PositionInfo，tagsCode存储延时投递时间，存入”SCHEDULE\_TOPIC\_XXXX”的ConsumeQueue中。delayLevel有16个，因此最多情况下SCHEDULE\_TOPIC\_XXXX会有16个ConsumeQueue。Broker启动时，ScheduleMessageService会启动16个线程对应16个delayLevel的读取服务，有序的读取ConsumeQueue里的PositionInfo。ScheduleMessageService会在 \[当前时间<=延时投递时间\] 时从CommitLog中提取这消息，去除封装，抹去delayLevel属性，从新存入CommitLog，并马上更新延时投递偏移量dealyOffset。ReputMessageService再次为当前消息生成PositionInfo，因为不存在delayLevel，PositionInfo存入Topic为%RETRY%+consumeGroup，queueId为0的ConsumeQueue中。每个消费者在启动时都订阅了自身消费者组的重试队列，当重试队列里有位置信息时，拉取相应消息进行重新消费。消息的第一次重试会发回给原始的消费者(执行sendMessageBack的消费者)，之后的多次重试统一由订阅了QueueId = 0 的消费者消费。  
    ![RocketMQ消息重投递流程][RocketMQ]
6.  FlushRealTimeService(异步写)：每500ms对CommitLog进行一次Flush，当新写入数据超过16KB，或者距离上次Flush的时间间隔超过10S，将CommitLog位于内存中的数据同步到磁盘文件。ConsumeQueue和IndexFile的刷盘机制也类似，仅仅是刷盘周期，大周期和最小刷盘数据不同而已。因此Broker突然宕机时，会丢失少量数据。
7.  ReputMessageService：Broker在启动时，启动ReputMessageService，其每隔10ms为CommitLog里的消息生成位置信息PositionInfo，存入此消息对应的ConsumeQueue。一个ConsumeQueue对应着一个MessageQueue，Consume拉取消息时首先从ConsumeQueue里获取PositionInfo，对比tag是否匹配。匹配的话提取其在CommitLog的Offset，msg size，然后去CommitLog里查询消息。
8.  Consumer的每次消息拉取会指定一个进度，这个进度就是指ConsumeQueue里的进度，也就是从哪个位置开始提取PositionInfo，同时Broker会定期的持久化每个ConsumeQueue的消费进度。当消费请求提交的进度已经达到ConsumeQueue的最大值，也就是没有新的消息生成时，Broker挂起请求。每隔5S查看CnsumeQueue的maxOffset是否超过请求提交的Offset，如果是，则执行消息拉取。当挂起的时间超过15S时，强制结束请求。每当ConsumeQueue有新的PositionInfo生成时，匹配tag，若符合，立即唤醒消息拉取请求。执行拉取流程。
9.  ReputMessageService生成PositionInfo的同时，也会生成IndexInfo，也就是消息的索引信息，存放在Index目录下的IndexFile中。Producer在发送消息时，会为每条消息生成一个唯一的key，当然也可以自定义key。唯一的key会在发送结果中返回，用户可以使用这些key查询发送的消息。也就是说获取消息可以通过ConsumeQueue，也可以通过Index。
10. RocketMQ在接收消息时是多线程的，但向内存写入消息时是单线程的。不过因为借助虚拟内存映射技术，其写入性能非常高，且是顺序写入，不会出现内存碎片。多个线程竞争写入权限时使用自旋锁，因为写入时间非常短，自旋锁在此处比其他锁更适应，这也是RocketMQ高吞吐的原理之一。

[SouthEast]: /images/20220527/1ece361bea5e4b2cae48047e3433c6e2.png
[RocketMQ]: /images/20220527/43b7eae899744179895cb007dd55f93c.png