今天来分享 RocketMQ 的定时任务。通过这些定时任务能让我们更加理解 RocketMQ 的消息处理机制和设计理念。从 RocketMQ 4.9.4 的源代码上看RocketMQ 的定时任务有很多今天主要讲解一些核心的定时任务。1 架构回顾首先再来回顾一下 RocketMQ 的架构图Name Server 集群部署但是节点之间并不会同步数据因为每个节点都会保存完整的数据。因此单个节点挂掉并不会对集群产生影响。Broker 可以采用主从集群部署实现多副本存储和高可用。每个 Broker 节点都要跟所有的 Name Server 节点建立长连接定义注册 Topic 路由信息和发送心跳。Producer 和 Consumer 跟 Name Server 的任意一个节点建立长连接定期从 Name Server 拉取 Topic 路由信息。2 Producer 和 Consumer2.1 获取 NameServer 地址Producer 和 Consumer 要跟 Name Server 建立连接就必须首先获取 Name Server 地址。Producer 和 Consumer 采用定时任务每两分钟获取 Name Server 地址并更新本地缓存。代码如下//MQClientInstance类 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr(); } catch (Exception e) { log.error(ScheduledTask fetchNameServerAddr exception, e); } } }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);2.2 更新路由信息Producer 和 Consumer 会定时从 Name Server 获取定时订阅信息更新本地缓存默认间隔是 30s可以配置。代码如下//MQClientInstance类 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { MQClientInstance.this.updateTopicRouteInfoFromNameServer(); } catch (Exception e) { log.error(ScheduledTask updateTopicRouteInfoFromNameServer exception, e); } } }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);2.3 向 Broker 发送心跳Producer 和 Consumer 会从本地缓存的 Broker 列表中定时清除离线的 Broker并且向 Broker 发送心跳默认间隔是 30s可以配置。代码如下//MQClientInstance类 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { MQClientInstance.this.cleanOfflineBroker(); MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); } catch (Exception e) { log.error(ScheduledTask sendHeartbeatToAllBroker exception, e); } } }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);2.4 持久化 Offset消费者需要定时持久化 MessageQueue 的偏移量默认每 5s 更新一次可以配置。注意集群模式需要向 Broker 发送持久化消息因为集群模式偏移量保存在 Broker 端而广播模式只需要把偏移量保存在消费者本地文件。代码如下//MQClientInstance类 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { MQClientInstance.this.persistAllConsumerOffset(); } catch (Exception e) { log.error(ScheduledTask persistAllConsumerOffset exception, e); } } }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);2.5 调整核心线程数对于消费者采用推模式的情况消费者会根据未消费的消息数量定期更新核心线程数默认每 1m 一次。注意在 4.9.4 这个版本更新核心线程数的代码并没有实现只是预留了接口。代码如下//MQClientInstance类 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { MQClientInstance.this.adjustThreadPool(); } catch (Exception e) { log.error(ScheduledTask adjustThreadPool exception, e); } } }, 1, 1, TimeUnit.MINUTES);2.6 失效过期请求Producer 和 Consumer 会定时扫描缓存在本地的请求如果请求开始时间加超时时间再加 1s小于当前时间则这个请求过期。通过定时任务3s 一次让过期请求失效并且触发回调函数。//NettyRemotingClient.java this.timer.scheduleAtFixedRate(new TimerTask() { Override public void run() { try { NettyRemotingClient.this.scanResponseTable(); } catch (Throwable e) { log.error(scanResponseTable exception, e); } } }, 1000 * 3, 1000);2.7 生产者2.7.1 性能记录生产者发送消息后会对成功失败的状态、花费时间进行记录以此来计算吞吐量 TPS响应时间 RT代码如下//Producer.java executorService.scheduleAtFixedRate(new TimerTask() { Override public void run() { snapshotList.addLast(statsBenchmark.createSnapshot()); if (snapshotList.size() 10) { snapshotList.removeFirst(); } } }, 1000, 1000, TimeUnit.MILLISECONDS); executorService.scheduleAtFixedRate(new TimerTask() { private void printStats() { if (snapshotList.size() 10) { doPrintStats(snapshotList, statsBenchmark, false); } } Override public void run() { try { this.printStats(); } catch (Exception e) { e.printStackTrace(); } } }, 10000, 10000, TimeUnit.MILLISECONDS);2.8 消费者2.8.1 MessageQueue 加锁对于顺序消息要保证同一个 MessageQueue 只能被同一个 Consumer 消费。消费者初始化的时候会启动一个定时任务定时默认 20s可以配置地向 Broker 发送锁定消息Broker 收到请求后就会把 MessageQueue、group 和 clientId 进行绑定这样其他客户端就不能从这个 MessageQueue 拉取消息。代码如下//ConsumeMessageOrderlyService.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { ConsumeMessageOrderlyService.this.lockMQPeriodically(); } catch (Throwable e) { log.error(scheduleAtFixedRate lockMQPeriodically exception, e); } } }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);注意Broker 的加锁是有时效的默认 60s可以配置过期后有可能被其他 Consumer 进行消费。2.8.2 性能快照Consumer 每秒会记录一次性能快照比如消息从创建到消费花费的时间消息从保存到消费花费的时间接收到消息的总数量失败总数量。代码如下//Consumer.java executorService.scheduleAtFixedRate(new TimerTask() { Override public void run() { snapshotList.addLast(statsBenchmarkConsumer.createSnapshot()); if (snapshotList.size() 10) { snapshotList.removeFirst(); } } }, 1000, 1000, TimeUnit.MILLISECONDS);上面记录了性能快照后Consumer 会每隔 10s 进行性能参数计算和打印。代码如下//Consumer.java executorService.scheduleAtFixedRate(new TimerTask() { private void printStats() { if (snapshotList.size() 10) { Long[] begin snapshotList.getFirst(); Long[] end snapshotList.getLast(); final long consumeTps (long) (((end[1] - begin[1]) / (double) (end[0] - begin[0])) * 1000L); final double averageB2CRT (end[2] - begin[2]) / (double) (end[1] - begin[1]); final double averageS2CRT (end[3] - begin[3]) / (double) (end[1] - begin[1]); final long failCount end[4] - begin[4]; final long b2cMax statsBenchmarkConsumer.getBorn2ConsumerMaxRT().get(); final long s2cMax statsBenchmarkConsumer.getStore2ConsumerMaxRT().get(); statsBenchmarkConsumer.getBorn2ConsumerMaxRT().set(0); statsBenchmarkConsumer.getStore2ConsumerMaxRT().set(0); System.out.printf(Current Time: %s TPS: %d FAIL: %d AVG(B2C) RT(ms): %7.3f AVG(S2C) RT(ms): %7.3f MAX(B2C) RT(ms): %d MAX(S2C) RT(ms): %d%n, System.currentTimeMillis(), consumeTps, failCount, averageB2CRT, averageS2CRT, b2cMax, s2cMax ); } }通过性能参数的日志输出可以很方便的对 RocketMQ 的消费者进行监控。2.8.3 清除过期消息消费者会定期检查本地拉取的消息列表如果列表中的消息已经过期默认 15 分钟过期可以配置则把过期消息再次发送给 Broker然后从本地消息列表删除。代码如下//ConsumeMessageConcurrentlyService.java this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() { Override public void run() { try { cleanExpireMsg(); } catch (Throwable e) { log.error(scheduleAtFixedRate cleanExpireMsg exception, e); } } }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);2.8.4 清除过期消息消费者会每隔 30s 向 NameServer 拉取 MessageQueue 信息然后跟本地保存的进行比较如果不一致则更新本地缓存信息。代码如下//DefaultLitePullConsumerImpl.java scheduledExecutorService.scheduleAtFixedRate( new Runnable() { Override public void run() { try { fetchTopicMessageQueuesAndCompare(); } catch (Exception e) { log.error(ScheduledTask fetchMessageQueuesAndCompare exception, e); } } }, 1000 * 10, this.getDefaultLitePullConsumer().getTopicMetadataCheckIntervalMillis(), TimeUnit.MILLISECONDS);3 Broker3.1 状态采样Broker 端会对状态进行采用比如一个 Topic、MessageQueue、Group 总共发送了多少条消息Topic 总共发送的消息大小。Broker 会对这些状态按照秒、分钟、小时为单位进行采样并且定时打印这里一共有 6 个定时任务。比如下面是按照秒进行采样的定时任务//StatsItemSet.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { samplingInSeconds(); } catch (Throwable ignored) { } } }, 0, 10, TimeUnit.SECONDS);3.2 记录消息延时Broker 读取消息时会记录消息从保存磁盘到被读取的时间差并定时打印。定时任务代码如下//MomentStatsItemSet.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { printAtMinutes(); } catch (Throwable ignored) { } } }, Math.abs(UtilAll.computeNextMinutesTimeMillis() - System.currentTimeMillis()), 1000 * 60 * 5, TimeUnit.MILLISECONDS);3.3 持久化数据Broker 会定时持久化消费偏移量、Topic 配置、定阅组配置等默认 10s 一次可以配置。代码如下//ScheduleMessageService.java this.deliverExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { if (started.get()) { ScheduleMessageService.this.persist(); } } catch (Throwable e) { log.error(scheduleAtFixedRate flush exception, e); } } }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);3.4 失效过期请求Broker 会定时扫描缓存在本地的请求如果请求开始时间加超时时间再加 1s小于当前时间则这个请求过期。通过定时任务3s 一次让过期请求失效并且触发回调函数。//NettyRemotingServer.java this.timer.scheduleAtFixedRate(new TimerTask() { Override public void run() { try { NettyRemotingServer.this.scanResponseTable(); } catch (Throwable e) { log.error(scanResponseTable exception, e); } } }, 1000 * 3, 1000);3.5 过滤服务消费者可能会向 Broker 注册 filterClass 用来过滤消息。Broker 收到消费者注册的 filterClass 后会用定时任务来创建 FilterServer。代码如下//FilterServerManager.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { FilterServerManager.this.createFilterServer(); } catch (Exception e) { log.error(, e); } } }, 1000 * 5, 1000 * 30, TimeUnit.MILLISECONDS);这样消费者拉取消息时首先从 FilterServer 拉取消息FilterServer 从 Broker 拉取消息后进行过滤只把消费者感兴趣的消息返回给消费者。一个 Broker 可以有多个 FilterServer。如下图3.6 记录消息总量Broker 每天会记录前一天收发消息的总数量定时任务如下period 是 1 天//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.getBrokerStats().record(); } catch (Throwable e) { log.error(schedule record error., e); } } }, initialDelay, period, TimeUnit.MILLISECONDS);3.7 持久化 OffsetBroker 默认每隔 5s可以配置 会持久化一次消息的 Offset代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.consumerOffsetManager.persist(); } catch (Throwable e) { log.error(schedule persist consumerOffset error., e); } } }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);3.8 持久化过滤参数上面提到过消费者可能会向 Broker 注册 filterClassBroker 解析消费者注册的 filterClass 后会把解析后的 FilterData 持久化到文件代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.consumerFilterManager.persist(); } catch (Throwable e) { log.error(schedule persist consumer filter error., e); } } }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);3.9 Broker 自我保护当消费者读取消息缓慢时Broker 为了保护自己会把这个消费者设置为不允许读取的状态这样这个消费组就不能再拉取消息了代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.protectBroker(); } catch (Throwable e) { log.error(protectBroker error., e); } } }, 3, 3, TimeUnit.MINUTES);3.10 Broker 打印水位Broker 会每隔 1s 打印一次水位包括发送消息的延迟、接收消息的延迟、事务消息的延迟、查询消息的延迟代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.printWaterMark(); } catch (Throwable e) { log.error(printWaterMark error., e); } } }, 10, 1, TimeUnit.SECONDS);3.11 Broker 打印 Offset 差Broker 会定时打印最新的消息 Offset 和已经分发给 MessageQueue 和 Index 索引的 Offset 差距代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { log.info(dispatch behind commit log {} bytes, BrokerController.this.getMessageStore().dispatchBehindBytes()); } catch (Throwable e) { log.error(schedule dispatchBehindBytes error., e); } } }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);3.12 获取 NameServer 地址Broker 会定期获取 NameServer 的地址并更新本地缓存代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.brokerOuterAPI.fetchNameServerAddr(); } catch (Throwable e) { log.error(ScheduledTask fetchNameServerAddr exception, e); } } }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);3.13 打印主从偏移量差距Broker 会定时打印 master 节点和 slave 节点消息 Offset 的差距代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.printMasterAndSlaveDiff(); } catch (Throwable e) { log.error(schedule printMasterAndSlaveDiff error., e); } } }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);3.14 向 NameServer 注册Broker 会定时向默认 30s可配置最高不超过 60s所有 NameServer 发送注册消息代码如下//BrokerController.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister()); } catch (Throwable e) { log.error(registerBrokerAll Exception, e); } } }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);3.15 同步 SlaveBroker 的 Master 节点会每间隔 10s 向 Slave 节点同步数据包括 Topic 配置、消费偏移量、延迟偏移量、消费组配置代码如下//BrokerController.java slaveSyncFuture this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { BrokerController.this.slaveSynchronize.syncAll(); } catch (Throwable e) { log.error(ScheduledTask SlaveSynchronize syncAll error., e); } } }, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS);3.16 删除过期文件Broker 会周期性默认 10s可以配置地执行删除任务删除过期的 CommitLog 文件和 ConsumeQueue 文件代码如下//DefaultMessageStore.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { DefaultMessageStore.this.cleanFilesPeriodically(); } }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);3.17 文件大小检查Broker 会每隔 10 分钟检查 CommitLog 文件和 ConsumeQueue 文件用当前文件的最小起始 Offset 减去上一个文件最小起始 Offset如果不等于一个文件的大小就说明文件存在问题。代码如下//DefaultMessageStore.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { DefaultMessageStore.this.checkSelf(); } }, 1, 10, TimeUnit.MINUTES);3.18 保存堆栈映射Broker 会每隔 1s 记录所有存活线程的堆栈映射信息前提是 debugLockEnable 开关配置是打开的。代码如下//DefaultMessageStore.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { if (DefaultMessageStore.this.getMessageStoreConfig().isDebugLockEnable()) { try { if (DefaultMessageStore.this.commitLog.getBeginTimeInLock() ! 0) { long lockTime System.currentTimeMillis() - DefaultMessageStore.this.commitLog.getBeginTimeInLock(); if (lockTime 1000 lockTime 10000000) { String stack UtilAll.jstack(); final String fileName System.getProperty(user.home) File.separator debug/lock/stack- DefaultMessageStore.this.commitLog.getBeginTimeInLock() - lockTime; MixAll.string2FileNotSafe(stack, fileName); } } } catch (Exception e) { } } } }, 1, 1, TimeUnit.SECONDS);3.19 检查物理磁盘Broker 会每隔 10s 检查保存 CommitLog 的磁盘空间是否达到阈值如果达到会打印 error 级别的日志。代码如下//DefaultMessageStore.java this.diskCheckScheduledExecutorService.scheduleAtFixedRate(new Runnable() { public void run() { DefaultMessageStore.this.cleanCommitLogService.isSpaceFull(); } }, 1000L, 10000L, TimeUnit.MILLISECONDS);3.20 持久化延时消息偏移量RocketMQ 的延时消费分为 18 个级别定义如下//ScheduleMessageService.java private String messageDelayLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h;RocketMQ 会为每个延时级别定义要给 ConsumeQueue每隔 ConsumeQueue 都会有一个 Offset通过 offsetTable(ConcurrentMap) 来记录不同延时级别对应的 Offset。RocketMQ 会周期性地默认 10s可以配置把 offsetTable 中保存的 Offset 持久化到文件。代码如下//DefaultMessageStore.java this.deliverExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { if (started.get()) { ScheduleMessageService.this.persist(); } } catch (Throwable e) { log.error(scheduleAtFixedRate flush exception, e); } } }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);3.21 关闭异常连接Broker 会定时扫描所有的长连接主要包括生产者、消费者和 FilterServer如果连接不活跃则关闭该连接并从本地连接列表中移除。代码如下//ClientHousekeepingService.java this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { Override public void run() { try { ClientHousekeepingService.this.scanExceptionChannel(); } catch (Throwable e) { log.error(Error occurred when scan not active client channels., e); } } }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);3.22 清理过期消息如果 Broker 配置了允许快速失败brokerFastFailureEnable则会每隔 10ms 定时清理过期请求包括要发送的消息、接收的消息、心跳消息、要结束的事务消息。代码如下scheduledExecutorService.scheduleAtFixedRate( new Runnable() { Override public void run() { try { fetchTopicMessageQueuesAndCompare(); } catch (Exception e) { log.error(ScheduledTask fetchMessageQueuesAndCompare exception, e); } } }, 1000 * 10, this.getDefaultLitePullConsumer().getTopicMetadataCheckIntervalMillis(), TimeUnit.MILLISECONDS);注意清理消息前会判断是否系统繁忙如果系统繁忙会给发送队列中的消息直接返回系统繁忙暂时不做过期消息清理。4 NameServer4.1 检查过期 Broker在 3.14 节中讲过Broker 会跟 NameServer 建立长连接定时向 NameServer 发送注册消息。NameServer 会在本地维护一个 Broker 列表定时任务会轮询本地保存的 Broker 列表检查注册消息是否过期超过 120s如果注册消息过期则关闭长连接从本地缓存删除这个 Broker。代码如下//NamesrvController.java this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker, 5, 10, TimeUnit.SECONDS);4.2 打印配置NameServer 启动时会加载 KV 格式的配置文件到 configTable 这个变量NameServer 客户端也可以发送一个 KV 配置请求给 NameServerNameServer 收到请求后也会保存到 configTable。NameServer 会定时打印 configTable 中的配置代码如下//NamesrvController.java this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.kvConfigManager::printAllPeriodically, 1, 10, TimeUnit.MINUTES);5 总结RocketMQ 的定时任务很多这些定时任务的加入让 RocketMQ 的设计更加完备包括业务处理、监控日志、心跳、清理任务、关闭连接、持久化数据等。通过对定时任务的理解能够更深入地理解 RocketMQ 的设计理念。