中间件

Redis

Redis 消息传递：PubSub、Stream 与 ...

Redis 消息传递：PubSub、Stream 与 List 消息队列

适用 Redis 版本: 2.0+（PubSub）/ 5.0+（Stream）/ 1.0+（List） 最后更新: 2026-07-11

1. 消息队列概述

1.1 三个核心角色

在任何消息传递系统中，都离不开三个基本角色：

角色	说明	Redis 中的体现
消息队列（Message Queue）	存储消息的中间件，负责暂存生产者发送的消息直到被消费	Redis 数据结构（List / Stream），或内存通道（PubSub）
生产者（Producer）	向消息队列发送消息的客户端	`LPUSH`、`PUBLISH`、`XADD`
消费者（Consumer）	从消息队列接收并处理消息的客户端	`BRPOP`、`SUBSCRIBE`、`XREADGROUP`

生产者 ──发送消息──▶ [消息队列] ──投递消息──▶ 消费者
         (Producer)     (Queue)      (Consumer)

1.2 Redis 中的三种实现方案

Redis 提供了三种不同范式来实现消息传递，各自面向不同的场景：

方案	数据结构	引入版本	核心特征
List	链表（quicklist）	1.0	简单的 FIFO 队列，阻塞读取，无 ACK
PubSub	频道订阅表	2.0	发布/订阅，即发即弃，多消费者广播
Stream	追加日志（Rax 基数树）	5.0	持久化消息队列，消费者组，消息回溯，ACK 确认

1.3 选型速查表

场景	推荐方案	原因
简单任务队列（发邮件、图片处理）	List	实现简单，阻塞读取天然支持"按需消费"
实时通知推送（WebSocket 广播）	PubSub	即发即弃，广播模式，低延迟
聊天室消息分发	PubSub	多用户实时接收同一消息
配置热更新广播	PubSub	所有服务节点同时收到配置变更通知
可靠消息队列（订单处理、流水日志）	Stream	持久化、ACK 确认、消费者组、消息回溯
事件溯源（Event Sourcing）	Stream	消息持久化且可按时间范围回溯，天然适合
需要多个消费者独立消费同一消息	PubSub（广播）或 Stream（多消费者组）	List 只支持单消费者
需要消息确认和重试机制	Stream	List 和 PubSub 均无 ACK
消息需要历史回溯	Stream	List 消费后即删除，PubSub 不持久化

2. List 实现消息队列

2.1 原理回顾

Redis List 底层使用 quicklist（快速链表），天然支持在两端操作。利用这一特性，可以轻松实现一个简单的 FIFO（先进先出）消息队列：

生产者                     Redis List                    消费者
         LPUSH (头部插入)                  BRPOP (尾部阻塞弹出)
  ──────────────────────▶  [msg3, msg2, msg1]  ◀────────────────────
                           head ←──────→ tail

生产者使用 LPUSH 从左侧（头部）插入消息，消费者使用 BRPOP / BLPOP 从右侧（尾部）阻塞弹出消息，实现先进先出。

核心命令回顾：

# 生产者：向队列尾部添加消息
LPUSH queue_name "message1"
LPUSH queue_name "message2"

# 消费者：从队列头部阻塞取出消息（timeout=0 表示永久阻塞）
BRPOP queue_name 5          # 最多阻塞 5 秒
BRPOP queue_name 0          # 永久阻塞，直到有消息

# 查看队列中剩余消息数
LLEN queue_name

2.2 💡 优点

数据安全：List 数据受 Redis 持久化（RDB/AOF）保护，Redis 重启后消息不丢失（取决于持久化配置）。
消息有序：FIFO 严格有序，先 LPUSH 的消息先被 RPOP 取出。
实现简单：仅需两个命令（LPUSH + BRPOP），无需额外配置，所有 Redis 版本均支持。
阻塞模式天然省资源：BRPOP 阻塞等待而不是轮询，消费者空闲时不消耗 CPU。

2.3 🚨 缺点

不支持消息确认（ACK）：BRPOP 取出消息后立即从 List 中删除。如果消费者处理过程中宕机，消息将永久丢失，无法重试。
只支持单消费者（竞争消费）：多个消费者 BRPOP 同一个 List 时是竞争关系，一条消息只会被一个消费者取走。无法实现"一条消息被多个消费者独立处理"的广播模式。
不支持消息回溯：消息被取出即删除，无法像 Stream 一样按时间范围查阅历史消息。
没有消费者组概念：无法区分不同消费者组，无法实现组内负载均衡 + 组间广播。
无法重试失败消息：没有 pending 状态，处理失败的消息无法被重新投递。

2.4 适用场景

简单的异步任务队列（如发邮件、生成报表、图片缩放）
不需要 ACK 确认的场景（允许少量消息丢失）
任务处理足够快、消费者故障率低的环境
快速原型和 MVP 阶段的临时方案

2.5 完整示例：简单任务队列

以下是一个完整的生产者-消费者示例（使用 Redis CLI 模拟）：

生产者：持续向队列添加任务

# 生产者循环：每隔 2 秒产生一个任务
while true; do
  LPUSH task_queue "task_$(date +%s)"
  sleep 2
done

消费者：阻塞等待并处理任务

# 消费者：阻塞等待任务
while true; do
  BRPOP task_queue 0   # 永久阻塞，直到有消息
  # 实际应用中，这里连接代码处理消息内容
done

Go 语言示例：

// 生产者
func producer(rdb *redis.Client, queue string) {
    for i := 0; ; i++ {
        msg := fmt.Sprintf("task_%d_%d", i, time.Now().Unix())
        rdb.LPush(ctx, queue, msg)
        time.Sleep(2 * time.Second)
    }
}

// 消费者
func consumer(rdb *redis.Client, queue string) {
    for {
        result, err := rdb.BRPop(ctx, 0, queue).Result()
        if err != nil {
            log.Printf("brpop error: %v", err)
            continue
        }
        // result[0] = key, result[1] = value
        msg := result[1]
        processTask(msg) // 处理消息的代码
    }
}

🚨 陷阱提醒：如果消费者在 processTask() 阶段崩溃，消息已经通过 BRPOP 从队列中删除了，无法恢复。对于需要可靠性的场景，请使用 Stream。

3. PubSub（发布订阅）

3.1 模型介绍

Redis PubSub 是 Redis 2.0 引入的消息传递模型，实现了经典的发布/订阅模式。消息即发即弃（fire-and-forget），不进行持久化。

核心概念：

概念	说明
Channel（频道）	消息的逻辑通道，发布者向指定频道发送消息，订阅者从指定频道接收消息
Publisher（发布者）	通过 `PUBLISH` 向频道发送消息的客户端
Subscriber（订阅者）	通过 `SUBSCRIBE` 或 `PSUBSCRIBE` 接收频道消息的客户端
Pattern（模式）	支持通配符的频道匹配表达式，用于一次订阅多个频道

                  PUBLISH channel1 "hello"
Publisher-1 ──────────────────────────────────┐
                                              ▼
Publisher-2 ────▶  Channel-1  ────▶  Subscriber-A
                                   ───▶  Subscriber-B
                                   
Publisher-3 ────▶  Channel-2  ────▶  Subscriber-C
                                   ───▶  Subscriber-D (via PSUBSCRIBE news.*)

3.2 命令详解

SUBSCRIBE — 订阅频道

