消息队列(MQ)消息延迟及过滤设计方案

如题所述

第1个回答  2022-07-02

消息队列接收到请求后,会将消息顺序写入Physic log文件,对于延迟消息,将消息按照将要投递的时间,以小时为单位异步分割存储,每个小时的消息顺序写入延迟物理文件delay log,并把索引存储在delay index文件,索引记录消息在delay log中的{offset,size,投递时间戳}元信息,基于有限内存以及延迟消息分发特性,我们仅将最近两个小时的delay log文件序列采用mmap内存映射机制进行读写,延迟2个小时以上的消息直接写入磁盘文件。

但消息分发时,每次需要将一个小时的的索引文件,全部加载到内存,由于每个小时的消息索引是顺序写入delay index的,而消息分发投递时间又是随机的,写入顺序与消息投递顺序并不一致,所以,索引加载到内存后,需要按照消息具体投递的秒级时间戳进行排序,再根据排序后的索引读取delay log中的消息进行分发投递。

这种存储方案有以下问题: 1.一次需要加载整个小时的消息索引到内存,若并发比较高,内存压力比较大。 2.按照消息投递秒级时间戳进行排序后,实时到来的新的消息,需要实时插入排序,性能较低,延迟大。

为了解决上述问题,我们将delay index中索引元信息{offset, size, 投递时间戳}改为{offset, size, localIndex, globalIndex, preGlobalIndex },其中:

通过globalIndex可以直接定位到delay index中的索引单元,从而确定delay log中的一条消息,而preGlobalIndex又可以定位到同一秒内的上一条消息,因此只要落地存储每个小时,每秒最后一条消息的索引ID,即可逆序查出每秒所有消息。一个小时内只有3600秒,只需要将3600个16字节的索引ID加载到内存,即可实现每秒消息的实时加载。

为了降低消息分发延迟,可将最近10s的消息索引提前预加载到内存,对于实时接收到的消息,根据时间戳匹配到对应的秒,更新这一秒最新一条消息的索引globalIndex与逆向索引preGlobalIndex,不需要做排序,消息插入与读取的复杂度都为O(1)。

采用的由数组加链表实现的多级时间轮机制,分别是秒级和小时级,小时级时间轮前移一个槽,对应秒级时间轮旋转一圈,秒级时间轮上一共3600个槽,每个槽的时间跨度最大为1s,时间轮每秒前移一个槽。小时级实践论每个槽时间跨度是1小时,每小时移动一个槽,将后面两个小时的delay log开启内存映射,同时清除两个小时之前delay log文件内存映射。

当我们只有一个2小时5分钟的消息发送时,秒级时间轮需要推动2圈后即小时级时间轮移动2个槽,剩5分钟的延迟,再降级到秒级时间轮。这叫造成了时间轮的空转。

一般会把每个使用到的槽都会放到DelayQueue中,然后根据DelayQueue来 协助时间轮的推进 ,防止空推进的情况。例如,当有延迟500s的任务时,除了挂载到时间轮外,我们还会把其放到DelayQueue中,这样DelayQueue的头结点为延迟500s,如果期间没有小于500s的延迟任务再加进来时,我们只需要等待500s,时间轮推进一次即可。如果有小于500s的定时任务新加进来,我们只需要唤醒DelayQueue,重新计算等待时间即可。

即当有定时任务新增时,如果对应槽为新槽(即新增任务为该槽的第一个任务),在DelayQueue中增加延迟任务,并判断是否为头结点,是的话唤醒DelayQueue重新计算等待时间。

当master发生漂移或者网络异常时,时间轮分发控制需要从原master节点切换到新的master节点。为了保证分发状态的连续性与一致性,master节点定时每隔50ms分别将两个时间轮上分发的tick信息同步到其它slave节点。通过tick可确定具体分发到第几秒,但不能确定分发到这一秒的第几条消息,为此二级时间轮增加同步了一个参数localIndex,记录当前秒分发到第几条消息,并且每个节点都会定时将分发状态持久化。

每当master发生切换时,原master节点切换为slave,会立即停止当前时间轮的分发任务,并清空分发状态;而新的master节点根据当前已同步过来的分发状态初始化两级时间轮,但master切换会有一定的延迟周期或者极端情况下不同节点间时钟存在偏差,新的master初始完时间轮的tick后,该tick对应的秒级时间戳有可能与节点实际时间不一致,启动分发任务前需要做特殊调整,若tick时间戳小于当前时间,则分发任务sleep等待直至时间对齐,若tick时间戳大于当前时间,说明存在已到期的消息未分发,此时连续推进tick迁移,并对到期消息直接异步投递,直到tick对应时间戳小于当前时间。

正常master切换分为两种情况,一种为主动释放master,如节点重启与master负载均衡过程,这种情况节点在drop master之前,会首先同步时间轮分发状态到其它slave节点,此时master切换时间轮分发时完全连续一致的;另一种是一些异常情况下master被动漂移,此时新的master节点上时间轮分发状态可能存在最大50ms的延迟,会出现部分消息重复分发现象。把时间轮分发状态信息封装到到期投递消息协议扩展字段中,paxos请求同步消息时携带时间轮状态,即可做到实时同步。

众所周知,RocketMQ是支持消息过滤的,即发送消息时,可以给消息设置一个TAG。订阅主题的时候,可以设置只消费携带某些TAG的消息,起到消息过滤的作用。

客户端拉取消息时,在服务端得到tag的hash集合codeSet,然后从ConsumerQueue获取一条记录,判断记录的hashCode是否在codeSet中,以达到消息过滤的目的,决定是否将该消息发送给consumer。

因为Hash存在冲突,过滤不完全准确,所以,客户端收到消息后,会进行再次精准过滤。

还有一种过滤方式,把TAG通过哈希转换为long,索引中保存所有TAG的哈希值按位或的结果。当拉取消息时,通过订阅设置的TAG哈希值与索引中的哈希值进行按位与操作,如果结果等于订阅设置的TAG哈希值,说明该索引对应的消息可能符合条件,二次精准过滤依旧在客户端来做;否则,一定不符合条件,直接过滤掉。

相似回答
大家正在搜