Go+Redis实现延迟队列实操
来源:脚本之家
时间:2023-02-23 12:04:22 426浏览 收藏
IT行业相对于一般传统行业,发展更新速度更快,一旦停止了学习,很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习,精进自己的技术,尤其是初学者。今天golang学习网给大家整理了《Go+Redis实现延迟队列实操》,聊聊Redis、延迟、队列,我们一起来看看吧!
一般我们实现延迟消息都需要依赖底层的有序结构,比如堆,而Redis刚好提供了zset
这种数据类型,它的底层实现是哈希表+跳表
,也是一种有序的结构,所以这篇文章主要是使用Go+Redis来实现延迟队列。
当然Redis本身并不支持延迟队列,所以我们只是实现一个比较简单的延迟队列,而且Redis不太适合大量消息堆积,所以只适合比较简单的场景,如果需要更加强大稳定的消息队列,可以使用RocketMQ
等自带延迟消息的消息队列。
我们这里先定一下我们要实现的几个目标:
- 消息必须至少被消费一次
- 多个生产者
- 多个消费者
然后我们定义一个简单的接口:
- Push(msg) error:添加消息到队列
- Consume(topic, batchSize, func(msg) error):消费消息
简单的实现
- 每个主题最多可以被一个消费者消费,因为不会对主题进行分区
- 但是可以多个生产者同时进行生产,因为Push操作是原子的
- 同时需要消费操作返回值error为nil才删除消息,保证消息至少被消费一次
定义消息
这个消息参考了Kafka的消息结构:
- Topic可以是某个队列的名字
- Key是消息的唯一标识,在一个队列里面不可以重复
- Body是消息的内容
- Delay是消息的延迟时间
- ReadyTime是消息准备好执行的时间
// Msg 消息 type Msg struct { Topic string // 消息的主题 Key string // 消息的Key Body []byte // 消息的Body Delay time.Duration // 延迟时间(秒) ReadyTime time.Time // 消息准备好执行的时间(now + delay) }
Push
由于我们需要把消息的Body存储到Hash
,把消息的ReadyTime存储到ZSet
,所以我们需要一个简单的Lua脚本来保证这两个操作是原子的
。
同时我们不会覆盖已经存在的相同Key的消息。
const delayQueuePushRedisScript = ` -- KEYS[1]: topicZSet -- KEYS[2]: topicHash -- ARGV[1]: 消息的Key -- ARGV[2]: 消息的Body -- ARGV[3]: 消息准备好执行的时间 local topicZSet = KEYS[1] local topicHash = KEYS[2] local key = ARGV[1] local body = ARGV[2] local readyTime = tonumber(ARGV[3]) -- 添加readyTime到zset local count = redis.call("zadd", topicZSet, readyTime, key) -- 消息已经存在 if count == 0 then return 0 end -- 添加body到hash redis.call("hsetnx", topicHash, key, body) return 1 `
func (q *SimpleRedisDelayQueue) Push(ctx context.Context, msg *Msg) error { // 如果设置了ReadyTime,就使用RedisTime var readyTime int64 if !msg.ReadyTime.IsZero() { readyTime = msg.ReadyTime.Unix() } else { // 否则使用Delay readyTime = time.Now().Add(msg.Delay).Unix() } success, err := q.pushScript.Run(ctx, q.client, []string{q.topicZSet(msg.Topic), q.topicHash(msg.Topic)}, msg.Key, msg.Body, readyTime).Bool() if err != nil { return err } if !success { return ErrDuplicateMessage } return nil }
Consume
其中第二个参数batchSize
表示用于批量获取已经准备好执行的消息,减少网络请求。
fn
是对消息进行处理的函数,它有一个返回值error
,如果是nil
才表示消息消费成功,然后调用删除脚本把成功消费的消息给删除(需要原子的删除ZSet和Hash里面的内容)。
const delayQueueDelRedisScript = ` -- KEYS[1]: topicZSet -- KEYS[2]: topicHash -- ARGV[1]: 消息的Key local topicZSet = KEYS[1] local topicHash = KEYS[2] local key = ARGV[1] -- 删除zset和hash关于这条消息的内容 redis.call("zrem", topicZSet, key) redis.call("hdel", topicHash, key) return 1 `
func (q *SimpleRedisDelayQueue) Consume(topic string, batchSize int, fn func(msg *Msg) error) { for { // 批量获取已经准备好执行的消息 now := time.Now().Unix() zs, err := q.client.ZRangeByScoreWithScores(context.Background(), q.topicZSet(topic), &redis.ZRangeBy{ Min: "-inf", Max: strconv.Itoa(int(now)), Count: int64(batchSize), }).Result() // 如果获取出错或者获取不到消息,则休眠一秒 if err != nil || len(zs) == 0 { time.Sleep(time.Second) continue } // 遍历每个消息 for _, z := range zs { key := z.Member.(string) // 获取消息的body body, err := q.client.HGet(context.Background(), q.topicHash(topic), key).Bytes() if err != nil { continue } // 处理消息 err = fn(&Msg{ Topic: topic, Key: key, Body: body, ReadyTime: time.Unix(int64(z.Score), 0), }) if err != nil { continue } // 如果消息处理成功,删除消息 q.delScript.Run(context.Background(), q.client, []string{q.topicZSet(topic), q.topicHash(topic)}, key) } } }
存在的问题
如果多个线程同时调用Consume
函数,那么多个线程会拉取相同的可执行的消息,造成消息重复的被消费。
多消费者实现
- 每个主题最多可以被分区个数个消费者消费,会对主题进行分区
定义消息
- 我们添加了一个Partition字段表示消息的分区号
// Msg 消息 type Msg struct { Topic string // 消息的主题 Key string // 消息的Key Body []byte // 消息的Body Partition int // 分区号 Delay time.Duration // 延迟时间(秒) ReadyTime time.Time // 消息准备好执行的时间 }
Push
代码与SimpleRedisDelayQueue的Push相同,只是我们会使用Msg里面的Partition字段对主题进行分区。
func (q *PartitionRedisDelayQueue) topicZSet(topic string, partition int) string { return fmt.Sprintf("%s:%d:z", topic, partition) } func (q *PartitionRedisDelayQueue) topicHash(topic string, partition int) string { return fmt.Sprintf("%s:%d:h", topic, partition) }
Consume
代码与SimpleRedisDelayQueue的Consume相同,我们也只是对Consume多加了一个partition
参数用于指定消费的分区。
func (q *PartitionRedisDelayQueue) Consume(topic string, batchSize, partition int, fn func(msg *Msg) error) { // ... }
存在的问题
一个比较大的问题就是我们需要手动指定分区而不是自动分配分区,这个问题对于Push操作解决起来比较容易,可以通过哈希算法对Key进行哈希取模进行分区,比如murmur3。但是对于Consume就比较复杂,因为我们必须记录哪个分区已经被消费者消费了。如果真的需要更加复杂的场景还是建议使用RocketMQ
、Kafka
等消息队列进行实现。
总结
- 使用Redis的ZSet可以很容易的实现一个高性能消息队列
- 但是Redis的ZSet实现的消息队列不适合大量消息堆积的场景,同时如果需要实现自动分区消费功能会比较复杂
- 适合消息量不是很大,且不是很复杂的场景
- 如果需要大量堆积消息和稳定的多消费者功能,可以使用自带延迟消息的
RocketMQ
本篇关于《Go+Redis实现延迟队列实操》的介绍就到此结束啦,但是学无止境,想要了解学习更多关于Golang的相关知识,请关注golang学习网公众号!
-
286 收藏
-
117 收藏
-
321 收藏
-
185 收藏
-
134 收藏
-
444 收藏
-
311 收藏
-
266 收藏
-
188 收藏
-
317 收藏
-
430 收藏
-
- 前端进阶之JavaScript设计模式
- 设计模式是开发人员在软件开发过程中面临一般问题时的解决方案,代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景,打造一站式知识长龙服务,适合有JS基础的同学学习。
- 立即学习 542次学习
-
- GO语言核心编程课程
- 本课程采用真实案例,全面具体可落地,从理论到实践,一步一步将GO核心编程技术、编程思想、底层实现融会贯通,使学习者贴近时代脉搏,做IT互联网时代的弄潮儿。
- 立即学习 507次学习
-
- 简单聊聊mysql8与网络通信
- 如有问题加微信:Le-studyg;在课程中,我们将首先介绍MySQL8的新特性,包括性能优化、安全增强、新数据类型等,帮助学生快速熟悉MySQL8的最新功能。接着,我们将深入解析MySQL的网络通信机制,包括协议、连接管理、数据传输等,让
- 立即学习 497次学习
-
- JavaScript正则表达式基础与实战
- 在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
- 立即学习 487次学习
-
- 从零制作响应式网站—Grid布局
- 本系列教程将展示从零制作一个假想的网络科技公司官网,分为导航,轮播,关于我们,成功案例,服务流程,团队介绍,数据部分,公司动态,底部信息等内容区块。网站整体采用CSSGrid布局,支持响应式,有流畅过渡和展现动画。
- 立即学习 484次学习