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

阳光穿透心脏的1/2处 2023-09-28 18:47 103阅读 0赞

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

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

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

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

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

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

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

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

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

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

本文的完整代码：

> https://github.com/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.SendDela