延时消息常见实现方案
共 4704字,需浏览 10分钟
·
2022-01-22 16:08
延时消息(定时消息)指的在分布式异步消息场景下,生产端发送一条消息,希望在指定延时或者指定时间点被消费端消费到,而不是立刻被消费。
延时消息适用的业务场景非常的广泛,在分布式系统环境下,延时消息的功能一般会在下沉到中间件层,通常是 MQ 中内置这个功能或者内聚成一个公共基础服务。
本文旨在探讨常见延时消息的实现方案以及方案设计的优缺点。
1. 基于外部存储实现的方案
这里讨论的外部存储指的是在 MQ 本身自带的存储以外又引入的其他的存储系统。
基于外部存储的方案本质上都是一个套路,将 MQ 和 延时模块 区分开来,延时消息模块是一个独立的服务/进程。延时消息先保留到其他存储介质中,然后在消息到期时再投递到 MQ。当然还有一些细节性的设计,比如消息进入的延时消息模块时已经到期则直接投递这类的逻辑,这里不展开讨论。
下述方案不同的是,采用了不同的存储系统。
基于 数据库(如MySQL)
基于关系型数据库(如MySQL)延时消息表的方式来实现。
CREATE TABLE `delay_msg` (
`id` bigint unsigned NOT NULL AUTO_INCREMENT,
`delivery_time` DATETIME NOT NULL COMMENT '投递时间',
`payloads` blob COMMENT '消息内容',
PRIMARY KEY (`id`),
KEY `time_index` (`delivery_time`)
)
通过定时线程定时扫描到期的消息,然后进行投递。定时线程的扫描间隔理论上就是你延时消息的最小时间精度。
优点:
实现简单;
缺点:
B+Tree索引不适合消息场景的大量写入;
RocksDB 的方案其实就是在上述方案上选择了比较合适的存储介质。
RocksDB 在笔者之前的文章中有聊过,LSM 树更适合大量写入的场景。滴滴开源的DDMQ中的延时消息模块 Chronos 就是采用了这个方案。
DDMQ 这个项目简单来说就是在 RocketMQ 外面加了一层统一的代理层,在这个代理层就可以做一些功能维度的扩展。延时消息的逻辑就是代理层实现了对延时消息的转发,如果是延时消息,会先投递到 RocketMQ 中 Chronos 专用的 topic 中。延时消息模块 Chronos 消费得到延时消息转储到 RocksDB,后面就是类似的逻辑了,定时扫描到期的消息,然后往 RocketMQ 中投递。
这个方案老实说是一个比较重的方案。因为基于 RocksDB 来实现的话,从数据可用性的角度考虑,你还需要自己去处理多副本的数据同步等逻辑。
优点:
RocksDB LSM 树很适合消息场景的大量写入;
缺点:
实现方案较重,如果你采用这个方案,需要自己实现 RocksDB 的数据容灾逻辑;
再来聊聊 Redis 的方案。下面放一个比较完善的方案。
本方案来源于:基于Redis实现延时队列服务
Messages Pool 所有的延时消息存放,结构为KV结构,key为消息ID,value为一个具体的message(这里选择Redis Hash结构主要是因为hash结构能存储较大的数据量,数据较多时候会进行渐进式rehash扩容,并且对于HSET和HGET命令来说时间复杂度都是O(1))
Delayed Queue是16个有序队列(队列支持水平扩展),结构为ZSET,value 为 messages pool中消息ID,score为过期时间**(分为多个队列是为了提高扫描的速度)**
Worker 代表处理线程,通过定时任务扫描 Delayed Queue 中到期的消息
这个方案选用 Redis 存储在我看来有几点考虑。
Redis ZSET 很适合实现延时队列
性能问题,虽然 ZSET 插入是一个 O(logn) 的操作,但是Redis 基于内存操作,并且内部做了很多性能方面的优化。
但是这个方案其实也有需要斟酌的地方,上述方案通过创建多个 Delayed Queue 来满足对于并发性能的要求,但这也带来了多个 Delayed Queue 如何在多个节点情况下均匀分配,并且很可能出现到期消息并发重复处理的情况,是否要引入分布式锁之类的并发控制设计?
在量不大的场景下,上述方案的架构其实可以蜕化成主从架构,只允许主节点来处理任务,从节点只做容灾备份。实现难度更低更可控。
上述几个方案中,都通过线程定时扫描的方案来获取到期的消息。
定时线程的方案在消息量较少的时候,会浪费资源,在消息量非常多的时候,又会出现因为扫描间隔设置不合理导致延时时间不准确的问题。可以借助 JDK Timer 类中的思想,通过 wait-notify 来节省 CPU 资源。
获取中最近的延时消息,然后wait(执行时间-当前时间),这样就不需要浪费资源到达时间时会自动响应,如果有新的消息进入,并且比我们等待的消息还要小,那么直接notify唤醒,重新获取这个更小的消息,然后又wait,如此循环。
2. 开源 MQ 中的实现方案
再来讲讲目前自带延时消息功能的开源MQ,它们是如何实现的
Level 数固定,每个 Level 有自己的定时器,开销不大 将 Level 相同的消息放入到同一个 Queue 中,保证了同一 Level 消息的顺序性;不同 Level 放到不同的 Queue 中,保证了投递的时间准确性; 通过只支持固定的Level,将不同延时消息的排序变成了固定Level Topic 的追加写操作
Level 配置的修改代价太大,固定 Level 不灵活 CommitLog 会因为延时消息的存在变得很大
内存开销: 维护延时消息索引的队列是放在堆外内存中的,并且这个队列是以订阅组(Kafka中的消费组)为维度的,比如你这个 Topic 有 N 个订阅组,那么如果你这个 Topic 使用了延时消息,就会创建 N 个 队列;并且随着延时消息的增多,时间跨度的增加,每个队列的内存占用也会上升。(是的,在这个方案下,支持任意的延时消息反而有可能让这个缺陷更严重) 故障转移之后延时消息索引队列的重建时间开销: 对于跨度时间长的大规模延时消息,重建时间可能会到小时级别。(摘自 Pulsar 官方公众号文章) 存储开销:延时消息的时间跨度会影响到 Pulsar 中已经消费的消息数据的空间回收。打个比方,你的 Topic 如果业务上要求支持一个月跨度的延时消息,然后你发了一个延时一个月的消息,那么你这个 Topic 中底层的存储就会保留整整一个月的消息数据,即使这一个月中99%的正常消息都已经消费了。
如果对时间轮不熟悉的可以阅读笔者的这篇文章 从 Kafka 看时间轮算法设计
时间轮算法适合延时/定时消息的场景,省去延时消息的排序,插入删除操作都是 O(1) 的时间复杂度; 通过多级时间轮设计,支持了超大时间跨度的延时消息; 通过延时加载,内存中只会有最近要消费的消息,更久的延时消息会被存储在磁盘中,对内存友好; 延时消息单独存储(schedule log),不会影响到正常消息的空间回收;
作者:Richard_Yi
来源:juejin.cn/post/7052894117105238053