延迟队列:一种带有 延迟功能 的消息队列
这样解释,整个设计就清楚了。你的目的是 延时,承载容器是 mq。
列举一下我日常业务中可能存在的场景:
为了解决以上问题,最简单直接的办法就是定时去扫表:
服务启动时,开启一个异步协程 → 定时扫描 msg table,到了事件触发事件,调用对应的 handler
几个缺点:
最大问题其实是什么?
调度模型基本统一,不要做重复的业务逻辑
我们可以考虑将逻辑从具体的业务逻辑里面抽出来,变成一个公共的部分。
而这个调度模型,就是 延时队列 。
其实说白了:
延时队列模型,就是将未来执行的事件提前存储好,然后不断扫描这个存储,触发执行时间则执行对应的任务逻辑。
那么开源界是否已有现成的方案呢?答案是肯定的。Beanstalk (https://github.com/beanstalkd/beanstalkd) 它基本上已经满足以上需求
一次说说上述这些目的的设计方向:
这个概念取自 mq 。mq 中提供了消费投递的几个方向:
at most once
→ 至多一次,消息可能会丢,但不会重复at least once
→ 至少一次,消息肯定不会丢失,但可能重复exactly once
→ 有且只有一次,消息不丢失不重复,且只消费一次。exactly once
尽可能是 producer + consumer 两端都保证。当 producer 没办法保证是,那 consumer 需要在消费前做一个去重,达到消费过一次不会重复消费,这个在延迟队列内部直接保证。
最简单:使用 redis 的 setNX 达到 job id 的唯一消费
支持多实例部署。挂掉一个实例后,还有后备实例继续提供服务。
这个对外提供的 API 使用 cluster 模型,内部将多个 node 封装起来,多个 node 之间冗余存储。
考虑过类似基于 kafka/rocketmq 等消息队列作为存储的方案,最后从存储设计模型放弃了这类选择。
举个例子,假设以 Kafka 这种消息队列存储来实现延时功能,每个队列的时间都需要创建一个单独的 topic(如: Q1-1s, Q1-2s..)。这种设计在延时时间比较固定的场景下问题不太大,但如果是延时时间变化比较大会导致 topic 数目过多,会把磁盘从顺序读写会变成随机读写从导致性能衰减,同时也会带来其他类似重启或者恢复时间过长的问题。
producer
producer.At(msg []byte, at time.Time)
producer.Delay(body []byte, delay time.Duration)
producer.Revoke(ids string)
consumer
consumer.Consume(consume handler)
使用延时队列后,服务整体结构如下,以及队列中 job 的状态变迁:
producer.At(msg []byte, at time.Time)
→ 插入延时job到 tube 中主要介绍一下在日常开发,我们使用到延时队列的哪些具体功能。
producer.At(msg []byte, at time.Time)
producer.Revoke(ids string)
,对其删除,然后重新插入首先,框架层面保证了消费行为的 exactly once
,但是上层业务逻辑消费失败或者是出现网络问题,亦或者是各种各样的问题,导致消费失败,兜底交给业务开发做。这样做的原因:
这里描述一下框架消费端是怎么保证消费行为的统一:
分为 cluster 和 node。cluster:
https://github.com/tal-tech/go-queue/blob/master/dq/consumer.go#L45
node:
https://github.com/tal-tech/go-queue/blob/master/dq/consumernode.go#L36
所以对于消费端,开发者需要自己实现消费的幂等性。
go-queue
是基于 go-zero
实现的,go-zero
在 github 上 Used by
有300+,开源一年获得11k+ stars.
go-zero: https://github.com/zeromicro/go-zero
go-queue: https://github.com/tal-tech/go-queue
欢迎使用并 star 支持我们!
关注『微服务实践』公众号并点击 交流群 获取社区群二维码。