极简消息队列：用 Redis List + Lua 脚本实现可靠投递与死信处理 极简消息队列用 Redis List Lua 脚本实现可靠投递与死信处理一、MQ 选型的困境三个微服务不需要 Kafka消息队列MQ在微服务架构中是核心基础设施。但当你的系统只有 3 个微服务、日均消息量不到 1 万条时部署 Kafka 或 RabbitMQ 就像用大炮打蚊子——运维成本超过业务价值。Redis 的 List 数据结构天然支持消息队列LPUSHBRPOP构成了生产者-消费者模型。但它缺少可靠投递消息确认、死信处理处理失败的消息、延迟队列等高级特性。这些可以通过 Lua 脚本补充。Redis Lua 构建的轻量消息队列的定位是消息量 10 万/天、消费者数量 5、不需要消息持久化到磁盘Redis 的 RDB/AOF 足够。graph TB P[生产者] --|LPUSH| Q[Redis Listbr/task:queue] C[消费者] --|BRPOP LPUSH| Q C --|处理成功| DONE[ACK: 从 processing 移除] C --|处理失败| RETRY{重试次数 3?} RETRY --|是| DELAY[延迟队列br/task:delay:N] RETRY --|否| DLQ[死信队列br/task:dead] DELAY --|到期| Q subgraph Monitoring[监控] M1[队列长度告警] M2[死信队列检查] M3[消费速率统计] end style DLQ fill:#ff6b6b,color:#fff style DONE fill:#51cf66,color:#fff二、原子性操作的必要性为什么 Lua 脚本是必须的Redis 的单个命令是原子的但多个命令的组合不是。考虑消费一条消息的过程BRPOP task:queue 0— 从队列弹出一条消息LPUSH task:processing:{msgId} message— 标记为处理中消息处理...DEL task:processing:{msgId}— 处理完成删除标记如果消费者在第 2 步和第 3 步之间崩溃消息已经丢失已从队列弹出但未处理完成。解决方案是使用BRPOPLPUSHRedis 6.2 后的BLMOVE原子地从源队列弹出并推入处理中队列。更复杂的逻辑如检查重试次数、计算延迟时间、判断是否进入死信队列需要多个 Redis 命令。这些必须用 Lua 脚本包装保证整个判断-操作链在一个原子事务中完成。三、Redis 消息队列的完整实现package taskq import ( context encoding/json fmt time github.com/redis/go-redis/v9 ) type Task struct { ID string json:id Type string json:type Data string json:data RetryCount int json:retry_count MaxRetries int json:max_retries CreatedAt int64 json:created_at } type Queue struct { client *redis.Client name string maxRetries int } func NewQueue(client *redis.Client, name string, maxRetries int) *Queue { return Queue{ client: client, name: name, maxRetries: maxRetries, } } // Enqueue 入队 func (q *Queue) Enqueue(ctx context.Context, task Task) error { task.CreatedAt time.Now().Unix() if task.MaxRetries 0 { task.MaxRetries q.maxRetries } data, err : json.Marshal(task) if err ! nil { return fmt.Errorf(marshal task: %w, err) } return q.client.LPush(ctx, q.queueKey(), data).Err() } // Dequeue 出队阻塞式同时原子地将消息移到 processing 队列 func (q *Queue) Dequeue(ctx context.Context, timeout time.Duration) (*Task, error) { result, err : q.client.BLMove(ctx, q.queueKey(), q.processingKey(), RIGHT, LEFT, timeout).Result() if err ! nil { if err redis.Nil { return nil, nil // 超时无消息 } return nil, fmt.Errorf(blmove: %w, err) } var task Task if err : json.Unmarshal([]byte(result), task); err ! nil { // 无法解析的消息移到死信队列 q.client.LPush(ctx, q.deadKey(), result) return nil, fmt.Errorf(unmarshal task: %w, err) } return task, nil } // ACK 确认消息处理成功 func (q *Queue) ACK(ctx context.Context, task Task) error { data, _ : json.Marshal(task) return q.client.LRem(ctx, q.processingKey(), 0, string(data)).Err() } // NACK 处理失败根据重试次数决定是重试还是进入死信队列 func (q *Queue) NACK(ctx context.Context, task Task) error { task.RetryCount if task.RetryCount task.MaxRetries { // 进入死信队列 data, _ : json.Marshal(task) if err : q.client.LPush(ctx, q.deadKey(), data).Err(); err ! nil { return fmt.Errorf(push to dead queue: %w, err) } // 从 processing 移除 data2, _ : json.Marshal(task) return q.client.LRem(ctx, q.processingKey(), 0, string(data2)).Err() } // 延迟重试指数退避 delaySeconds : int64(1 task.RetryCount) // 1s, 2s, 4s, 8s... if delaySeconds 3600 { delaySeconds 3600 // 最大 1 小时 } // 使用 Lua 脚本原子操作 script : redis.NewScript( local msg ARGV[1] local processingKey KEYS[1] local queueKey KEYS[2] local delayKey KEYS[3] local delaySeconds tonumber(ARGV[2]) -- 从 processing 中移除 redis.call(LREM, processingKey, 0, msg) -- 加入延迟队列sorted set local executeAt redis.call(TIME)[1] delaySeconds redis.call(ZADD, delayKey, executeAt, msg) return OK ) data, _ : json.Marshal(task) return script.Run(ctx, q.client, []string{q.processingKey(), q.queueKey(), q.delayKey()}, string(data), delaySeconds, ).Err() } // ProcessDelayed 将到期的延迟消息重新入队 func (q *Queue) ProcessDelayed(ctx context.Context) (int64, error) { script : redis.NewScript( local delayKey KEYS[1] local queueKey KEYS[2] local now redis.call(TIME)[1] -- 获取所有到期的消息 local messages redis.call(ZRANGEBYSCORE, delayKey, 0, now) if #messages 0 then return 0 end -- 移到主队列 for i, msg in ipairs(messages) do redis.call(LPUSH, queueKey, msg) end -- 从延迟队列中移除 redis.call(ZREMRANGEBYSCORE, delayKey, 0, now) return #messages ) result, err : script.Run(ctx, q.client, []string{q.delayKey(), q.queueKey()}).Result() if err ! nil { return 0, err } return result.(int64), nil } // DeadLetterLen 返回死信队列长度 func (q *Queue) DeadLetterLen(ctx context.Context) (int64, error) { return q.client.LLen(ctx, q.deadKey()).Result() } // RetryDead 将死信队列中的消息重新入队 func (q *Queue) RetryDead(ctx context.Context, count int) (int64, error) { moved : int64(0) for i : 0; i count; i { msg, err : q.client.RPopLPush(ctx, q.deadKey(), q.queueKey()).Result() if err ! nil { break } // 重置重试次数 var task Task json.Unmarshal([]byte(msg), task) task.RetryCount 0 data, _ : json.Marshal(task) q.client.LRem(ctx, q.queueKey(), 0, msg) q.client.LPush(ctx, q.queueKey(), data) moved } return moved, nil } func (q *Queue) queueKey() string { return fmt.Sprintf(taskq:%s:queue, q.name) } func (q *Queue) processingKey() string { return fmt.Sprintf(taskq:%s:processing, q.name) } func (q *Queue) deadKey() string { return fmt.Sprintf(taskq:%s:dead, q.name) } func (q *Queue) delayKey() string { return fmt.Sprintf(taskq:%s:delay, q.name) }消费者示例func StartConsumer(ctx context.Context, q *Queue, handler func(Task) error) { // 延迟消息处理协程 go func() { ticker : time.NewTicker(1 * time.Second) defer ticker.Stop() for range ticker.C { count, _ : q.ProcessDelayed(ctx) if count 0 { log.Printf(Processed %d delayed tasks, count) } } }() // 主消费循环 for { task, err : q.Dequeue(ctx, 5*time.Second) if err ! nil { log.Printf(Dequeue error: %v, err) continue } if task nil { continue } if err : handler(*task); err ! nil { log.Printf(Task %s failed (retry %d/%d): %v, task.ID, task.RetryCount1, task.MaxRetries, err) q.NACK(ctx, *task) } else { q.ACK(ctx, *task) } } }四、这个 MQ 的保障边界At-Least-Once 保证通过BLMOVE原子操作消息绝不会丢失。但消费者可能重复处理同一条消息处理成功但 ACK 失败。消费者需要实现幂等性。不支持多消费者组Redis List 是简单的 FIFO 队列每条消息只能被一个消费者处理。如果需要广播同一条消息给多个消费者改用 Redis Pub/Sub。持久化依赖消息持久化依赖 Redis 的 RDB/AOF。如果 Redis 崩溃且 AOF 未同步到磁盘可能丢失最近的消息。扩展上限单实例 Redis 的吞吐量上限约 10 万 QPS。超过这个量级需要考虑 Redis Cluster 或迁移到专业 MQ。五、总结Redis List Lua 脚本构建的消息队列用不到 200 行 Go 代码实现了可靠投递、指数退避重试、死信队列三个核心能力。对于日均消息量 10 万、消费者 5 的场景这比部署 Kafka 更务实。落地路径先用 Redis List 实现最基本的Enqueue Dequeue再引入BLMOVE和 processing 队列实现可靠投递最后加入重试逻辑和死信队列。少即是多。消息队列不需要全家桶——你需要的只是一个队列、一个处理中标记、一个死信兜底。