订阅一个或多个频道，进入订阅模式。进入订阅模式后，客户端只能执行订阅/退订相关命令（SUBSCRIBE、UNSUBSCRIBE、PSUBSCRIBE、PUNSUBSCRIBE、PING、RESET、QUIT）。

SUBSCRIBE channel [channel ...]

示例：

# 终端 1：订阅者
127.0.0.1:6379> SUBSCRIBE news music
1) "subscribe"
2) "news"
3) (integer) 1
1) "subscribe"
2) "music"
3) (integer) 2

# 之后收到消息时的输出格式：
1) "message"       # 消息类型（message/pmessage）
2) "news"          # 频道名
3) "breaking: something happened"  # 消息内容

🚨 重要：SUBSCRIBE 不支持通配符。需要模式匹配请使用 PSUBSCRIBE。

UNSUBSCRIBE — 退订频道

UNSUBSCRIBE [channel [channel ...]]

不带参数则退订所有频道。每个退订成功的频道会返回一条确认消息：

127.0.0.1:6379> UNSUBSCRIBE news
1) "unsubscribe"
2) "news"
3) (integer) 0

PSUBSCRIBE — 模式订阅

使用通配符模式订阅频道，支持 *、?、[] 三种通配符：

PSUBSCRIBE pattern [pattern ...]

通配符	含义	示例
`*`	匹配任意多个字符	`news.*` 匹配 `news.sports`、`news.tech`
`?`	匹配单个字符	`news.?` 匹配 `news.1`，不匹配 `news.12`
`[...]`	匹配方括号中的任意一个字符	`news.[st]ports` 匹配 `news.sports`、`news.tports`

示例：

# 订阅所有以 "news." 开头的频道
127.0.0.1:6379> PSUBSCRIBE news.*
1) "psubscribe"
2) "news.*"
3) (integer) 1

# 通过模式订阅接收到的消息格式不同（pmessage）：
1) "pmessage"      # 消息类型（带模式前缀 p）
2) "news.*"        # 匹配的模式
3) "news.sports"   # 实际频道名
4) "goal!"         # 消息内容

🔬 注意：同时使用 SUBSCRIBE 和 PSUBSCRIBE 时，如果一条消息同时被具体频道订阅和模式匹配，客户端会收到两条消息（一条 message，一条 pmessage）。

PUNSUBSCRIBE — 退订模式

PUNSUBSCRIBE [pattern [pattern ...]]

不带参数则退订所有模式订阅。

PUBLISH — 发布消息

PUBLISH channel message

返回值：收到消息的订阅者数量（整数）。

示例：

# 发布消息到频道 news
127.0.0.1:6379> PUBLISH news "breaking news: redis 8.0 released"
(integer) 3    # 有 3 个订阅者（包括模式订阅者）收到了消息

# 发布到无人订阅的频道
127.0.0.1:6379> PUBLISH nobody_here "hello?"
(integer) 0    # 0 个订阅者，消息直接丢弃

PUBSUB — 频道管理

用于查看 PubSub 系统的内部状态。

# 查看活跃频道（至少有一个订阅者的频道）
PUBSUB CHANNELS [pattern]

127.0.0.1:6379> PUBSUB CHANNELS
1) "news"
2) "music"

127.0.0.1:6379> PUBSUB CHANNELS n*
1) "news"

# 查看指定频道的订阅者数量
PUBSUB NUMSUB [channel [channel ...]]

127.0.0.1:6379> PUBSUB NUMSUB news music chat
1) "news"
2) (integer) 5    # news 有 5 个订阅者
3) "music"
4) (integer) 2    # music 有 2 个订阅者
5) "chat"
6) (integer) 0    # chat 没有订阅者

# 查看模式订阅数量
PUBSUB NUMPAT

127.0.0.1:6379> PUBSUB NUMPAT
(integer) 3       # 当前有 3 个活跃的模式订阅

3.3 🔬 深入原理

订阅关系存储

Redis 服务端使用两种数据结构来维护订阅关系：

pubsub_channels (字典 / hash table)
┌──────────────────┬──────────────────────────┐
│  channel name    │  client list (链表)       │
├──────────────────┼──────────────────────────┤
│  "news"          │  [client-A, client-B]    │
│  "music"         │  [client-C]              │
│  "sports"        │  [client-D, client-E]    │
└──────────────────┴──────────────────────────┘

pubsub_patterns (链表 / linked list)
┌──────────────┬──────────────────────┐
│  pattern     │  client              │
├──────────────┼──────────────────────┤
│  "news.*"    │  client-F            │
│  "*.log"     │  client-G            │
│  "sports.?"  │  client-H            │
└──────────────┴──────────────────────┘

pubsub_channels 是一个字典（hash table），key 是频道名，value 是订阅该频道的客户端链表。精确订阅（SUBSCRIBE）通过这个结构实现 O(1) 查找。
pubsub_patterns 是一个链表，每个节点存储 (pattern, client)。模式订阅（PSUBSCRIBE）需要遍历整个链表来匹配，复杂度 O(N)，其中 N 是模式数量。

消息投递流程

当执行 PUBLISH channel message 时：

1. 在 pubsub_channels 字典中查找 channel → 得到客户端链表
2. 遍历链表，向每个客户端发送消息
3. 遍历 pubsub_patterns 链表，对每个模式进行检查：
   - 如果当前 channel 匹配该模式 → 向该客户端发送消息（pmessage 格式）
4. 返回收到消息的客户端总数 = step2 客户端数 + step3 匹配的客户端数

⚡ 性能提示：模式订阅（PSUBSCRIBE）越多，PUBLISH 时的遍历开销越大。如果频道数量巨大，优先使用精确频道名，避免过度依赖通配符模式。

🔬 Sharded PubSub（Redis 7.0+）

Redis 7.0 引入了分片发布订阅（Sharded PubSub），解决集群模式下传统 PubSub 的广播风暴问题。

传统 PubSub 在集群中的问题：

PUBLISH channel1 "hello"
       │
       ▼
  [Node 1]────广播到───▶ [Node 2]
     │                       │
     ├─▶ Client-A (订阅 channel1)    ├─▶ Client-B (订阅 channel1)
     │                       │
     └─▶ Client-C (订阅 channel1)    (C 也收到了，即使它不该收)

在传统 PubSub 中，集群每个节点向所有其他节点广播 PUBLISH 消息（无论该节点是否有订阅者），造成跨节点广播风暴。

Sharded PubSub 的改进：

# 订阅：使用 SSUBSCRIBE（替代 SUBSCRIBE）
SSUBSCRIBE channel [channel ...]

# 退订：使用 SUNSUBSCRIBE（替代 UNSUBSCRIBE）
SUNSUBSCRIBE [channel [channel ...]]

# 发布：使用 SPUBLISH（替代 PUBLISH）
SPUBLISH channel message

SPUBLISH 将 channel 按 Key 分片规则哈希到特定 slot，消息只发送到负责该 slot 的节点。
不再进行全集群广播，显著减少网络开销。

# 传统 PubSub（集群全广播）
127.0.0.1:6379> PUBLISH order.created "order_12345"
(integer) 5

# Sharded PubSub（按 slot 分发，不跨节点广播）
127.0.0.1:6379> SPUBLISH order.created "order_12345"
(integer) 2  # 只有负责该 slot 的节点上的订阅者收到

传统 vs 分片 PubSub 对比：

