服务器之家：专注于VPS、云服务器配置技术及软件下载分享
分类导航

Mysql|Sql Server|Oracle|Redis|MongoDB|PostgreSQL|Sqlite|DB2|mariadb|Access|数据库技术|

服务器之家 - 数据库 - Redis - 百行代码实现基于Redis的可靠延迟队列

百行代码实现基于Redis的可靠延迟队列

2022-10-24 15:48Finley Redis

本文主要介绍了百行代码实现基于Redis的可靠延迟队列，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

在之前探讨延时队列的文章中我们提到了 redisson delayqueue 使用 redis 的有序集合结构实现延时队列，遗憾的是 go 语言社区中并无类似的库。不过问题不大，没有轮子我们自己造。

本文的完整代码实现在hdt3213/delayqueue，可以直接 go get 安装使用。

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

然而消息队列不是将消息发给消费者就万事大吉，它们还有一个重要职责是确保送达和消费。通常的实现方式是当消费者收到消息后向消息队列返回确认(ack)，若消费者返回否定确认(nack)或超时未返回，消息队列则会按照预定规则重新发送，直到到达最大重试次数后停止。如何实现 ack 和重试机制是我们要重点考虑的问题。

我们的消息队列允许分布式地部署多个生产者和消费者，消费者实例定时执行 lua 脚本驱动消息在队列中的流转无需部署额外组件。由于 Redis 保证了 lua 脚本执行的原子性，整个流程无需加锁。

消费者采用拉模式获得消息，保证每条消息至少投递一次，消息队列会重试超时或者被否定确认的消息(nack) 直至到达最大重试次数。一条消息最多有一个消费者正在处理，减少了要考虑的并发问题。

请注意：若消费时间超过了 MaxConsumeDuration 消息队列会认为消费超时并重新投递，此时可能有多个消费者同时消费。

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

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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.WithRetryCount(3))
        if err != nil {
            panic(err)
        }
    }
 
    // start consume
    done := queue.StartConsume()
    <-done
}

由于数据存储在 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, 值为剩余的重试次数

流程如下图所示：

百行代码实现基于Redis的可靠延迟队列

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

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

  • 都是原子性的
  • 不会重复处理同一条消息
  • 操作前后消息队列始终处于正确的状态

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

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

pending2ReadyScript

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

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

ready2UnackScript

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

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

unack2RetryScript

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

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

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

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

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

ack

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

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

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

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

consume

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

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

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

到此这篇关于百行代码实现基于Redis的可靠延迟队列的文章就介绍到这了,更多相关百行代码实现基于Redis的可靠延迟队列内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！

原文链接：https://www.cnblogs.com/Finley/p/16400287.html

  • redis
  • 延迟队列

    • 延伸 · 阅读

    精彩推荐
    • RedisRedis 中使用 list,streams,pub/sub 几种方式实现消息队列的问题

      Redis 中使用 list,streams,pub/sub 几种方式实现消息队列的问题

      这篇文章主要介绍了Redis 中使用 list,streams,pub/sub 几种方式实现消息队列,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借...

      Zhan-LiZ10782022-10-10
    • RedisRedis，就是这么朴实无华

      Redis，就是这么朴实无华

      Redis是2009年发布的，到今天已经超过10岁了。作为必备技能之一，关于它也有聊不完的话题。本文中的任何一个点，都可以展开，完成一篇中等规模的文章...

      小姐姐味道3422020-11-30
    • RedisNestJS+Redis实现缓存步骤详解

      NestJS+Redis实现缓存步骤详解

      这篇文章主要介绍了NestJS+Redis实现缓存,本文分步骤给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下...

      老胡Andy10652021-09-18
    • RedisJedis操作Redis实现模拟验证码发送功能

      Jedis操作Redis实现模拟验证码发送功能

      Redis是一个著名的key-value存储系统，也是nosql中的最常见的一种，这篇文章主要给大家介绍Jedis操作Redis实现模拟验证码发送功能，感兴趣的朋友一起看看吧...

      Andrew02195042021-11-18
    • RedisRedis使用bloom-filter过滤器实现推荐去重

      Redis使用bloom-filter过滤器实现推荐去重

      这篇文章主要介绍了Redis使用bloom-filter过滤器实现推荐去重,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友...

      -零13162021-01-03
    • RedisSpring boot+redis实现消息发布与订阅的代码

      Spring boot+redis实现消息发布与订阅的代码

      这篇文章主要介绍了Spring boot+redis实现消息发布与订阅,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值需要的朋友...

      源码分析8432020-05-31
    • Redis详解用Redis实现Session功能

      详解用Redis实现Session功能

      本篇文章主要介绍了用Redis实现Session功能，具有一定的参考价值，小编觉得挺不错的，现在分享给大家，也给大家做个参考。 ...

      Jlins5152019-11-01
    • Redis浅谈Redis高并发缓存架构性能优化实战

      浅谈Redis高并发缓存架构性能优化实战

      本文主要介绍了浅谈Redis高并发缓存架构性能优化实战，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友...

      枫度柚子5032022-10-17
    最近更新
    编辑推荐
    阅读排行
    热门标签
    16384 　 批量提权 　 intset 　 节点迁移 　 事务机制 　 主从同步 　 十六进制 　 内部数据结构 　 Redis集群 　 Redis集群搭建 　 持久化方案 　 AOF 　 ziplist 　 限流器 　 内存型数据库 　 RDB 　 简介 　 切片集群 　 批量读写 　 密码认证 　 redis命令 　 未授权 　 访问漏洞 　 redislive 　 发布 　 集合类型 　 列表类型 　 快照模式 　 集群搭建 　 键 　
    © 2019-2022 服务器之家(www.tuohang.net) 版权所有关于我们联系我们版权申明网站地图