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

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

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

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

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

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

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

具体使用也非常简单,只需要注册处理消息的回调函数并调用 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.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, 值为剩余的重试次数

流程如下图所示:

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

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

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

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

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

pending2ReadyScript

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

-- 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:

-- 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 中:

-- 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 命令将他们删除:

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 只需要将消息彻底删除即可:

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

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 来消费消息:

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实现延迟队列的正确姿势的更多相关文章

  1. 【分布式缓存系列】Redis实现分布式锁的正确姿势

    一.前言 在我们日常工作中,除了Spring和Mybatis外,用到最多无外乎分布式缓存框架——Redis.但是很多工作很多年的朋友对Redis还处于一个最基础的使用和认识.所以我就像把自己对分布式缓 ...

  2. Redis实现分布式锁的正确姿势

    分布式锁一般有三种实现方式:1. 数据库乐观锁:2. 基于Redis的分布式锁:3. 基于ZooKeeper的分布式锁.本篇博客将介绍第二种方式,基于Redis实现分布式锁.虽然网上已经有各种介绍Re ...

  3. 基于Redis实现延迟队列

    背景 在后端服务中,经常有这样一种场景,写数据库操作在异步队列中执行,且这个异步队列是多进程运行的,这时如果对同一资源进行写库操作,很有可能产生数据被覆盖等问题,于是就需要业务层在更新数据库之前进行加 ...

  4. [线段树系列] LCT打延迟标记的正确姿势

    这一篇博客将教你什么? 如何用LCT打延迟标记,LCT和线段树延迟标记间的关系,为什么延迟标记要这样打. ——正片开始—— 学习这一篇博客前,确保你会以下知识: Link-Cut-Tree,普通线段树 ...

  5. 【分布式缓存系列】集群环境下Redis分布式锁的正确姿势

    一.前言 在上一篇文章中,已经介绍了基于Redis实现分布式锁的正确姿势,但是上篇文章存在一定的缺陷——它加锁只作用在一个Redis节点上,如果通过sentinel保证高可用,如果master节点由于 ...

  6. 解锁redis锁的正确姿势

    解锁redis锁的正确姿势 redis是php的好朋友,在php写业务过程中,有时候会使用到锁的概念,同时只能有一个人可以操作某个行为.这个时候我们就要用到锁.锁的方式有好几种,php不能在内存中用锁 ...

  7. 基于redis的延迟消息队列设计

    需求背景 用户下订单成功之后隔20分钟给用户发送上门服务通知短信 订单完成一个小时之后通知用户对上门服务进行评价 业务执行失败之后隔10分钟重试一次 类似的场景比较多 简单的处理方式就是使用定时任务 ...

  8. php使用redis的有序集合zset实现延迟队列

    延迟队列就是个带延迟功能的消息队列,相对于普通队列,它可以在指定时间消费掉消息. 延迟队列的应用场景: 1.新用户注册,10分钟后发送邮件或站内信. 2.用户下单后,30分钟未支付,订单自动作废. 我 ...

  9. 基于redis的延迟消息队列设计(转)

    需求背景 用户下订单成功之后隔20分钟给用户发送上门服务通知短信 订单完成一个小时之后通知用户对上门服务进行评价 业务执行失败之后隔10分钟重试一次 类似的场景比较多 简单的处理方式就是使用定时任务 ...

  10. Redis全方位详解--数据类型使用场景和redis分布式锁的正确姿势

    一.Redis数据类型 1.string string是Redis的最基本数据类型,一个key对应一个value,每个value最大可存储512M.string一半用来存图片或者序列化的数据. 2.h ...

随机推荐

  1. Vue.js学习 Item12 – 内部响应式原理探究

    深入响应式原理 大部分的基础内容我们已经讲到了,现在讲点底层内容.Vue.js 最显著的一个功能是响应系统 —— 模型只是普通对象,修改它则更新视图.这让状态管理非常简单且直观,不过理解它的原理也很重 ...

  2. js运动 多数据运动 含JSON

    <!DOCTYPE HTML> <html> <head> <meta http-equiv="Content-Type" content ...

  3. python网络编程——将IPv4地址转换成不同的格式

    1.将IPv4地址转换为32位二进制格式,用做底层网络函数. import socket from binascii import hexlify def convert_IPv4_address() ...

  4. Python - 缩写(capwords) 和 创建转换表(maketrans) 详细说明

    缩写(capwords) 和 创建转换表(maketrans) 详细说明 本文地址: http://blog.csdn.net/caroline_wendy/article/details/27051 ...

  5. bzoj 4835: 遗忘之树 [树形DP]

    4835: 遗忘之树 题意:点分治,选标号最小的重心,上一次重心向下一次重心连有向边,求原树方案数. md我真不知道当初比赛时干什么去了...现在一眼秒啊... \(size[v]=\frac{siz ...

  6. .NET Core 事件总线,分布式事务解决方案:CAP 基于Kafka

    背景 相信前面几篇关于微服务的文章也介绍了那么多了,在构建微服务的过程中确实需要这么一个东西,即便不是在构建微服务,那么在构建分布式应用的过程中也会遇到分布式事务的问题,那么 CAP 就是在这样的背景 ...

  7. Python中元类

    元类(metaclass) 简单地说,元类就是一个能创建类的类,而类class 是由type创建的,class可以创建对象 type与object的关系详见:python中type和object 1. ...

  8. linux客户端WinSCP

    WinSCP是一个Windows环境下使用SSH的开源图形化SFTP客户端.同时支持SCP协议.它的主要功能就是在本地与远程计算机间安全的复制文件.   这是一个中文版的介绍.从这里链接出去的大多数文 ...

  9. Windows中snmputil.exe工具的使用

    一.检查windows 系统是否安装了 snmp 组件.可以在 cmd 中输入"net start snmp" 就可以出现显示信息了 二. snmputil,就是程序名拉,呵呵. ...

  10. cakephp中使用大括号的形式避免用点号连接sql语句

    在cakephp中可以使用{}的形式来代替点号连接sql语句,减少出错的几率