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

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

    • 当订单一直处于未支付状态时，如何及时的关闭订单，并退还库存？

    • 新创建店铺，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() 即可：

    1. package main
    2.  
    3. import (
    4.  "github.com/go-redis/redis/v8"
    5.  "github.com/hdt3213/delayqueue"
    6.  "strconv"
    7.  "time"
    8. )
    9.  
    10. func main() {
    11.  redisCli := redis.NewClient(&redis.Options{
    12.   Addr: "127.0.0.1:6379",
    13.  })
    14.  queue := delayqueue.NewQueue("example-queue", redisCli, func(payload string) bool {
    15.   // 注册处理消息的回调函数
    16.         // 返回 true 表示已成功消费，返回 false 消息队列会重新投递次消息
    17.   return true
    18.  })
    19.  // 发送延时消息
    20.  for i := 0; i < 10; i++ {
    21.   err := queue.SendDelayMsg(strconv.Itoa(i), time.Hour, delayqueue.WithRetryCount(3))
    22.   if err != nil {
    23.    panic(err)
    24.   }
    25.  }
    26.  
    27.  // 启动消费协程
    28.  done := queue.StartConsume()
    29.  // 阻塞等待消费协程退出
    30.  <-done
    31. }

    由于数据存储在 redis 中所以我们最多能保证在 redis 无故障且消息队列相关 key 未被外部篡改的情况下不会丢失消息。

    原理详解

    消息队列涉及几个关键的 redis 数据结构：

    • msgKey: 为了避免两条内容完全相同的消息造成意外的影响，我们将每条消息放到一个字符串类型的键中，并分配一个 UUID 作为它的唯一标识。其它数据结构中只存储 UUID 而不存储完整的消息内容。每个 msg 拥有一个独立的 key 而不是将所有消息放到一个哈希表是为了利用 TTL 机制避免

    • pendingKey: 有序集合类型，member 为消息 ID, score 为投递时间的 unix 时间戳。

    • readyKey: 列表类型，需要投递的消息 ID。

    • unAckKey: 有序集合类型，member 为消息 ID, score 为重试时间的 unix 时间戳。

    • retryKey: 列表类型，已到重试时间的消息 ID

    • garbageKey: 集合类型，用于暂存已达重试上线的消息 ID

    • retryCountKey: 哈希表类型，键为消息 ID, 值为剩余的重试次数

    流程如下图所示：

    由于我们允许分布式地部署多个消费者，每个消费者都在定时执行 lua 脚本，所以多个消费者可能处于上述流程中不同状态，我们无法预知（或控制）上图中五个操作发生的先后顺序，也无法控制有多少实例正在执行同一个操作。

    因此我们需要保证上图中五个操作满足三个条件：

    • 都是原子性的

    • 不会重复处理同一条消息

    • 操作前后消息队列始终处于正确的状态

    只要满足这三个条件，我们就可以部署多个实例且不需要使用分布式锁等技术来进行状态同步。

    是不是听起来有点吓人？😂 其实简单的很，让我们一起来详细看看吧~

    pending2ReadyScript

    pending2ReadyScript 使用 zrangebyscore 扫描已到投递时间的消息ID并把它们移动到 ready 中:

    1. -- keys: pendingKey, readyKey
    2. -- argv: currentTime
    3. local msgs = redis.call('ZRangeByScore', KEYS[1], '0', ARGV[1])  -- 从 pending key 中找出已到投递时间的消息
    4. if (#msgs == 0) then return end
    5. local args2 = {'LPush', KEYS[2]} -- 将他们放入 ready key 中
    6. for _,v in ipairs(msgs) do
    7.  table.insert(args2, v) 
    8. end
    9. redis.call(unpack(args2))
    10. redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1])  -- 从 pending key 中删除已投递的消息

    ready2UnackScript

    ready2UnackScript 从 ready 或者 retry 中取出一条消息发送给消费者并放入 unack 中，类似于 RPopLPush：

    1. -- keys: readyKey/retryKey, unackKey
    2. -- argv: retryTime
    3. local msg = redis.call('RPop', KEYS[1])
    4. if (not msg) then return end
    5. redis.call('ZAdd', KEYS[2], ARGV[1], msg)
    6. return msg

    unack2RetryScript

    unack2RetryScript 从 retry 中找出所有已到重试时间的消息并把它们移动到 unack 中：

    1. -- keys: unackKey, retryCountKey, retryKey, garbageKey
    2. -- argv: currentTime
    3. local msgs = redis.call('ZRangeByScore', KEYS[1], '0', ARGV[1])  -- 找到已到重试时间的消息
    4. if (#msgs == 0) then return end
    5. local retryCounts = redis.call('HMGet', KEYS[2], unpack(msgs)) -- 查询剩余重试次数
    6. for i,v in ipairs(retryCounts) do
    7.  local k = msgs[i]
    8.  if tonumber(v) > 0 then -- 剩余次数大于 0
    9.   redis.call("HIncrBy", KEYS[2], k, -1) -- 减少剩余重试次数
    10.   redis.call("LPush", KEYS[3], k) -- 添加到 retry key 中
    11.  else -- 剩余重试次数为 0
    12.   redis.call("HDel", KEYS[2], k) -- 删除重试次数记录
    13.   redis.call("SAdd", KEYS[4], k) -- 添加到垃圾桶，等待后续删除
    14.  end
    15. end
    16. redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1])  -- 将已处理的消息从 unack key 中删除

    因为 redis 要求 lua 脚本必须在执行前在 KEYS 参数中声明自己要访问的 key, 而我们将每个 msg 有一个独立的 key，我们在执行 unack2RetryScript 之前是不知道哪些 msg key 需要被删除。所以 lua 脚本只将需要删除的消息记在 garbage key 中，脚本执行完后再通过 del 命令将他们删除:

    1. func (q *DelayQueue) garbageCollect() error {
    2.  ctx := context.Background()
    3.  msgIds, err := q.redisCli.SMembers(ctx, q.garbageKey).Result()
    4.  if err != nil {
    5.   return fmt.Errorf("smembers failed: %v", err)
    6.  }
    7.  if len(msgIds) == 0 {
    8.   return nil
    9.  }
    10.  // allow concurrent clean
    11.  msgKeys := make([]string, 0, len(msgIds))
    12.  for _, idStr := range msgIds {
    13.   msgKeys = append(msgKeys, q.genMsgKey(idStr))
    14.  }
    15.  err = q.redisCli.Del(ctx, msgKeys...).Err()
    16.  if err != nil && err != redis.Nil {
    17.   return fmt.Errorf("del msgs failed: %v", err)
    18.  }
    19.  err = q.redisCli.SRem(ctx, q.garbageKey, msgIds).Err()
    20.  if err != nil && err != redis.Nil {
    21.   return fmt.Errorf("remove from garbage key failed: %v", err)
    22.  }
    23.  return nil
    24. }

    之前提到的 lua 脚本都是原子性执行的，不会有其它命令插入其中。gc 函数由 3 条 redis 命令组成，在执行过程中可能会有其它命令插入执行过程中，不过考虑到一条消息进入垃圾回收流程之后不会复活所以不需要保证 3 条命令原子性。

    ack

    ack 只需要将消息彻底删除即可:

    1. func (q *DelayQueue) ack(idStr string) error {
    2.  ctx := context.Background()
    3.  err := q.redisCli.ZRem(ctx, q.unAckKey, idStr).Err()
    4.  if err != nil {
    5.   return fmt.Errorf("remove from unack failed: %v", err)
    6.  }
    7.  // msg key has ttl, ignore result of delete
    8.  _ = q.redisCli.Del(ctx, q.genMsgKey(idStr)).Err()
    9.  q.redisCli.HDel(ctx, q.retryCountKey, idStr)
    10.  return nil
    11. }

    否定确认只需要将 unack key 中消息的重试时间改为现在，随后执行的 unack2RetryScript 会立即将它移动到 retry key

    1. func (q *DelayQueue) nack(idStr string) error {
    2.  ctx := context.Background()
    3.  // update retry time as now, unack2Retry will move it to retry immediately
    4.  err := q.redisCli.ZAdd(ctx, q.unAckKey, &redis.Z{
    5.   Member: idStr,
    6.   Score:  float64(time.Now().Unix()),
    7.  }).Err()
    8.  if err != nil {
    9.   return fmt.Errorf("negative ack failed: %v", err)
    10.  }
    11.  return nil
    12. }

    consume

    消息队列的核心逻辑是每秒执行一次的 consume 函数，它负责调用上述脚本将消息转移到正确的集合中并回调 consumer 来消费消息：

    1. func (q *DelayQueue) consume() error {
    2.  // 执行 pending2ready，将已到时间的消息转移到 ready
    3.  err := q.pending2Ready()
    4.  if err != nil {
    5.   return err
    6.  }
    7.  // 循环调用 ready2Unack 拉取消息进行消费
    8.  var fetchCount uint
    9.  for {
    10.   idStr, err := q.ready2Unack()
    11.   if err == redis.Nil { // consumed all
    12.    break
    13.   }
    14.   if err != nil {
    15.    return err
    16.   }
    17.   fetchCount++
    18.   ack, err := q.callback(idStr)
    19.   if err != nil {
    20.    return err
    21.   }
    22.   if ack {
    23.    err = q.ack(idStr)
    24.   } else {
    25.    err = q.nack(idStr)
    26.   }
    27.   if err != nil {
    28.    return err
    29.   }
    30.   if fetchCount >= q.fetchLimit {
    31.    break
    32.   }
    33.  }
    34.  // 将 nack 或超时的消息放入重试队列
    35.  err = q.unack2Retry()
    36.  if err != nil {
    37.   return err
    38.  }
    39.     // 清理已达到最大重试次数的消息
    40.  err = q.garbageCollect()
    41.  if err != nil {
    42.   return err
    43.  }
    44.  // 消费重试队列
    45.  fetchCount = 0
    46.  for {
    47.   idStr, err := q.retry2Unack()
    48.   if err == redis.Nil { // consumed all
    49.    break
    50.   }
    51.   if err != nil {
    52.    return err
    53.   }
    54.   fetchCount++
    55.   ack, err := q.callback(idStr)
    56.   if err != nil {
    57.    return err
    58.   }
    59.   if ack {
    60.    err = q.ack(idStr)
    61.   } else {
    62.    err = q.nack(idStr)
    63.   }
    64.   if err != nil {
    65.    return err
    66.   }
    67.   if fetchCount >= q.fetchLimit {
    68.    break
    69.   }
    70.  }
    71.  return nil
    72. }

    至此一个简单可靠的延时队列就做好了，何不赶紧开始试用呢？

    如果本文对你有帮助，别忘记给我个3连 ，点赞，转发，评论，

    学习更多JAVA知识与技巧，关注博主学习JAVA 课件，源码，安装包，还有最新大厂面试资料等等等

    咱们下期见。

    收藏 等于白嫖，点赞才是真情。

     

     

  • 相关阅读:
    【多媒体信号处理课程】Course Introduction-1 AVI Walk Through-2 Audio coding basics-3 AI翻译
    性能分析优化的道与术
    MySQL高级篇知识点——索引优化与查询优化
    【智能家居入门2】（MQTT协议、微信小程序、STM32、ONENET云平台）
    day 1
    一文看懂Linux文件的属性
    Java学习笔记6.1.3 字节流 - 字节流缓冲区与缓冲字节流
    字符设备驱动_2：register_chrcev() 简单字符设备驱动的实现
    模板（详解）
    广州虚拟动力数字人实时驱动解决方案，赋能虚拟IP、虚拟直播、品牌发布会...
  • 原文地址：https://blog.csdn.net/uuqaz/article/details/126181313