特性	传统 PubSub	Sharded PubSub (7.0+)
消息分发范围	全集群广播	按 slot 分发到特定节点
频道名限制	无	必须与 key 遵循相同的 slot 规则（放到同一 slot）
订阅命令	`SUBSCRIBE` / `PSUBSCRIBE`	`SSUBSCRIBE`
发布命令	`PUBLISH`	`SPUBLISH`
模式订阅	支持	不支持
适用场景	通知广播、配置更新	按租户/用户分片的消息推送

💡 最佳实践：在集群模式下，如果只需要局部消息分发，应优先使用 SSUBSCRIBE / SPUBLISH。

3.4 🚨 局限性

局限	详情
消息不持久化	消息只在内存中流转，Redis 重启后所有消息和订阅关系丢失
没有 ACK 确认	投递即"成功"，无论消费者是否成功处理，也无法重试
离线期间消息无补收	消费者断线期间发布的消息永远丢失，重新连接后只接收新消息
无堆积能力	没有消费者时，发布的消息直接丢弃，`PUBLISH` 返回 0（不像 Stream 会保留）
无消费进度记录	无法知道消费者"读到了哪一条"，无法做 offset 管理
subscribe 后独占连接	客户端进入订阅模式后不能再执行普通命令（只能新开连接）

3.5 适用场景

实时通知：WebSocket 消息转发——用户 A 操作后，后端通过 PubSub 通知 WebSocket 网关推送给用户 B。
配置热更新广播：配置中心更新后，通过 PubSub 通知所有服务实例重新加载配置。
聊天室消息分发：聊天消息需要实时广播给房间内所有在线用户。
服务发现/健康状态变更通知：节点上下线时即时广播。
分布式缓存失效通知：缓存更新时通知所有节点清除本地缓存。

💡 最佳实践：在设计时，默认假设 PubSub 消息是"可能丢失的"。永远不要用 PubSub 传递需要可靠投递的业务数据（如订单、支付流水）。用 Stream 或专业 MQ 兜底。

4. Stream（Redis 5.0+ 消息队列）

4.1 模型介绍

Redis Stream 是 Redis 5.0 引入的功能完善的持久化消息队列。它不是你理解的"另一个 PubSub"，而是对标 Kafka 的、支持追加日志（append-only log）模式的可靠消息系统。

核心概念：

Stream: mystream
┌──────────────────────────────────────────────────────┐
│  Entry ID        │  Field-Value Pairs                 │
├──────────────────────────────────────────────────────┤
│  1699999999999-0 │  {"user": "alice", "msg": "hi"}   │  ← 最旧
│  1699999999999-1 │  {"user": "bob",   "msg": "hello"}│
│  1700000000000-0 │  {"user": "carol", "msg": "hey"}  │
│  1700000000001-0 │  {"user": "dave",  "msg": "yo"}   │  ← 最新
└──────────────────────────────────────────────────────┘

概念	说明
Stream	一个消息流的名称（对应一个 key），可以存在多个
Entry（消息）	Stream 中的一条记录，由唯一 ID 和多个 field-value 对组成
Consumer（消费者）	从 Stream 读取消息的客户端
Consumer Group（消费者组）	一组消费者共享消费进度，组内竞争消费，组间独立消费
Pending List（待确认列表）	已投递但尚未被 ACK 的消息集合，用于故障恢复

消费模型：

                   mystream
         ┌─────────────────────────┐
         │  msg7  msg6  msg5  ...  │
         └──────────┬──────────────┘
                    │
    ┌───────────────┼───────────────┐
    ▼                               ▼
 Group-A (消费者组 A)          Group-B (消费者组 B)
    │                               │
    ├─ Consumer-A1 (竞争消费)       ├─ Consumer-B1
    └─ Consumer-A2                 └─ Consumer-B2

组内竞争：同一组内的消费者竞争消费，一条消息只会被组内一个消费者处理。
组间独立：不同消费者组各自维护独立的消费进度（last_delivered_id），互不影响。

4.2 消息 ID

Stream 中每条消息都有一个全局唯一的 ID，格式为：

<millisecondsTime>-<sequenceNumber>

格式部分	说明	示例
`millisecondsTime`	Unix 毫秒时间戳	`1699999999999`
`sequenceNumber`	同一毫秒内的序号（从 0 开始）	`0`、`1`、`2`

ID 生成方式：

# 自动生成（推荐）：使用 *
XADD mystream * field value
# 返回 "1699999999999-0"

# 手动指定 ID（必须大于上一条消息的 ID）
XADD mystream 1700000000000-0 field value
# 返回 "1700000000000-0"

ID 比较规则：

比较方法：先比较毫秒时间戳，再比较序号。

1699999999999-0  <  1699999999999-1    (同一毫秒，序号递增)
1699999999999-5  <  1700000000000-0    (不同毫秒，时间戳递增)

🚨 陷阱：手动指定 ID 时，如果 ID 不大于已有消息的 ID，XADD 会返回错误： (error) ERR The ID specified in XADD is equal or smaller than the target stream top item 在生产代码中几乎总是使用 * 自动生成。

4.3 基础命令

XADD — 追加消息

XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT limit]] *|ID field value [field value ...]

参数	说明
`NOMKSTREAM`	如果 key 不存在，不创建 Stream（默认会自动创建）
`MAXLEN ~ threshold`	近似裁剪（Approximate trimming），保留大约 threshold 条最新消息
`MAXLEN = threshold`	精确裁剪（Exact trimming），严格保留 threshold 条最新消息
`MINID ~ threshold`	近似裁剪，保留 ID 大于 threshold 的消息
`MINID = threshold`	精确裁剪，保留 ID 大于 threshold 的消息
`LIMIT limit`	配合 `~` 使用，限制一次裁剪移除的最大条数
`*`	自动生成消息 ID（推荐）

示例：

# 基础用法：添加一条消息
XADD mystream * name alice message "hello world"
# 返回: "1699999999999-0"

# 限制 Stream 长度：保留最新 1000 条
XADD mystream MAXLEN ~ 1000 * name bob message "hi"
# 返回: "1699999999999-1"

# 精确裁剪：严格保留 1000 条
XADD mystream MAXLEN = 1000 * name carol message "hey"
# 返回: "1699999999999-2"

# 配合 LIMIT：每次最多删除约 100 条旧消息（性能优化）
XADD mystream MAXLEN ~ 1000 LIMIT 100 * name dave message "yo"
# 返回: "1699999999999-3"

# NOMKSTREAM：如果 stream 不存在则报错，不自动创建
XADD noexist NOMKSTREAM * test value
# 返回: (nil)  ← 不创建，也不添加

⚡ 性能提示：~ 近似裁剪比 = 精确裁剪更高效。精确裁剪必须删除恰好数量的消息，可能触发内存重分配；近似裁剪允许少量误差（约几十条），性能显著更好。生产环境始终使用 ~。

XREAD — 阻塞/非阻塞读取

XREAD [COUNT count] [BLOCK ms] STREAMS key [key ...] ID [ID ...]

参数	说明
`COUNT count`	最多返回的消息条数
`BLOCK ms`	阻塞等待时间（毫秒），`0` 表示永久阻塞
`STREAMS key ...`	要读取的 Stream 列表
`ID ...`	每个 Stream 对应的起始读取 ID

ID 取值说明：

ID 值	含义
`0` 或 `0-0`	从 Stream 的第一条消息开始读取
`$`	只读取新到达的消息（跳过已有历史消息）
具体 ID（如 `1699999999999-0`）	从该 ID 之后（不含该 ID）开始读取

