极简消息队列：用 Redis List + Lua 脚本实现可靠投递与死信处理-港品优选

极简消息队列：用 Redis List + Lua 脚本实现可靠投递与死信处理

一、MQ 选型的困境：三个微服务不需要 Kafka

消息队列（MQ）在微服务架构中是核心基础设施。但当你的系统只有 3 个微服务、日均消息量不到 1 万条时，部署 Kafka 或 RabbitMQ 就像用大炮打蚊子——运维成本超过业务价值。

Redis 的 List 数据结构天然支持消息队列：LPUSH+BRPOP构成了生产者-消费者模型。但它缺少可靠投递（消息确认）、死信处理（处理失败的消息）、延迟队列等高级特性。这些可以通过 Lua 脚本补充。

Redis + Lua 构建的轻量消息队列的定位是：消息量 < 10 万/天、消费者数量 < 5、不需要消息持久化到磁盘（Redis 的 RDB/AOF 足够）。

graph TB P[生产者] -->|LPUSH| Q[Redis List<br/>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 步之间崩溃，消息已经丢失（已从队列弹出，但未处理完成）。解决方案是使用BRPOPLPUSH（Redis 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.RetryCount+1, 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 队列实现可靠投递；最后加入重试逻辑和死信队列。

少即是多。消息队列不需要"全家桶"——你需要的只是一个队列、一个处理中标记、一个死信兜底。

企业官网建设流程全解析