Golang实现基于Redis的可靠延迟队列
在Go语言生态中，相比Java的Redisson，原生的Redis延迟队列库较为稀缺。本文分享一种基于Redis数据结构实现的可靠延迟队列方案，利用Lua脚本保障操作原子性，配合ACK/NACK确认机制，完美解决分布式场景下消息重复消费或丢失的问题。
核心设计与架构目标
在设计一个可靠的分布式消息队列时，核心挑战在于如何在无锁环境下保证状态的一致性。本方案允许多个生产者和消费者实例分布式部署，通过定时执行Lua脚本来驱动消息流转。由于Redis执行Lua脚本具有原子性，整个过程无需依赖额外的分布式锁服务。
核心机制说明：
- 拉模式消费：消费者主动拉取消息，便于控制流量和并发度。
- 至少投递一次（At-least-once）：若消费者返回肯定确认（ACK），消息被彻底删除；若返回否定确认（NACK）或超时未回复，消息将被重新投递。
- 并发安全：一条消息在同一时刻只会被分配给一个消费者处理，除非消费时间超过了预设的
MaxConsumeDuration，此时系统判定为超时并重新投递，可能引发并发消费。
以下为完整的使用示例，仅需注册回调函数并调用 StartConsume() 即可启动队列：
package main
import (
"github.com/go-redis/redis/v8"
"github.com/hdt3213/delayqueue"
"strconv"
"time"
)
func main() {
// 初始化 Redis 客户端
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
})
// 发送延时消息（1小时后投递，最多重试3次）
for i := 0; i < 10; i++ {
err := queue.SendDelayMsg(strconv.Itoa(i), time.Hour, delayqueue.WithRetryCount(3))
if err != nil {
panic(err)
}
}
// 启动消费者协程
done := queue.StartConsume()
<-done
}
数据可靠性边界：由于数据存储在Redis中，只要Redis节点故障率极低且相关Key未被外部恶意篡改，该方案能保证消息不丢失。
底层数据结构设计
为了实现复杂的流转状态，我们组合使用多种Redis数据结构。每条消息在内部由唯一的 UUID 标识，消息体存储在独立的 String Key 中以利用 TTL 自动清理。
- msgKey (String): 存储实际的消息内容（Payload）。每个消息ID对应一个独立Key，便于利用Redis的过期策略。
- pendingKey (ZSet): 有序集合。Member为消息ID，Score为预期投递时间的Unix时间戳。用于存储尚未到期的延迟消息。
- readyKey (List): 列表。存储已达到投递时间、准备分发给消费者的消息ID。
- unAckKey (ZSet): 有序集合。Member为消息ID，Score为下次重试时间。存储正在被消费者处理但尚未确认的消息。
- retryKey (List): 列表。存储已准备好进行重试的消息ID。
- garbageKey (Set): 集合。暂存已达到最大重试次数、需要被彻底清理的消息ID。
- retryCountKey (Hash): 哈希表。键为消息ID，值为剩余的重试次数。
如上图所示，消息在不同Key之间流转。由于多实例并发运行，操作顺序不可预知，因此所有关键转换操作必须满足三个条件：原子性、幂等性（不重复处理）以及状态机始终正确。
Lua脚本驱动的状态流转
我们将复杂的CRUD操作封装为Lua脚本，由Redis原生执行以保证原子性。
第1步：Pending转Ready
目的：扫描已到期的延迟消息，将其移至Ready列表，等待消费者拉取。
工具：ZRangeByScore 扫描 ZRemRangeByScore 移除。
-- keys: pendingKey, readyKey
-- argv: currentTime
local msgs = redis.call('ZRangeByScore', KEYS[1], '0', ARGV[1])
if (#msgs == 0) then return end
-- 将消息推入 Ready List
local args2 = {'LPush', KEYS[2]}
for _,v in ipairs(msgs) do
table.insert(args2, v)
end
redis.call(unpack(args2))
-- 从 Pending ZSet 中移除已处理的消息
redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1])
第2步：Ready转Unack
目的：从Ready或Retry列表中取出一条消息，标记为“正在处理”（放入Unack），防止其他实例重复拉取。
工具：RPop 弹出，ZAdd 加入有序集合。
-- 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
第3步：Unack转Retry
目的：检查长时间未确认（超时）的消息。根据剩余重试次数，决定是重新投入重试队列，还是移入垃圾回收区。
注意事项：由于Redis Lua限制，无法动态访问未知的 msgKey。因此，脚本仅更新状态，并将需要物理删除的消息ID记入 garbageKey。
-- 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
-- 还有重试机会：扣减次数，丢入 Retry List
redis.call("HIncrBy", KEYS[2], k, -1)
redis.call("LPush", KEYS[3], k)
else
-- 超过最大重试次数：清理计数，丢入 Garbage Set 待删除
redis.call("HDel", KEYS[2], k)
redis.call("SAdd", KEYS[4], k)
end
end
-- 从未确认集合中移除这些消息
redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1])
垃圾回收与确认机制
ACK 确认（成功）
当业务逻辑处理成功后，调用 ack 方法。这会从 unAckKey 移除记录，并从Redis中彻底删除 msgKey。
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)
}
// 删除实际消息数据，忽略非Nil错误
_ = q.redisCli.Del(ctx, q.genMsgKey(idStr)).Err()
q.redisCli.HDel(ctx, q.retryCountKey, idStr)
return nil
}
NACK 否定确认（失败）
当业务处理失败时，调用 nack 方法。它将消息在 unAckKey 中的 Score 更新为当前时间，在下一次轮询时立即触发 unack2RetryScript 进行重试。
func (q *DelayQueue) nack(idStr string) error {
ctx := context.Background()
// 更新时间戳，使得下一次 consume 周期能立刻捕获并重投
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
}
垃圾回收（Garbage Collect）
对于达到最大重试次数仍失败的消息，它们被标记在 garbageKey 中。我们需要定期清理这些无效数据和对应的 Payload。
func (q *DelayQueue) garbageCollect() error {
ctx := context.Background()
msgIds, err := q.redisCli.SMembers(ctx, q.garbageKey).Result()
if err != nil || len(msgIds) == 0 {
return err
}
// 收集对应的实际消息 Key
msgKeys := make([]string, 0, len(msgIds))
for _, idStr := range msgIds {
msgKeys = append(msgKeys, q.genMsgKey(idStr))
}
// 批量删除消息体和ID记录
err = q.redisCli.Del(ctx, msgKeys...).Err()
if err != nil && err != redis.Nil {
return fmt.Errorf("del msgs failed: %v", err)
}
return q.redisCli.SRem(ctx, q.garbageKey, msgIds).Err()
}
调度循环主流程
整个系统的核心是每秒执行的 consume 循环。它协调了上述所有脚本的执行：
func (q *DelayQueue) consume() error {
// 1. 到期消息移入就绪队列
err := q.pending2Ready()
if err != nil { return err }
// 2. 拉取新消息进行处理
var fetchCount uint
for {
idStr, err := q.ready2Unack()
if err == redis.Nil { break }
if err != nil { return err }
fetchCount++
ack, callbackErr := q.callback(idStr)
if callbackErr != nil { return callbackErr }
// 根据结果确认或重投
if ack { err = q.ack(idStr) } else { err = q.nack(idStr) }
if err != nil { return err }
if fetchCount >= q.fetchLimit { break }
}
// 3. 处理超时/失败消息的重试逻辑
err = q.unack2Retry()
if err != nil { return err }
// 4. 清理彻底失败的消息
err = q.garbageCollect()
if err != nil { return err }
// 5. （可选）处理专门的重试队列...
return nil
}
以上就是Golang实现基于Redis的可靠延迟队列的详细内容，更多关于Golang Redis可靠延迟队列的资料请关注本站其它相关文章！
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
