Go+Redis实现延迟队列实操_Golang

前言

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

一般我们实现延迟消息都需要依赖底层的有序结构，比如堆，而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等消息队列进行实现。