示例：

# 前置：添加一些消息
XADD mystream * msg "first"
XADD mystream * msg "second"
XADD mystream * msg "third"

# 从头读取所有消息
XREAD COUNT 10 STREAMS mystream 0
# 返回:
# 1) 1) "mystream"
#    2) 1) 1) "1699999999999-0"
#          2) 1) "msg"
#             2) "first"
#       2) 1) "1699999999999-1"
#          2) 1) "msg"
#             2) "second"
#       3) 1) "1699999999999-2"
#          2) 1) "msg"
#             2) "third"

# 阻塞读取新消息（超时 5 秒）
XREAD BLOCK 5000 STREAMS mystream $
# 如果 5 秒内没有新消息，返回 (nil)

# 阻塞读取新消息（永久阻塞，直到有新消息）
XREAD BLOCK 0 STREAMS mystream $

# 同时读取多个 Stream
XREAD BLOCK 5000 STREAMS stream1 stream2 $ $
# 返回每个 Stream 各自的新消息

XRANGE / XREVRANGE — 范围查询

# 正向范围查询（按 ID 从小到大）
XRANGE key start end [COUNT count]

# 反向范围查询（按 ID 从大到小）
XREVRANGE key end start [COUNT count]

特殊值	含义
`-`	最小可能的 ID（`0-0`）
`+`	最大可能的 ID（`18446744073709551615-18446744073709551615`）

示例：

# 查询所有消息
XRANGE mystream - +
# 返回所有消息，按 ID 升序排列

# 查询前 5 条消息
XRANGE mystream - + COUNT 5

# 查询指定时间戳之后的消息
XRANGE mystream 1700000000000-0 + COUNT 100

# 查询两条消息之间的消息（含边界）
XRANGE mystream 1699999999999-0 1699999999999-5

# 反向查询：最新的 3 条消息
XREVRANGE mystream + - COUNT 3

💡 最佳实践：XRANGE / XREVRANGE 不受消费者组影响，是独立的只读查询。适合排查问题：查看消息内容、搜索特定时间范围的消息。

XLEN — Stream 长度

XLEN key

127.0.0.1:6379> XLEN mystream
(integer) 1042

XDEL — 删除消息

从 Stream 中删除指定 ID 的消息（注意：仅标记删除，不释放内存）。

XDEL key ID [ID ...]

# 删除一条消息
127.0.0.1:6379> XDEL mystream 1699999999999-0
(integer) 1    # 成功删除 1 条

# 删除多条消息
127.0.0.1:6379> XDEL mystream 1699999999999-1 1699999999999-2
(integer) 2    # 成功删除 2 条

🔬 深入原理：XDEL 只是将消息标记为删除，不会真正释放内存。底层 Rax 树中的空间不会归还给操作系统。真正释放内存需要通过 XTRIM 裁剪 Stream 或定期重启 Redis。

XTRIM — 裁剪 Stream

XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]

功能与 XADD 中的 MAXLEN/MINID 完全相同，但作为独立命令使用，不需要同时添加消息。

# 保留最新 1000 条（近似）
XTRIM mystream MAXLEN ~ 1000
# 返回: (integer) 42  ← 删除了约 42 条旧消息

# 精确保留 1000 条
XTRIM mystream MAXLEN = 1000

# 保留 ID 大于指定值的消息
XTRIM mystream MINID ~ 1700000000000-0

# 配合 LIMIT，每次最多删除 100 条（避免阻塞）
XTRIM mystream MAXLEN ~ 1000 LIMIT 100

⚡ 性能提示：在 XADD 中使用 MAXLEN ~ 比单独调用 XTRIM 更高效——因为在追加消息时顺带裁剪，减少了额外的遍历开销。

4.4 消费者组

消费者组是 Stream 区别于 List 和 PubSub 的最核心特性，提供了可靠消费所需的关键能力。

核心概念

                   mystream
         ┌─────────────────────────┐
         │  msg7  msg6  msg5  ...  │
         └──────────┬──────────────┘
                    │
         ┌─────────┴─────────┐
         ▼                   ▼
    Group: "processors"  Group: "auditors"
    ┌─────────────────┐  ┌─────────────────┐
    │ last_delivered: │  │ last_delivered: │
    │ msg5             │  │ msg3             │  ← 独立进度
    │                  │  │                  │
    │ Consumers:       │  │ Consumers:       │
    │  C1 ── pending ──│  │  C3 ── pending ──│
    │       [msg5]     │  │       [msg3]     │
    │  C2              │  │                  │
    └─────────────────┘  └─────────────────┘

last_delivered_id：消费者组级别的"读游标"，记录最后一条投递给本组的消息 ID。
pending list：已投递但尚未 XACK 的消息列表。每条 pending 消息记录了：ID、所有者（消费者名）、空闲时间等。
组内竞争：同一组内，一条消息只会投递给一个消费者。

XGROUP CREATE — 创建消费者组

XGROUP CREATE key groupname id|$ [MKSTREAM] [ENTRIESREAD entries-read]

参数	说明
`id`	起始 ID：`0` 表示从头消费历史所有消息，`$` 表示只消费新消息
`MKSTREAM`	如果 Stream 不存在，自动创建一个空的 Stream
`ENTRIESREAD`	(Redis 7.0+) 将指定数量的历史消息标记为"已读"（跳过它们）

# 创建消费者组，从 Stream 开头开始消费（消费所有历史消息）
XGROUP CREATE mystream mygroup 0
# 返回: OK

# 创建消费者组，只消费新消息（跳过历史）
XGROUP CREATE mystream mygroup $
# 返回: OK

# 带 MKSTREAM：如果 mystream 不存在，创建空 Stream
XGROUP CREATE mystream mygroup $ MKSTREAM
# 返回: OK

💡 最佳实践：首次创建消费者组时，大多数场景使用 $（从最新开始），历史消息通常不需要重新消费。如需重放历史消息（如数据修复），可以选择 0。

XGROUP DESTROY / CREATECONSUMER / DELCONSUMER / SETID

# 销毁消费者组
XGROUP DESTROY key groupname
# 返回: (integer) 1  ← 成功销毁

# 显式创建消费者（通常不需要，XREADGROUP 会自动创建）
XGROUP CREATECONSUMER key groupname consumername
# 返回: (integer) 1  ← 成功创建

# 删除消费者（该消费者的 pending 消息会重新投递给组内其他消费者）
XGROUP DELCONSUMER key groupname consumername
# 返回: (integer) 42  ← 该消费者有 42 条 pending 消息，已释放

# 设置最后投递 ID（可用于跳过消息或重置消费位置）
XGROUP SETID key groupname id|$
# 返回: OK

# 示例：将消费者组重置到 Stream 开头（重放所有消息）
XGROUP SETID mystream mygroup 0

# 示例：跳转到新消息（跳过历史）
XGROUP SETID mystream mygroup $

XREADGROUP — 消费者组消费

XREADGROUP GROUP group consumer [COUNT count] [BLOCK ms] [NOACK] STREAMS key [key ...] ID [ID ...]

参数	说明
`GROUP group consumer`	消费者组名和消费者名
`COUNT count`	最多返回几条消息
`BLOCK ms`	阻塞等待时间
`NOACK`	不需要确认（不推荐生产环境使用）
`ID`	关键：`>` 读新消息，具体 ID 读 pending 消息

