用 Redis 做一个可靠的延迟队列

快来打我* 2023-09-28 16:09 9阅读 0赞

我们先看看以下业务场景：

1.  当订单一直处于未支付状态时，如何及时的关闭订单，并退还库存？
2.  新创建店铺，N天内没有上传商品，系统如何知道该信息，并发送激活短信？

上述场景最简单直接的解决方案是定时扫表。我们假设 10 分钟未支付则关闭订单、定时任务设置为 5 分钟一次，那么一个订单最晚会在 15 分钟关闭。高达 5 分钟的误差是业务难以接受的。另一方面频繁的扫表可能消耗过多数据库资源，影响线上交易吞吐量。

此外还有朋友使用 Redis 的过期通知、时间轮、Java 的 DelayQueue 等方式实现延时任务。我们在之前的文章中讨论过他们的缺陷：比如使用 Redis 过期通知不保证准时、发送即忘不保证送达，时间轮缺乏持久化机制容易丢失等。

总结一下，我们对于延时队列的要求有下列几条（从重要到不重要排列）：

1.  持久化: 服务重启或崩溃不能丢失任务
2.  确认重试机制: 任务处理失败或超时应该有重试
3.  定时尽量精确

最合适的解决方案是使用 Pulsa、RocketMQ 等专业消息队列的延时投递功能。不过引入新的中间件通常存在各种非技术方面的麻烦。Redis 作为广泛使用的中间件，何不用 Redis 来制作延时队列呢？

使用有序集合结构实现延时队列的方法已经广为人知，无非是将消息作为有序集合的 member 投递时间戳作为 score，使用 zrangebyscore 命令搜索已到投递时间的消息然后将其发给消费者。

除了基本的延时投递之外我们的消息队列具有下列优势：

1.  提供 ACK 和重试机制
2.  只需要 Redis 和消费者即可运行，无需其它组件
3.  提供 At-Least-One 投递语义、并保证消息不会并发消费

本文的完整代码实现在hdt3213/delayqueue，可以直接 `go get github.com/hdt3213/delayqueue` 完成安装。

具体使用也非常简单，只需要注册处理消息的回调函数并调用 start() 即可：

package main
    
    import (
      "github.com/go-redis/redis/v8"
      "github.com/hdt3213/delayqueue"
      "strconv"
      "time"
    )
    
    func main() {
      redisCli := redis.NewClient(&redis.Options{
        Addr: "127.0.0.1:6379",
      })
      queue := delayqueue.NewQueue("example-queue", redisCli, func(payload string) bool {
        // 注册处理消息的回调函数
            // 返回 true 表示已成功消费，返回 false 消息队列会重新投递次消息
        return true
      })
      // 发送延时消息
      for i := 0; i < 10; i++ {
        err := queue.SendDelayMsg(strconv.Itoa(i), time.Hour, delayqueue.With