RocketMQ Broker定时任务总结

比眉伴天荒 2022-05-29 07:05 354阅读 0赞

1.BrokerStartup： RocketMQ在启动Broker时，通过"-c"来指定配置文件的位置，BrokerStartup的main方法中将properties类型的配置文件中的键值对提取出来，然后按键的名称和配置类的属性名称一致性原则注入到相应的配置类中，这些配置类有：BrokerConfig，NettyServerConfig，NettyClientConfig，MessageStoreConfig。这些Config中大部分属性都有默认值，我们的配置是去覆盖这些默认值，所以需要配置那些可以查看其源码。

2.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。

3.ScheduleMessageService：**按顺序投递延时消息**。  
Producer发送消息是，指定delayLevel；或者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]

4.ScheduleMessageService：每隔10S持久化每个延时队列的投递进度

"offsetTable":{3:4,4:4,5:4}

5.ConsumerOffsetManager：Broker每隔5S持久化消费进度,将 ConsumerOffsetManager\#offsetTable 属性序列化到consumerOffset.json 文件，以覆盖的形式重新写入，offsetTable 是一个Map类型的属性，key 是：topic@consumeGroup ，value是每个ConsumeQueue的消费进度，也是一个集合，key是id，value是offset，

"offsetTable":{
    		"test1@rmq-group":{0:6,1:6,2:7,3:8
    		},
    		"%RETRY%rmq-group@rmq-group":{0:0
    		}
     }

6.FlushConsumeQueueService： 每隔1S执行刷新ConsumeQueue，当某个ConsumeQueue新写入的数据超过2页(8kb)，强制Flush数据至磁盘；同时每隔60S对所有的ConsumeQueue执行一次flush，不管新写入数据量

7.ReputMessageService：每隔1ms进行一次Reput工作，将新消息的位置信息存入ConsumeQueue，key信息存入IndexFile，同时唤醒那些订阅了新消息所属队列的消费者请求，让它们执行消息的拉取工作

8.CommitRealTimeService(开启写入缓冲池)：将缓冲池中的数据Commit到CommitLog的FileChannel中

9.FlushRealTimeService(异步写)：每500ms对CommitLog进行一次Flush，当新写入数据超过16KB，或者距离上次Flush的时间间隔超过10S，将CommitLog位于内存中的数据同步到磁盘文件

10.CleanCommitLogService：每隔10S执行一次清理失效CommitLog日志文件，默认清理72h之前的

11.CleanConsumeQueueService：每隔10S执行一次清理失效ConsumeQueue和IndexFile文件

12.PullRequestHoldService：持有针对每个ConsumeQueue的消息PullRequest，每隔5S，根据条件：maxOffset > pullFromOffset 来确定是否要唤醒订阅相应ConsumeQueue的PullRequest

13.ClientHousekeepingService：每隔10S扫描持有的ProducerChannel，ConsumerChannel，FilterChannel，将那些超过2m没有发送心跳的连接关闭掉

14.每隔30S向指定的一个或多个Namesrc注册 Broker信息

15.HAService：负责Master与Slave间的通信及消息的同步

[RocketMQ]: /images/20220529/bebed7e1fed4458e87c81ebc48f19803.png