ID 的特殊含义在 XREADGROUP 中：

ID 值	含义
`>`	从未投递给本组的消息（新消息）——这是最常见的用法
`0` 或其他具体 ID	读取本消费者的 pending 消息（已投递但未 ACK）

完整消费流程示例：

# ===== 初始化 =====
# 1. 创建消费者组
XGROUP CREATE mystream mygroup $ MKSTREAM
# 返回: OK

# 2. 生产者：添加消息
XADD mystream * msg "task_1"
XADD mystream * msg "task_2"
XADD mystream * msg "task_3"

# ===== 消费 =====
# 3. 消费者读取新消息（> 表示从未投递过的消息）
XREADGROUP GROUP mygroup consumer1 COUNT 2 STREAMS mystream >
# 返回:
# 1) 1) "mystream"
#    2) 1) 1) "1699999999999-0"
#          2) 1) "msg"
#             2) "task_1"
#       2) 1) "1699999999999-1"
#          2) 1) "msg"
#             2) "task_2"

# 4. 消费成功后，确认消息
XACK mystream mygroup 1699999999999-0 1699999999999-1
# 返回: (integer) 2

# 5. 再次读取新消息
XREADGROUP GROUP mygroup consumer1 STREAMS mystream >
# 返回:
# 1) 1) "mystream"
#    2) 1) 1) "1699999999999-2"
#          2) 1) "msg"
#             2) "task_3"

# 6. ACK 确认
XACK mystream mygroup 1699999999999-2
# 返回: (integer) 1

# 7. 没有新消息了
XREADGROUP GROUP mygroup consumer1 STREAMS mystream >
# 返回: (nil)  或阻塞等待（如果用了 BLOCK）

XACK — 消息确认

XACK key group ID [ID ...]

消息被消费者处理后，必须调用 XACK 确认。确认后，消息从该消费者组的 pending 列表中移除。

# 确认单条消息
XACK mystream mygroup 1699999999999-0
# 返回: (integer) 1

# 确认多条消息
XACK mystream mygroup 1699999999999-0 1699999999999-1 1699999999999-2
# 返回: (integer) 3

# 确认一条并非 pending 的消息（已确认或不存在）
XACK mystream mygroup 1699999999999-0
# 返回: (integer) 0    ← 该消息不在 pending 列表中

🚨 陷阱：忘记 XACK 是 Stream 最常见的错误。未确认的消息会永远留在 pending list 中，导致：

Pending list 持续膨胀，占用内存

消费者重启后，pending 消息需要重新处理（可能重复消费）

必须在业务处理成功后立即调用 XACK。

XPENDING — 查看 pending 消息

# 查看 Pending 消息概要
XPENDING key group

# 查看 Pending 消息详情
XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

概要模式（不带范围参数）：

127.0.0.1:6379> XPENDING mystream mygroup
1) (integer) 5              # Pending 消息总数
2) "1699999999999-0"        # 最小 pending 消息 ID
3) "1699999999999-4"        # 最大 pending 消息 ID
4) 1) 1) "consumer1"        # 各消费者的 pending 分布
      2) "3"                # consumer1 有 3 条 pending
   2) 1) "consumer2"
      2) "2"                # consumer2 有 2 条 pending

详情模式（带范围参数）：

# 查看前 10 条 pending 消息详情
XPENDING mystream mygroup - + 10

# 返回每条消息的详细信息：
# 1) 1) "1699999999999-0"            # 消息 ID
#    2) "consumer1"                   # 当前所有者（消费者名）
#    3) (integer) 120000              # 空闲时间（毫秒）
#    4) (integer) 3                   # 已被投递的次数

按 IDLE 时间筛选（发现超时消息）：

# 查看空闲超过 60 秒的 pending 消息
XPENDING mystream mygroup IDLE 60000 - + 10

# 查看特定消费者的超时 pending 消息
XPENDING mystream mygroup IDLE 60000 - + 10 consumer1

💡 最佳实践：定期检查 pending 消息的空闲时间。如果发现大量超过预期处理时间的 pending 消息，说明有消费者处理失败或挂掉了，需要通过 XCLAIM 或 XAUTOCLAIM 进行故障转移。

XCLAIM — 转移消息所有权

XCLAIM key group consumer min-idle-time ID [ID ...]
       [IDLE ms] [TIME ms-unix-time] [RETRYCOUNT count]
       [FORCE] [JUSTID] [LASTID lastid]

当消费者崩溃或处理超时，其 pending 消息需要转让给其他正常消费者。XCLAIM 正是为此设计：

参数	说明
`min-idle-time`	最小空闲时间（毫秒），只有空闲超过此值的消息才被认领
`IDLE ms`	将认领后消息的空闲时间设置为指定值
`TIME ms-unix-time`	设置认领时间的 Unix 毫秒时间戳（通常与 `IDLE` 配合使用）
`RETRYCOUNT count`	设置消息的重试次数
`FORCE`	强制认领，即使原消费者可能仍在线
`JUSTID`	只返回消息 ID，不返回完整内容（减少网络传输）
`LASTID lastid`	配合 `JUSTID` 使用，设置该消费者的最后投递 ID

示例：

# 假设 consumer1 崩溃，其 pending 消息空闲超过 60 秒
# 首先查看超时消息
XPENDING mystream mygroup IDLE 60000 - + 10 consumer1

# 将超过 60 秒空闲的消息转让给 consumer2
XCLAIM mystream mygroup consumer2 60000 1699999999999-0 1699999999999-1
# 返回: 被认领消息的完整内容（格式同 XRANGE）

# 只返回 ID，不返回完整内容
XCLAIM mystream mygroup consumer2 60000 1699999999999-0 JUSTID
# 返回:
# 1) "1699999999999-0"

# 认领并增加重试计数
XCLAIM mystream mygroup consumer2 60000 1699999999999-0 RETRYCOUNT 5

🔬 关键：XCLAIM 后，consumer2 需要通过 XREADGROUP 以具体 ID 读取 pending 消息来获取消息内容，或者直接使用 XCLAIM 返回的消息体进行处理。处理完成后仍需要调用 XACK 确认。

XAUTOCLAIM — 自动认领（Redis 6.2+）

XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]

XCLAIM 的手动流程（先 XPENDING 查超时消息 → 再 XCLAIM 逐一认领）很繁琐。Redis 6.2 引入的 XAUTOCLAIM 自动完成扫描 + 认领：

参数	说明
`min-idle-time`	最小空闲时间
`start`	扫描起始 ID，`0-0` 表示从头扫描
`COUNT count`	每次最多认领的消息数（默认 100）
`JUSTID`	只返回消息 ID

返回值：

1) "next-start-id"          # 下次调用应使用的起始 ID（用于分页）
2) [                          # 被认领的消息数组
     1) 1) "msg-id"
        2) [field-value pairs]
     2) ...
   ]
3) [                          # 已删除但仍在 pending 中的 ID 列表（可忽略）
     "deleted-msg-id",
     ...
   ]

生产级消费者故障恢复完整示例：

# 启动参数：空闲超过 30 秒的消息自动转让给当前消费者
# 每次最多认领 10 条

# 第一次调用（从 0-0 开始扫描）
XAUTOCLAIM mystream mygroup backup_consumer 30000 0-0 COUNT 10
# 返回:
# 1) "1699999999999-5"     ← 下次调用的起始 ID
# 2) 1) 1) "1699999999999-0"     ← 被认领的消息 1
#       2) 1) "msg"
#          2) "task_1"
#    2) 1) "1699999999999-3"     ← 被认领的消息 2
#       2) 1) "msg"
#          2) "task_3"
# 3) (empty array)               ← 没有待清理的已删除 pending ID

