夜雨聆风Redis 系列第一篇提到 Stream 是"Redis 5.0 加的，专门用来做消息队列"。当时只是列了一下数据类型，没有展开。
但 Streams 可能是 Redis 里最被低估的数据结构。
大部分人知道 Redis 能做消息队列——用 List 的 LPUSH + BRPOP。确实能用，但 List 有几个致命问题：
• 消息确认（ACK）机制。 BRPOP把消息弹出来就没了。如果消费者拿到消息后崩了，消息就丢了。• 消费者组。List 没法让多个消费者分担同一个队列的消息——要么每个消费者各拿一条（竞争消费），要么每个消费者都拿全部（广播），但做不到"一组消费者分摊处理，每条消息只被一个消费者处理"。 • 消息回溯。 BRPOP弹出就删，没法回溯。新加入的消费者看不到历史消息。
Streams 把这些问题全解决了。 它是 Redis 内置的、专门为消息队列设计的数据结构，支持消费者组、消息确认、pending 列表（PEL）、超时消息接管（XAUTOCLAIM）。如果你的场景不需要 Kafka 那种级别的吞吐和持久化保证，Streams 是一个极好的轻量级替代。
这篇文章的目标：用 Go 实现一个完整的、可以处理故障恢复的事件消费器。
如果你看过后面那篇 Outbox，会发现两篇文章刚好能接上：Outbox 解决的是"业务数据提交后，事件怎么可靠投递出去"；这篇 Streams 解决的是"事件到了队列后，消费者怎么可靠处理完"。一个偏生产端，一个偏消费端，真实项目里经常配在一起用。
完整代码在 redis-streams-consumer[1]。
在写代码之前，得先搞清楚 Streams 的几个核心概念。我用后端常见的类比来解释。
Stream ≈ 一个只追加的日志
Stream 就是一个有序的消息列表。每条消息有一个自动生成的 ID（时间戳-序号格式，比如 1747382400000-0），和一组键值对作为消息内容。
XADD events * type order.created order_id ORD-0001 amount 99.99# 返回: "1747382400000-0"* 表示让 Redis 自动生成 ID。消息一旦写入就不会消失（除非被 XDEL 或 MAXLEN 裁剪）。
类比 Kafka 的 topic：Stream 像一个 Kafka topic 的单 partition。消息有序、持久化、可回溯。区别是 Streams 在 Redis 内存里，Kafka 在磁盘上。
Consumer Group ≈ Kafka 的消费者组
这是 Streams 最核心的概念。
XGROUP CREATE events worker-group 0创建一个消费者组 worker-group，从 ID 0（最早的消息）开始消费。如果用 $ 代替 0，表示只消费创建组之后的新消息。
一个 Stream 可以有多个消费者组，每个组独立追踪消费进度。组内的多个消费者竞争消费——每条消息只会被组内的一个消费者拿到。
Stream: events ├── Consumer Group: worker-group │ ├── consumer-1 (处理消息 1, 3, 5) │ └── consumer-2 (处理消息 2, 4, 6) └── Consumer Group: analytics-group └── consumer-a (处理所有消息)这正好解决了 List 做不到的事：worker-group 里的两个消费者分担处理，每条消息只处理一次；analytics-group 独立消费同一份数据，互不影响。
XREADGROUP：消费者读消息
XREADGROUP GROUP worker-group consumer-1 COUNT 10 BLOCK 5000 STREAMS events >这行命令的意思是：消费者 consumer-1 从 worker-group 中读取最多 10 条新消息（> 表示还没有被分配给任何消费者的新消息），最多阻塞等待 5 秒。
> 是关键。 它告诉 Redis"给我没人领过的新消息"。如果你把 > 换成 0，Redis 会返回这个消费者已经领了但还没 ACK 的消息（pending 消息）。这个区别后面会用到。
XACK：确认消息已处理
XACK events worker-group 1747382400000-0告诉 Redis：消息 1747382400000-0 已经被成功处理了，可以从 pending 列表中移除。
不 ACK 的后果：消息会一直留在 PEL（Pending Entries List）中。 PEL 是 Streams 的核心数据结构之一，它记录了所有"已分配给消费者但还没被确认"的消息。
PEL（Pending Entries List）：未确认消息的追踪器
PEL 就是一个"待办事项"清单。每条消息被 XREADGROUP 读取后，就进入 PEL。XACK 后，从 PEL 移除。
XPENDING events worker-group - + 10# 返回 PEL 中的消息列表：ID、消费者、空闲时间、投递次数PEL 的存在让 Streams 有了至少一次（at-least-once）投递语义。消费者崩了？没关系，消息还在 PEL 里。重启后可以从 PEL 恢复未完成的工作。
但"至少一次"也意味着消息可能被处理多次——比如消费者处理了一半崩了，消息还在 PEL 中，恢复后会被重新处理。所以下游必须做幂等。这个后面会详细讲。
XAUTOCLAIM：接管超时未确认的消息
如果一个消费者彻底挂了（不是临时崩溃，而是永久下线），它 PEL 中的消息就永远不会被处理了。
XAUTOCLAIM 解决这个问题：从 PEL 中接管闲置超过指定时间的消息。
XAUTOCLAIM events worker-group consumer-2 30000 0-0 COUNT 10消费者 consumer-2 从 worker-group 的 PEL 中抢回闲置超过 30 秒的消息（最多 10 条）。抢回的消息会转移到 consumer-2 名下，相当于"我来接手处理这些没人管的消息"。
类比数据库的事务超时：一个事务开始了但超过一定时间还没提交，系统自动回滚。PEL + XAUTOCLAIM 的效果类似——消息被领走但超时没 ACK，其他消费者自动接管。
好，概念讲完了。从生产者开始写代码。
生产者：XADD + 幂等 ID
生产者的职责很简单：把事件写入 Stream。但有一个关键设计：每条消息必须携带一个业务级别的唯一 ID（event_id）。
为什么不用 Redis 自动生成的 Stream ID？因为 Stream ID 是 Redis 内部的概念，消费者端无法用它做幂等。如果同一条业务事件因为生产者重试被写入两次，它们的 Stream ID 不同，但 event_id 相同——消费者靠 event_id 判断"这条我处理过了"。
type Producer struct { rdb *redis.Client streamKey string maxLen int64 logger *zap.Logger}func (p *Producer) Publish(ctx context.Context, fields map[string]interface{}) (string, error) { if _, ok := fields["event_id"]; !ok { return "", fmt.Errorf("缺少 event_id 字段，无法保证幂等") } id, err := p.rdb.XAdd(ctx, &redis.XAddArgs{ Stream: p.streamKey, MaxLen: p.maxLen, Approx: true, // ~ 近似裁剪，性能更好 Values: fields, }).Result() if err != nil { return "", fmt.Errorf("XADD 失败: %w", err) } return id, nil}几个细节：
1. MaxLen+Approx: true。Stream 会无限增长，需要设上限。MaxLen: 10000表示保留最近 10000 条消息，Approx表示用近似裁剪（MAXLEN ~）——Redis 不会精确到刚好 10000 条，可能多保留一些，但性能比精确裁剪好得多。近似裁剪避免了每次 XADD 都做 O(N) 的删除操作。2. 强制要求 event_id。如果调用方没传，直接报错。这不是"防御性编程"——没有event_id的消息，下游没法做幂等，等于埋了一个必然会爆的雷。
发布消息的调用方：
eventID := uuid.New().String()fields := map[string]interface{}{ "event_id": eventID, "type": "order.created", "order_id": "ORD-0001", "amount": "99.99", "ts": time.Now().UnixMilli(),}id, err := producer.Publish(ctx, fields)运行生产者（cmd/producer）会往 Stream 写入 20 条模拟订单事件：
cd redis-streams-consumercp .env.example .envgo run cmd/producer/main.go# 已发布: order.paid → 1747382400123-0 (event_id=a1b2c3d4)# 已发布: order.shipped → 1747382400124-0 (event_id=e5f6g7h8)# ...# 发布完成。运行 go run cmd/consumer 开始消费。消费者：XREADGROUP + XACK
消费者的核心循环只有两步：读消息、ACK 消息。但启动时有一个容易被忽略的问题：之前崩掉的消费者可能留下了未 ACK 的消息。
新 consumer name vs 旧 pending
我们的代码每次启动生成一个新的 consumerName（hostname + UUID 后缀）。这意味着 XREADGROUP 用 ID=0 只能读到当前 consumerName 自己的 pending 消息——旧进程用不同的 name，它的 pending 消息对新进程不可见。
那旧消息谁来处理？答案是 Claimer（XAUTOCLAIM）。 旧消费者的 pending 消息闲置超过 minIdle 后，会被 Claimer 抢回到当前消费者名下。
这是一个刻意的设计选择：
• 固定 consumerName（比如固定用 hostname）：重启后 ReadPending 能直接恢复自己的旧消息，但多个进程如果 hostname 相同就会冲突。 • 动态 consumerName（我们的方案）：依赖 Claimer 回收旧消息，恢复有 minIdle的延迟，但天然支持多实例水平扩展。
ReadPending 在我们的方案里主要处理的是：显式配置固定 CONSUMER_NAME 后，当前 consumerName 自己留下的 pending 消息。 如果使用默认的动态 consumerName，跨进程的旧 pending 消息主要交给 Claimer。
type Consumer struct { rdb *redis.Client streamKey string groupName string consumerName string batchSize int64 blockTime time.Duration logger *zap.Logger}// ReadPending 读取自己未 ACK 的消息（ID = "0"）func (c *Consumer) ReadPending(ctx context.Context) ([]redis.XMessage, error) { streams, err := c.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ Group: c.groupName, Consumer: c.consumerName, Streams: []string{c.streamKey, "0"}, Count: c.batchSize, }).Result() if err != nil { if err == redis.Nil { return nil, nil } return nil, err } if len(streams) == 0 || len(streams[0].Messages) == 0 { return nil, nil } return streams[0].Messages, nil}// ReadNew 读取新消息（ID = ">"）func (c *Consumer) ReadNew(ctx context.Context) ([]redis.XMessage, error) { streams, err := c.rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ Group: c.groupName, Consumer: c.consumerName, Streams: []string{c.streamKey, ">"}, Count: c.batchSize, Block: c.blockTime, }).Result() // ...}// Ack 确认消息已处理func (c *Consumer) Ack(ctx context.Context, ids ...string) error { return c.rdb.XAck(ctx, c.streamKey, c.groupName, ids...).Err()}注意 ReadPending 不设 Block——pending 消息要么有要么没有，不需要等。ReadNew 设了 Block: 5s——没有新消息时阻塞等 5 秒再返回，避免空转浪费 CPU。
Block 是 Streams 的杀手级特性之一。 它让你不需要写 time.Sleep 来做轮询——Redis 帮你阻塞，有消息立即返回。比 BRPOP 更强的是，它可以一次读多条（Count），减少网络往返。
ACK 的时机
我们的实现是 "处理成功后 ACK"：
err := handler.Handle(ctx, msg)if err != nil { // 不 ACK，留在 PEL 中等待回收 continue}consumer.Ack(ctx, msg.ID)处理失败不 ACK。 消息留在 PEL 中，等 XAUTOCLAIM 回收后重新处理。这给了我们自动重试的能力——不需要自己写重试逻辑。
另一种选择是"读取后立即 ACK"——适合消息丢失可接受的场景（比如日志收集、指标统计）。处理是否成功不影响消息确认。吞吐更高，但可靠性低。
大部分业务场景用"处理成功后 ACK"。
超时消息接管：XAUTOCLAIM
消费者崩了，消息留在 PEL 里。怎么让其他消费者接管？
Redis 6.2 之前用 XPENDING + XCLAIM 两步操作——先查 PEL 找到超时消息，再用 XCLAIM 抢过来。Redis 6.2 引入了 XAUTOCLAIM，一步搞定。
type Claimer struct { rdb *redis.Client streamKey string groupName string consumerName string minIdle time.Duration // 最小闲置时间 interval time.Duration // 检查周期 nextStart string logger *zap.Logger}func NewClaimer(rdb *redis.Client, streamKey, groupName, consumerName string, minIdle, interval time.Duration, logger *zap.Logger) *Claimer { return &Claimer{ rdb: rdb, streamKey: streamKey, groupName: groupName, consumerName: consumerName, minIdle: minIdle, interval: interval, nextStart: "0-0", logger: logger, }}func (cl *Claimer) Run(ctx context.Context, msgCh chan<- redis.XMessage) { ticker := time.NewTicker(cl.interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: cl.claim(ctx, msgCh) } }}func (cl *Claimer) claim(ctx context.Context, msgCh chan<- redis.XMessage) { msgs, nextStart, err := cl.rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ Stream: cl.streamKey, Group: cl.groupName, Consumer: cl.consumerName, MinIdle: cl.minIdle, Start: cl.nextStart, Count: 10, }).Result() if err != nil { cl.logger.Warn("XAUTOCLAIM 失败", zap.Error(err)) return } cl.nextStart = nextStart if len(msgs) > 0 { cl.logger.Info("回收到超时消息", zap.Int("count", len(msgs))) for _, msg := range msgs { select { case msgCh <- msg: case <-ctx.Done(): return } } }}Claimer 是一个后台 goroutine，每隔 interval（比如 10 秒）执行一次 XAUTOCLAIM。MinIdle: 30s 表示只抢回闲置超过 30 秒的消息——不会抢走正在被别人处理的消息。
这里不要每次都从 "0-0" 开始。XAUTOCLAIM 会返回下一次扫描应该继续的位置，也就是 nextStart。把它记下来，下次从这个游标继续扫，PEL 很大时就不会反复检查同一批还没到 minIdle 的消息。扫到末尾时 Redis 会返回 "0-0"，下一轮自然从头开始。
minIdle 的设置很关键。 设太小，消息还在被正常处理就被抢走了（导致重复处理）。设太大，消费者崩了之后等太久才恢复。经验值是正常处理时间的 5-10 倍。如果你的消息平均处理时间是 3 秒，minIdle 设 30 秒比较合理。
幂等：消息可能被处理多次
"至少一次"投递语义意味着消息可能被处理多次。两种场景：
1. 消费者处理成功了但 ACK 失败（网络抖动）。消息留在 PEL 中，会被重新投递。 2. XAUTOCLAIM 抢回了一条正在被慢处理的消息。原消费者和新消费者各处理一次。
所以消费者必须做幂等。最简单的方案：用 Redis SET NX 做去重。
type IdempotentHandler struct { rdb *redis.Client inner EventHandler ttl time.Duration logger *zap.Logger}func (h *IdempotentHandler) Handle(ctx context.Context, msg redis.XMessage) error { eventID, ok := msg.Values["event_id"] if !ok { return fmt.Errorf("消息 %s 缺少 event_id", msg.ID) } // SETNX: 只有第一次设置成功的消费者才处理 dedupeKey := fmt.Sprintf("processed:%v", eventID) set, err := h.rdb.SetNX(ctx, dedupeKey, "1", h.ttl).Result() if err != nil { return fmt.Errorf("幂等检查失败: %w", err) } if !set { h.logger.Debug("跳过重复消息", zap.String("event_id", fmt.Sprint(eventID)), zap.String("msg_id", msg.ID), ) return nil // 已处理过，静默跳过 } if err := h.inner.Handle(ctx, msg); err != nil { // 处理失败，回滚去重标记，让 XAUTOCLAIM 捞回来后能重新处理 if rollbackErr := h.rdb.Del(ctx, dedupeKey).Err(); rollbackErr != nil { return fmt.Errorf("处理失败且回滚幂等标记失败: %w", errors.Join(err, rollbackErr)) } return err } return nil}逻辑很简单：处理消息前先用 SETNX 尝试写一个去重 key。写成功了说明是第一次处理，继续。写失败了说明之前已经处理过了，跳过。
但这里有一个微妙的问题：SETNX 成功但 inner.Handle 失败了怎么办？ 去重 key 已经设了，下次被 XAUTOCLAIM 捞回来后会被当成"已处理"而跳过——消息就丢了。
所以 Handle 失败时必须回滚去重标记：
if err := h.inner.Handle(ctx, msg); err != nil { if rollbackErr := h.rdb.Del(ctx, dedupeKey).Err(); rollbackErr != nil { return fmt.Errorf("处理失败且回滚幂等标记失败: %w", errors.Join(err, rollbackErr)) } return err}return nil代码里就是这么做的。回滚后，消息留在 PEL 中不 ACK，Claimer 捞回来时 SETNX 能再次成功，重试就能正常工作。
代价是：如果业务处理已经产生了部分副作用，但最终返回了 error，我们删除去重 key 后会允许重试。重试可能再次执行这些副作用。所以真正生产环境不能只靠这一层 SETNX，业务侧最好还有自己的幂等约束，比如订单表里的唯一键、状态机、或业务事件处理记录。这里的 Redis 去重层主要解决"同一个 event_id 被重复投递"的问题，不替代业务事务。
TTL 设多长？ 至少要大于消息在 PEL 中可能存活的最长时间。如果你的 minIdle 是 30 秒，消息最多被重试 3 次（30s × 3 = 90s），那 TTL 设 24 小时绰绰有余。
Worker Pool：并行消费
有了消费者、Claimer 和幂等处理器，最后一步是把它们组装成一个 worker pool。
type Pool struct { consumer *stream.Consumer claimer *stream.Claimer handler handler.EventHandler workers int logger *zap.Logger}func (p *Pool) Run(ctx context.Context) { msgCh := make(chan redis.XMessage, p.workers*2) var producerWg sync.WaitGroup // readLoop + claimer var workerWg sync.WaitGroup // 启动超时消息接管器 producerWg.Add(1) gofunc() { defer producerWg.Done() p.claimer.Run(ctx, msgCh) }() // 启动 N 个 worker for i := 0; i < p.workers; i++ { workerWg.Add(1) gofunc(id int) { defer workerWg.Done() p.work(ctx, id, msgCh) }(i) } // 消息读取循环 producerWg.Add(1) gofunc() { defer producerWg.Done() p.readLoop(ctx, msgCh) }() // 等 readLoop 和 Claimer 都退出后再关闭 channel，避免 send on closed channel producerWg.Wait() close(msgCh) workerWg.Wait()}架构很清晰：
┌──────────────┐ │ readLoop │ │ (1 goroutine) │ └──────┬───────┘ │ msgCh (buffered) │ ┌──────────────┼──────────────┐ │ │ │ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ worker 0 │ │ worker 1 │ │ worker 2 │ └─────────┘ └─────────┘ └─────────┘ │ ┌────▼────┐ │ Claimer │ (回收的消息也送进 msgCh) └─────────┘• 一个 readLoop goroutine 负责从 Redis 读消息（先 pending 再 new），写入 msgCh• N 个 worker goroutine 从 msgCh读消息并处理• 一个 Claimer goroutine 定期接管超时消息，也写入 msgCh• 消息通过 buffered channel 分发，天然实现了负载均衡
readLoop 的实现：
func (p *Pool) readLoop(ctx context.Context, msgCh chan<- redis.XMessage) { // Phase 1: 恢复 pending 消息 p.logger.Info("恢复 pending 消息...") for { if ctx.Err() != nil { return } msgs, err := p.consumer.ReadPending(ctx) if err != nil { p.logger.Error("读取 pending 失败", zap.Error(err)) return } if len(msgs) == 0 { break } for _, msg := range msgs { msgCh <- msg } } // Phase 2: 读取新消息 p.logger.Info("开始消费新消息...") for { if ctx.Err() != nil { return } msgs, err := p.consumer.ReadNew(ctx) if err != nil { if ctx.Err() != nil { return } p.logger.Error("XREADGROUP 失败", zap.Error(err)) continue } for _, msg := range msgs { msgCh <- msg } }}Worker 的实现：
func (p *Pool) work(ctx context.Context, id int, msgCh <-chan redis.XMessage) { for msg := range msgCh { if ctx.Err() != nil { return } err := p.handler.Handle(ctx, msg) if err != nil { p.logger.Error("处理失败，消息将留在 PEL 中等待回收", zap.String("msg_id", msg.ID), zap.Error(err), ) continue // 不 ACK } if err := p.consumer.Ack(ctx, msg.ID); err != nil { p.logger.Error("ACK 失败", zap.String("msg_id", msg.ID), zap.Error(err)) } }}处理失败时不 ACK，消息留在 PEL 中。 Claimer 会在 30 秒后把它捞回来重新投递。这是 Streams 给我们的"免费重试"——不需要自己实现重试队列、指数退避之类的东西。
启动入口：组装所有组件
func main() { cfg := config.Load() logger, _ := zap.NewDevelopment() rdb := redis.NewClient(&redis.Options{Addr: cfg.RedisAddr, Password: cfg.RedisPass}) // ... 连接检查 ... // 确保消费者组存在 stream.EnsureGroup(ctx, rdb, cfg.StreamKey, cfg.GroupName, "0") // 生成唯一 consumer name hostname, _ := os.Hostname() consumerName := fmt.Sprintf("%s-%s", hostname, uuid.New().String()[:8]) // 组装处理链：OrderHandler → IdempotentHandler orderHandler := handler.NewOrderHandler(logger) idempotentHandler := handler.NewIdempotentHandler(rdb, orderHandler, 24*time.Hour, logger) // 创建消费者和超时接管器 consumer := stream.NewConsumer(rdb, cfg.StreamKey, cfg.GroupName, consumerName, ...) claimer := stream.NewClaimer(rdb, cfg.StreamKey, cfg.GroupName, consumerName, ...) pool := worker.NewPool(consumer, claimer, idempotentHandler, cfg.WorkerCount, logger) // 优雅关闭 runCtx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel() pool.Run(runCtx)}每个 consumer 进程有唯一的 consumerName。 同一个消费者组内，不同的 consumerName 竞争消费。如果你要水平扩展，多起几个进程就行——它们自动分担消息。
跑起来看效果
终端 1：启动生产者
cd redis-streams-consumergo run cmd/producer/main.go# 发布 20 条事件终端 2：启动消费者
go run cmd/consumer/main.go你会看到类似这样的输出：
INFO 消费者启动 {"stream": "events", "group": "worker-group", "consumer": "MacBook-a1b2c3d4", "workers": 3}INFO 恢复 pending 消息...INFO 开始消费新消息...INFO 事件处理完成 {"worker": 0, "msg_id": "1747382400123-0", "type": "order.created", "order_id": "ORD-0001"}INFO 事件处理完成 {"worker": 1, "msg_id": "1747382400124-0", "type": "order.paid", "order_id": "ORD-0002"}WARN 处理失败（模拟） {"msg_id": "1747382400125-0", "type": "order.shipped", "order_id": "ORD-0003"}INFO 事件处理完成 {"worker": 2, "msg_id": "1747382400126-0", "type": "order.created", "order_id": "ORD-0004"}...INFO 回收到超时消息 {"count": 1}INFO 事件处理完成 {"worker": 0, "msg_id": "1747382400125-0", "type": "order.shipped", "order_id": "ORD-0003"}注意第 3 行 ORD-0003 处理失败了。30 秒后 Claimer 把它捞回来（"回收到超时消息"），worker 0 重新处理成功了。整个重试流程是自动的，不需要写任何重试代码。
如果你再跑一次 go run cmd/consumer/main.go（同一个消费者组），会发现 20 条消息已经全部被消费了，新消费者只会等待新消息。这就是消费者组的作用——进度是组级别的，不是进程级别的。
和 asynq 有什么区别
如果你用过 asynq[2]（go-skeleton 里用了它），可能会问：它们都基于 Redis 做异步任务，有什么区别？
ProcessIn） | ||
核心区别在于心智模型不同。
asynq 是"任务队列"——你提交一个"任务"，有一个"worker"处理它，处理完了任务就没了。它的重试、延迟、定时功能都是围绕"任务"这个概念设计的。
Streams 是"事件日志"——事件发生了就写入 Stream，多个消费者组可以独立消费同一份事件。事件处理完了还留在 Stream 里，可以回溯。它更接近 Kafka 的模型。
如果你的场景是"发一个任务，让 worker 处理"——用 asynq。 比如发邮件、生成报表、图片处理。asynq 的重试策略、延迟队列、Web 监控都比自己基于 Streams 做要成熟。
如果你的场景是"事件驱动"——用 Streams。 比如订单状态变更、用户行为追踪、日志聚合。事件可能被多个下游系统消费（订单系统、积分系统、推荐系统），每个系统有自己的消费者组。
实际项目中两者可以共存：业务事件走 Streams，后台任务走 asynq。
生产环境还需要注意的事
消息积压监控
XINFO GROUPS events关注 pending 字段——PEL 中的消息数量。如果持续增长，说明消费者处理速度跟不上生产速度，或者有大量消息处理失败。
Stream 内存管理
Stream 会无限增长。除了 XADD 时的 MAXLEN 裁剪，还可以定期用 XTRIM 主动清理：
XTRIM events MAXLEN ~ 10000如果你需要精确的时间窗口裁剪（比如只保留 7 天的数据），可以用 MINID：
XTRIM events MINID ~ 1746777600000-0 # 删除 7 天前的消息消费者数量 vs 分区
一个 Stream 只有一个"分区"（不像 Kafka 可以多 partition）。如果你有 10 个消费者但每秒只有 10 条消息，大部分消费者会空转。消费者数量应该匹配消息量——通常 3-5 个消费者进程就能处理很大的吞吐。
如果真的需要更高的并行度，可以创建多个 Stream（按业务维度分片），每个 Stream 配一个消费者组。这相当于手动做了 Kafka 的 partition。
什么时候该换 Kafka
如果你发现：
• 消息量超过单台 Redis 的内存容量 • 需要跨数据中心复制 • 需要精确一次（exactly-once）语义 • 需要消息保留 7 天以上
那就该考虑 Kafka 了。Streams 的定位是"轻量级消息队列"，不是要替代 Kafka。
总结
从 XADD 到一个完整的事件消费器，做了这些事：
• Producer 用 XADD写入事件，强制携带event_id保证下游幂等• Consumer 用 XREADGROUP读消息，启动时先恢复 pending（ID=0）再读新消息（ID=>）• ACK 策略 是"处理成功后 ACK"，失败的消息留在 PEL 中 • Claimer 用 XAUTOCLAIM定期接管闲置超过 30 秒的超时消息（注意这不是完整的死信队列——真正的 DLQ 还需要最大重试次数、失败归档、人工处理队列）• 幂等层 用 SETNX+event_id做去重，防止重复处理• Worker Pool 用 buffered channel 分发消息，多 goroutine 并行处理 • 和 asynq 的区别：Streams 是事件日志（多消费者组、可回溯），asynq 是任务队列（重试、延迟、定时）
代码在 redis-streams-consumer，clone 下来就能跑。
引用链接
[1] redis-streams-consumer: https://github.com/arixbit/redis-streams-consumer[2] asynq: https://github.com/hibiken/asynq
