文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Go+Redis实现延迟队列实操

2024-04-02 19:55

关注

前言

延迟队列是一种非常使用的数据结构,我们经常有需要延迟推送处理消息的场景,比如延迟60秒发送短信,延迟30分钟关闭订单,消息消费失败延迟重试等等。

一般我们实现延迟消息都需要依赖底层的有序结构,比如堆,而Redis刚好提供了zset这种数据类型,它的底层实现是哈希表+跳表,也是一种有序的结构,所以这篇文章主要是使用Go+Redis来实现延迟队列。

当然Redis本身并不支持延迟队列,所以我们只是实现一个比较简单的延迟队列,而且Redis不太适合大量消息堆积,所以只适合比较简单的场景,如果需要更加强大稳定的消息队列,可以使用RocketMQ等自带延迟消息的消息队列。

我们这里先定一下我们要实现的几个目标:

然后我们定义一个简单的接口:

简单的实现

定义消息

这个消息参考了Kafka的消息结构:

// 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函数,那么多个线程会拉取相同的可执行的消息,造成消息重复的被消费。

多消费者实现

定义消息

// 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就比较复杂,因为我们必须记录哪个分区已经被消费者消费了。如果真的需要更加复杂的场景还是建议使用RocketMQKafka等消息队列进行实现。

总结

到此这篇关于Go+Redis实现延迟队列实操的文章就介绍到这了,更多相关Go 延迟队列内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