# 处理认领到的消息，然后逐一 XACK
XACK mystream mygroup 1699999999999-0 1699999999999-3

# 继续扫描（使用上次返回的 next-start-id）
XAUTOCLAIM mystream mygroup backup_consumer 30000 1699999999999-5 COUNT 10

# 如果返回 "0-0" 表示没有更多 pending 消息需要处理
XAUTOCLAIM mystream mygroup backup_consumer 30000 1699999999999-5 COUNT 10
# 返回: 1) "0-0"    ← return "0-0" 表示扫描完成
#       2) (empty array)
#       3) (empty array)

💡 最佳实践：每个消费者启动时和执行完一轮消费后，都应该运行一次 XAUTOCLAIM 来接管其他崩溃消费者的超时消息。这是生产环境 Stream 消费的标准做法。

XINFO — 查看 Stream/Group/Consumer 状态

# 查看整个 Stream 的信息
XINFO STREAM key

127.0.0.1:6379> XINFO STREAM mystream
 1) "length"
 2) (integer) 1500                # Stream 中消息总数
 3) "radix-tree-keys"
 4) (integer) 1                   # Rax 树中的 key 数量
 5) "radix-tree-nodes"
 6) (integer) 2                   # Rax 树中的节点数量
 7) "last-generated-id"
 8) "1700000000123-5"             # 最后一条消息的 ID（最重要的监控指标）
 9) "max-deleted-entry-id"
10) "1699999999999-99"            # 最后一条被删除的消息 ID
11) "entries-added"
12) (integer) 2000                # 总共添加过的消息数
13) "recorded-first-entry-id"
14) "1699999999500-0"             # Stream 中第一条消息的 ID（裁剪后）
15) "groups"
16) (integer) 2                   # 消费者组数量
17) "first-entry"
18) 1) "1699999999500-0"          # 第一条消息的完整内容
    2) 1) "msg"
       2) "oldest"
19) "last-entry"
20) 1) "1700000000123-5"          # 最后一条消息的完整内容
    2) 1) "msg"
       2) "newest"

# 查看 Stream 的所有消费者组
XINFO GROUPS key

127.0.0.1:6379> XINFO GROUPS mystream
1) 1) "name"
   2) "mygroup"                   # 组名
   3) "consumers"
   4) (integer) 3                 # 消费者数量
   5) "pending"
   6) (integer) 42                # 该组的 pending 消息数
   7) "last-delivered-id"
   8) "1700000000120-0"           # 该组最后投递的消息 ID
   9) "entries-read"
  10) (integer) 1580              # 该组已读取的消息数量
  11) "lag"
  12) (integer) 0                 # 待消费的消息数（Stream 长度 - 已消费）

2) 1) "name"
   2) "another_group"
   3) "consumers"
   4) (integer) 2
   5) "pending"
   6) (integer) 5
   7) "last-delivered-id"
   8) "1700000000100-0"
   9) "entries-read"
  10) (integer) 1500
  11) "lag"
  12) (integer) 0

# 查看消费者组内每个消费者的详情
XINFO CONSUMERS key group

127.0.0.1:6379> XINFO CONSUMERS mystream mygroup
1) 1) "name"
   2) "consumer1"                 # 消费者名称
   3) "pending"
   4) (integer) 15                # 该消费者的 pending 消息数
   5) "idle"                      
   6) (integer) 3000              # 距离最后一次与 Redis 交互的毫秒数
   7) "inactive"
   8) (integer) -1                # Redis 7.2+: 该消费者实例不活跃的毫秒数

2) 1) "name"
   2) "consumer2"
   3) "pending"
   4) (integer) 27
   5) "idle"
   6) (integer) 120000            # 2 分钟未活动，可能已崩溃
   7) "inactive"
   8) (integer) -1

💡 监控要点：

lag > 0：消费速度赶不上生产速度，需要扩容消费者。

idle 过大（远超处理时间）：消费者可能已挂掉，pending 消息需要 XAUTOCLAIM。

pending 持续增长：未及时 XACK，或处理速度过慢。

4.5 消费策略

策略 1：从头消费（ID=0）

# 创建消费者组时指定 ID=0（消费所有历史消息）
XGROUP CREATE mystream mygroup 0 MKSTREAM

# 消费者读取
XREADGROUP GROUP mygroup consumer1 STREAMS mystream >

适用场景：新上线的服务需要处理所有已有数据（如数据迁移、报表生成）。

策略 2：从最新消费（ID=$）

# 创建消费者组时指定 ID=$（跳过历史消息）
XGROUP CREATE mystream mygroup $ MKSTREAM

# 消费者读取
XREADGROUP GROUP mygroup consumer1 BLOCK 5000 STREAMS mystream >

适用场景：常规服务只需要处理新产生的消息。

策略 3：消费 Pending 消息（故障恢复）

# Step 1: 查看有多少 pending 消息
XPENDING mystream mygroup - + 10

# Step 2: 用具体 ID（如 0）读取 pending 消息
XREADGROUP GROUP mygroup consumer1 STREAMS mystream 0
# 返回曾经投递给 consumer1 但未 ACK 的消息

# Step 3: 处理完成后确认
XACK mystream mygroup 1699999999999-0

或者使用 XAUTOCLAIM 自动化（推荐）：

# 自动认领其他消费者超过 30 秒未确认的消息
XAUTOCLAIM mystream mygroup consumer1 30000 0-0 COUNT 10
# 处理返回的消息 → XACK → 继续循环直到返回 "0-0"

策略 4：NOACK 模式（不推荐）

XREADGROUP GROUP mygroup consumer1 NOACK STREAMS mystream >

在 NOACK 模式下，消息一旦投递即被视为确认，不会进入 pending 列表。最大的风险是没有故障恢复能力——如果消费者在消息处理过程中崩溃，消息就丢了。

🚨 强烈建议：永远不要在生产环境使用 NOACK。即使是简单场景，也建议保留 ACK 机制——它几乎是零成本的保障。

4.6 🔬 深入原理

Stream 底层存储：Rax 基数树

Redis Stream 底层使用 Rax（Radix Tree） 来存储消息，而不是跳表或红黑树：

                        消息 ID: 1699999999999-0
                                   │
                        将 ID 拆分为 key 序列
                                   │
              ┌────────────────────┴────────────────────┐
              ▼                                         ▼
    Rax 树节点存储消息元数据                  消息体（field-value pairs）
       │                                             存储在 listpack 中
       ▼
    每个节点有 [iskey, iscompr, data] 标记

为什么选择 Rax 树？

特性	Rax 树	跳表	红黑树
内存效率	✅ 高（前缀压缩）	❌ 低（指针多）	❌ 低（指针多）
范围查询	✅ O(N+M)	✅ O(logN+M)	✅ O(logN+M)
ID 查找	✅ 按前缀快速定位	✅	✅
前缀共享	✅	❌	❌

Rax 树的关键优势是前缀压缩：相邻的消息 ID 共享相同的前缀（如 1699999999999），Rax 树可以将这些共享部分压缩为一个节点，大幅节省内存。

消息 ID 单调递增保证

消息 ID 格式：<毫秒时间戳>-<序号>

如果系统时钟回拨：
  - Redis 会等待直到时间追上上次的毫秒时间戳
  - 如果同一毫秒内序号到达上限（18446744073709551615），
    会等待进入下一毫秒
  - 这保证了 ID 绝对单调递增，不会出现"后写入的消息 ID 反而更小"的情况

消费者组状态存储

消费者组状态以 key 的形式存储在 Redis 数据库中，格式为：

{stream-key}
├── groups: 消费者组字典
│   ├── {group-name}: 组信息
│   │   ├── last_delivered_id
│   │   ├── consumers: 消费者字典
│   │   │   ├── {consumer-name}: 消费者信息
│   │   │   │   ├── pending_ids: Rax 树（存储该消费者的 pending 消息 ID）
│   │   │   │   └── seen_time
│   │   ├── pel: Rax 树（Pending Entries List，存储组级别 pending）

消费者组信息存储在 Stream key 的 key 元数据中（不占用额外 key 空间）。
pending list 也使用 Rax 树实现，方便按 ID 查找和范围扫描。

Pending List 实现

Pending 列表为每条 pending 消息存储：

pending entry = {
    id: "1699999999999-0",
    consumer: "consumer1",     // 当前所有者
    delivery_time: 1700000000000,  // 投递时间（毫秒）
    delivery_count: 3          // 投递次数
}

delivery_time 用于计算 “IDLE 时间”（空闲时间 = 当前时间 - delivery_time）。
delivery_count 用于追踪重试次数，辅助判断是否属于"毒药消息"（反复失败的消息）。

5. 三种方案对比

5.1 核心特性对照

特性	List	PubSub	Stream
消息持久化	✅ 受 RDB/AOF 保护	❌ 纯内存，重启丢失	✅ 受 RDB/AOF 保护
消息确认 (ACK)	❌ RPOP 即删除	❌ 即发即弃	✅ XACK 精确确认
消费者组	❌	❌	✅ 组内竞争 + 组间独立
消息回溯	❌ 消费即删除	❌ 不存历史	✅ 按 ID 范围查询任意历史
通配符订阅	❌	✅ PSUBSCRIBE	❌
阻塞读取	✅ BRPOP/BLPOP	❌ subscribe 阻断连接	✅ XREAD BLOCK
多消费者模型	竞争（一条消息一个消费者）	广播（所有订阅者收到）	组内竞争 + 组间广播
消息 ID	❌ 无序标识	❌ 无标识	✅ 全局唯一、单调递增
消息过期/裁剪	✅ LTRIM	❌	✅ MAXLEN / MINID
集群友好	✅ 按 key 分 slot	❌ 全集群广播（7.0 前）	✅ 按 key 分 slot
最大容量	4GB/key（受限于单 key）	受内存限制	受内存限制（可裁剪）
消费进度跟踪	❌	❌	✅ last_delivered_id

5.2 场景选型矩阵

场景	List	PubSub	Stream	推荐理由
简单任务队列（允许少量丢失）	✅	❌	✅	List 最简单；Stream 更可靠
任务队列（不允许丢失）	❌	❌	✅	Stream 有 ACK + pending 恢复
实时通知（服务端推送给客户端）	❌	✅	❌	PubSub 延迟最低
多服务独立消费同一消息	❌	✅	✅	PubSub 广播；Stream 多消费者组
配置变更广播	❌	✅	❌	无需保存历史
订单/支付流水	❌	❌	✅	ACK + 持久化 + 故障恢复
事件溯源	❌	❌	✅	按时间范围回溯 + 不可变消息
聊天消息分发	❌	✅	❌	实时、无需持久化
高吞吐日志收集	❌	❌	✅	Stream + 多消费者组并行
轻量级 Worker Pool	✅	❌	❌	List 实现最简单

5.3 与主流 MQ 对标

Redis 方案	对标的专业 MQ	相似度	差距
List	—（过于简陋）	—	无 ACK、无持久化保证
PubSub	NATS (at-most-once)	中	缺少集群广播优化（7.0 缓解）
Stream	Kafka	高	缺少分区、事务消息、磁盘存储、TTL 自动清理

💡 最佳实践：Stream 功能虽然对标 Kafka，但在磁盘存储策略、分区机制、事务支持方面仍有差距。对于高吞吐量（日均亿级）的日志/事件系统，Kafka 仍是更优解；对于中小规模的可靠消息队列，Stream 是更轻量的选择。

6. 🚨 常见陷阱

陷阱 1：PubSub 消息"无条件丢失"

这是 PubSub 最致命的特性：没有消费者订阅的频道，PUBLISH 的消息直接丢弃，返回值是 (integer) 0。

# 此时没有任何订阅者
PUBLISH order.created "new_order"
# 返回: (integer) 0    ← 消息已丢失，无法找回

解决：对于需要可靠投递的场景，使用 Stream 替代。如果需要保留 PubSub 的低延迟优势，可以同时写入 Stream 做持久化兜底。

陷阱 2：SUBSCRIBE 后不能执行其他命令

客户端执行 SUBSCRIBE 后进入"订阅模式"，该连接无法执行 GET、SET、XREAD 等普通命令。

# 错误示例（同一个连接）
127.0.0.1:6379> SUBSCRIBE news
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "news"
3) (integer) 1

127.0.0.1:6379> GET somekey
# (error) 订阅模式下无法执行此命令

解决：使用两个连接——一个专门用于订阅（进入 subscribe 模式），另一个用于普通命令。Go 客户端通常自动管理连接池，但需要留意：订阅连接的 goroutine 和普通命令的 goroutine 应使用不同的客户端实例或显式控制连接。

// Go 示例：两个独立连接
subConn := rdb.Subscribe(ctx, "news")  // 订阅连接
// 在另一个 goroutine 中接收消息
for msg := range subConn.Channel() {
    handleMessage(msg)
}

// 普通命令使用默认连接池
val, err := rdb.Get(ctx, "somekey").Result()

陷阱 3：Stream 忘记 XACK 导致 Pending 堆积

这是 Stream 使用者最常犯的错误。每条通过 XREADGROUP 读取的消息必须在处理完后调用 XACK，否则会永远留在 pending 列表中。

# 错误流程
XREADGROUP GROUP g1 c1 STREAMS mystream >
# 处理消息 msg_id_1...
# ❌ 忘记 XACK mystream g1 msg_id_1

# 后果：
# - XPENDING 看到消息持续积累
# - 消费者重启后，这些消息会被重新投递（可能重复处理）
# - pending 列表无限膨胀，消耗内存

正确的消费流程：

for {
    msgs, _ := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group:    "mygroup",
        Consumer: "consumer1",
        Streams:  []string{"mystream", ">"},
        Block:    5 * time.Second,
    }).Result()

    for _, msg := range msgs[0].Messages {
        // 1. 处理消息
        if err := processMessage(msg); err != nil {
            log.Printf("处理消息失败: %v, ID=%s", err, msg.ID)
            // 消息不 ACK，留在 pending 中等待重试
            continue
        }
        // 2. 处理成功后 ACK
        rdb.XAck(ctx, "mystream", "mygroup", msg.ID)
    }
}

陷阱 4：Stream MAXLEN 精确裁剪性能开销大

# ❌ 精确裁剪：每次 XADD 都可能触发大量删除操作
XADD mystream MAXLEN = 1000000 * field value

# ✅ 近似裁剪：允许略超 100 万，性能好得多
XADD mystream MAXLEN ~ 1000000 * field value

= 精确裁剪要求 Stream 长度严格 ≤ N，Redis 需要精确计算并删除刚好（当前长度 - N）条消息。在高吞吐量场景下（每秒几万条 XADD），这种精确计算会产生明显的性能开销。

~ 近似裁剪允许一定比例的"溢出"（通常不超过几十条），避免了精确计数和删除的开销。

⚡ 生产环境永远使用 ~ 近似裁剪。

陷阱 5：XREAD BLOCK 0 永久阻塞的客户端管理

# 永久阻塞——如果 Stream 长时间没有新消息
XREAD BLOCK 0 STREAMS mystream $

BLOCK 0 表示无限期阻塞，相当于客户端挂起在 Redis 上，直到有消息为止：

连接可能被 Redis 服务端 timeout 配置断开（timeout 参数默认 0 表示不主动断开，但很多部署会配置非零值）。
客户端无法响应健康检查：容器编排系统（如 K8s）可能认为消费者已挂掉，触发重启。
Redis 服务器端会占用一个连接和少量内存维护这个阻塞客户端。

解决方法：

# ✅ 使用有限超时 + 循环
while true; do
  XREAD BLOCK 5000 STREAMS mystream $    # 5 秒超时
  # 超时后返回循环顶部，继续 BLOCK
done

在 Go 客户端中，连接是有超时配置的，但仍建议使用有限 BLOCK 时间（如 5 秒），让消费者有机会检测到故障、重新连接或更新心跳。

陷阱 6：消费者组数量过多时的内存开销

每个消费者组维护：

last_delivered_id
所有消费者的 pending list（每个消费者一个 Rax 树）

如果创建大量消费者组（如按用户 ID 动态创建），内存和管理开销会线性增长。

# ❌ 反模式：每个用户一个消费者组
XGROUP CREATE stream user_group_001 ...    # 用户 001
XGROUP CREATE stream user_group_002 ...    # 用户 002
# ... 可能创建数万个消费者组

# ✅ 推荐：用一个消费者组，消费者按逻辑区分
# 所有用户共享一个消费者组，消费者内部路由到不同处理逻辑

💡 最佳实践：消费者组数量应在几十到几百以内。如果需要按租户隔离消费进度，考虑用不同 Stream key 而非不同消费者组。

陷阱 7：单条消息体不要过大

Stream 消息体存储在一个 key 中。虽然 Redis 没有显式限制单条消息的大小，但：

大消息会增加网络传输延迟
大消息会导致 Rax 树节点膨胀
XREAD 返回大消息时可能阻塞 Redis 的事件循环
裁剪（XTRIM/MAXLEN）删除大消息时内存回收不及时

# ❌ 不推荐：将整个文件内容放入 Stream
XADD filestream * filename "report.pdf" data "<几 MB 的 Base64 内容>"

# ✅ 推荐：Stream 中存储元数据/引用，实际数据存对象存储（如 S3/MinIO）
XADD filestream * filename "report.pdf" size 1048576 url "s3://bucket/report.pdf"

💡 最佳实践：将 Stream 消息体控制在 1MB 以内，推荐 几 KB 到几百 KB。需要传输大文件时，Stream 只记录引用（URL / 路径）。

陷阱 8：消费者崩溃后 Pending 消息无人处理

# consumer1 从 Stream 读取了 100 条消息
XREADGROUP GROUP mygroup consumer1 COUNT 100 STREAMS mystream >
# ... consumer1 崩溃，未 ACK 任何消息 ...

# 问题：100 条消息永远卡在 consumer1 的 pending 列表中
# 组内其他消费者（consumer2）读不到这些消息（用 > 只返回新消息）
XREADGROUP GROUP mygroup consumer2 STREAMS mystream >
# 返回: (nil)   ← consumer2 看不到 consumer1 的 pending

解决：每个消费者必须实现 pending 消息的认领逻辑：

func monitorPending() {
    for {
        // 认领其他消费者空闲超过 30 秒的消息
        result, _ := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
            Stream:   "mystream",
            Group:    "mygroup",
            Consumer: "healthy-consumer",
            MinIdle:  30 * time.Second,
            Start:    "0-0",
            Count:    10,
        }).Result()

        for _, msg := range result.Messages {
            processMessage(msg)      // 处理消息
            rdb.XAck(ctx, "...", "...", msg.ID)  // 确认
        }

        if result.Start == "0-0" {
            break  // 扫描完成
        }
        time.Sleep(time.Second)
    }
}

💡 最佳实践：这是 Stream 生产部署的必做功课。每个消费者启动时以及周期性执行 XAUTOCLAIM，确保没有"孤儿消息"。

陷阱 9：滥用模式订阅（PSUBSCRIBE）导致性能下降

# ❌ 创建大量模式订阅
PSUBSCRIBE *
PSUBSCRIBE a.*
PSUBSCRIBE b.*
# ... 创建数百个模式

# 后果：每次 PUBLISH 都要遍历整个 pubsub_patterns 链表
# 链表长度 = 所有模式订阅数量，复杂度 O(N)

每次 PUBLISH 到任意频道时，Redis 都会遍历 pubsub_patterns 链表（而非直接用哈希查找），检查模式是否匹配。模式越多，PUBLISH 越慢。

解决：

尽量使用精确频道名（SUBSCRIBE），利用 O(1) 的 pubsub_channels 字典查找。
如果必须用模式，控制模式数量在几十个以内。
使用固定前缀分层设计，如 app.module.event 而非深层通配。

陷阱 10：List 队列消费速度跟不上生产速度

# 生产者快速写入
for i in {1..1000000}; do
  LPUSH task_queue "task_$i"
done

# 消费者单线程慢速处理
BRPOP task_queue 0  # 每次只取一条，处理完再取

问题表现：

List 队列堆积百万级消息，内存持续增长
单个 LLEN task_queue 返回巨大数字
消费者永远追不上生产者

解决方案：

# 1. 批量消费（使用 LRANGE + LTRIM 而非 BRPOP）
LRANGE task_queue 0 99     # 一次取 100 条
LTRIM task_queue 100 -1    # 删除已取出的 100 条

# 2. 转用 Stream + 多消费者并行消费
# 3. 转用 Stream + 消费者组水平扩容

💡 最佳实践：当 List 队列长度持续超过数千条时，就该考虑升级到 Stream 方案了。

陷阱速查清单

编号	陷阱	严重程度	解决方案
1	PubSub 消息无条件丢失	🔴 致命	关键数据用 Stream 兜底
2	subscribe 后占用连接	🟡 注意	使用两个独立连接
3	忘记 XACK 导致 pending 堆积	🔴 致命	处理完成后立即 XACK
4	MAXLEN = 精确裁剪性能差	🟡 注意	生产环境用 ~ 近似裁剪
5	BLOCK 0 永久阻塞问题	🟡 注意	使用有限超时（如 5s）+ 循环
6	过多消费者组内存开销	🟡 注意	消费者组控制在几十~几百
7	消息体过大	🟡 注意	控制在 1MB 以内，大文件用引用
8	消费者崩溃后 pending 无人处理	🔴 致命	实现 XAUTOCLAIM 认领逻辑
9	大量模式订阅拖慢 PUBLISH	🟡 注意	控制模式数量，优先精确订阅
10	List 队列消费积压	🟡 注意	批量消费或升级到 Stream

Redis 数据类型完整参考持久化