Redis Stream 是 Redis 5.0 引入的专用数据结构，天生为消息队列而生：它用只追加的日志保存消息，并支持多个消费者组成消费组，既能保证消息不丢失，又能水平扩展处理能力。相比 List 和 Pub/Sub，Stream 在可靠性和可观测性上更胜一筹，是中小规模实时任务的理想选择。当你需要”发出去就一定要被处理一次”的语义时，它比发布订阅可靠得多。
一、为什么选 Stream，而不是 List 或 Pub/Sub
很多团队最早用 Redis 做队列时，会拿 List 配 BLPOP，或用 Pub/Sub 做发布订阅。但两者都有硬伤：List 没有原生的”消费组”概念，多消费者抢同一队列时难以做负载均衡与失败重投；Pub/Sub 则是”发完即焚”，下游宕机期间的消息直接丢失。如果你已经读过了《Redis 核心数据结构实战》，会知道五种基础类型里并没有一个结构能同时满足”持久化 + 多消费者 + 可追溯”。Stream 正是为此补齐的拼图：它把”消息存储””多消费者分工””失败兜底”三件事一次性做齐，不用你在业务代码里再发明一套确认机制。
| 维度 | List | Pub/Sub | Stream |
|---|---|---|---|
| 消息持久化 | 是（但无消费组） | 否（发完即丢） | 是 |
| 消费组 | 需自己实现 | 不支持 | 原生支持 |
| 消息追溯 | 有限 | 不支持 | 支持 XRANGE |
| 未确认重投 | 无 | 无 | PEL + XACK |
| 阻塞读取 | BLPOP | 订阅推送 | XREAD BLOCK |
二、基础命令三板斧：XADD / XRANGE / XREAD
Stream 的接口设计得很克制，核心就三个写读动作。XADD 负责写，XRANGE 负责按区间读，XREAD 负责阻塞地等新消息。先理解这三个，消费组只是在它们之上加了一层”组 + 消费者”的语义。下面每条命令都可以在 redis-cli 里直接敲，建议边看边试。
XADD 写入消息
XADD 向流里追加一条消息，* 让 Redis 自动生成 ID（毫秒时间戳-序号）。你也可以加 MAXLEN ~ N 做近似裁剪，避免流无限膨胀。
# 写入一条消息，* 自动生成 ID
XADD order_stream * event create_order user_id 1024 amount 99.00
# 返回 1717000000000-0 这类 ID
# 限制长度，近似截取，防止内存无限增长
XADD order_stream MAXLEN ~ 10000 * event pay user_id 1024 amount 9.90XRANGE 按区间读取
用 - 和 + 表示最小、最大 ID，可配合 COUNT 分页，这是 Stream 能做”消息追溯”的关键。比起 List 只能从两端弹，Stream 可以随机访问任意一段历史。
# 读取全部消息
XRANGE order_stream - +
# 只取最近 10 条
XRANGE order_stream - + COUNT 10XREAD 阻塞读取最新消息
# 阻塞等待新消息，从尾部 $ 读，最多等 5 秒
XREAD COUNT 5 BLOCK 5000 STREAMS order_stream $注意 $ 表示”只取调用之后新到达的消息”，不会回放历史；若想从头消费，第一次应传 0 或某个具体 ID。这个细节在消费组里同样重要。
三、消费组：XGROUP 与 XREADGROUP
消费组是 Stream 的精髓：一条消息只会被组内一个消费者拿到，处理成功后再显式确认，没确认的消息留在 PEL（Pending Entries List）里等待重投。这样既做了负载均衡，又保证了”至少一次”投递。一个流可以挂多个消费组，彼此独立消费同一份数据，常用于”主流程处理”和”异步审计”两套逻辑互不干扰地读同一队列。
创建消费组
# 0 表示从头消费；$ 表示从最新开始；MKSTREAM 自动建流
XGROUP CREATE order_stream g1 0 MKSTREAM组内消费与 PEL
# 消费者 c1 读取"尚未分配给本组任何人"的消息（> 表示新消息）
XREADGROUP GROUP g1 c1 COUNT 5 BLOCK 5000 STREAMS order_stream >
# 处理成功后确认，消息移出 PEL
XACK order_stream g1 1717000000000-0
# 查看未确认（PEL）消息，用于监控与重投
XPENDING order_stream g1XPENDING 返回的是”已投递但未确认”的消息清单，包含每条的 ID、所属消费者、空闲时长、投递次数。它是你判断消费者是否卡死的第一手证据。
四、Stream 的 ID 含义与内部结构
每一条 Stream 消息都带一个 ID，格式是 毫秒时间戳-序号（如 1717000000000-0）。时间戳部分让你可以用 XRANGE 按时间范围直接捞数据，比如 XRANGE order_stream 1717000000000 1717000009999 就能取出某一秒内的全部消息，这在排查”某分钟下单陡降”类问题时非常顺手。序号在同一毫秒内自增，保证同一毫秒内多条消息也有稳定顺序，不会出现乱序。底层实现上，Stream 用基数树（radix tree）按 ID 前缀组织消息，叶子是 listpack 紧凑存储，因此即便存百万条，内存占用也远低于等量的独立 key；这也是它敢于”只追加、可追溯”的底气。换句话说，Stream 不是把消息塞进一个超长列表，而是在时间和空间两个维度上都做了紧凑编码。
五、实战：用 Stream 搭一个可靠任务队列（Python）
下面用 redis-py 实现一个最小可用、带消费组与确认机制的队列。生产环境起 Redis 最简单的方式是用 Docker Compose 一键编排。比起手写 List + 轮询，消费组自动帮你做分工与重投，业务侧只需关心”怎么处理”和”什么时候 ack”。
docker run -d --name redis -p 6379:6379 redis:7生产者
import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
def publish(event, payload):
# maxlen 控制流长度，近似裁剪避免内存膨胀
msg_id = r.xadd("order_stream", {"event": event, **payload}, maxlen=10000)
return msg_id
publish("create_order", {"user_id": "1024", "amount": "99.00"})消费者（消费组）
import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.xgroup_create("order_stream", "g1", id="0", mkstream=True)
def handle(fields):
# 你的业务逻辑：落库、调下游、发通知……
print("handle", fields)
while True:
resp = r.xreadgroup("g1", "c1", {"order_stream": ">"}, count=5, block=5000)
if not resp:
continue
for _, messages in resp:
for msg_id, fields in messages:
try:
handle(fields)
r.xack("order_stream", "g1", msg_id) # 成功才确认
except Exception:
# 不 ack，消息留在 PEL，稍后可重投
pass死信与重试策略
消费失败不要立刻重试到天荒地老。结合幂等设计的思路：用 XPENDING 捞出闲置过久的 PEL 消息，按重试次数决定重新投递还是转入死信流（如 order_stream_dlq）。业务处理本身要保证幂等，避免重复消费造成副作用。一个稳妥的做法是：单条消息最多重试 3 次，超过就 XACK 并写入死信流，由人工或独立脚本兜底，既不让 PEL 无限膨胀，也不丢消息。
六、Stream 与 Kafka / RabbitMQ 怎么选
| 维度 | Redis Stream | Kafka | RabbitMQ |
|---|---|---|---|
| 吞吐 | 中 | 极高 | 中高 |
| 持久化 | AOF/RDB | 分区日志 | 队列+镜像 |
| 消费组 | 原生 | 原生 | 需 Exchange |
| 运维复杂度 | 低 | 高 | 中 |
| 适用场景 | 中小规模、低延迟 | 海量日志/事件 | 企业级消息 |
如果你的消息量还没到 Kafka 的”主场”，又想要比 RabbitMQ 更轻的运维，Stream 是极佳折中。当然，当积压真成了问题，也要参考Kafka 消息积压排查那样，从消费 lag 入手定位瓶颈。Stream 也有 lag 概念，只是它体现在每个消费组的 PEL 规模上。
七、监控与容量规划
上线后真正要盯的是 PEL 的长度，而不是流本身有多长。可以用 XINFO GROUPS 看每个消费组的 pending 数，用 XINFO STREAM 看消息总量与首尾 ID 间隔，推算堆积速度。建议把”某消费组 pending 超过 N 条且持续 5 分钟”设为告警，而不是等消息真正丢失才发现问题。容量方面，单条消息建议控制在几 KB 以内，字段用简短 key（如 u 代替 user_id）能显著省内存；当历史不再需要时，用 XTRIM 主动裁剪，比依赖写入时的 MAXLEN 更可控。配合 Redis 的 AOF 持久化，即便进程重启，已确认的消息与 PEL 都不会丢。
# 查看流概况：消息数、首尾 ID、消费组数
XINFO STREAM order_stream
# 查看消费组 pending 情况
XINFO GROUPS order_stream
# 按时间裁剪，只保留该 ID 之后的消息
XTRIM order_stream MINID 1717000000000八、生产环境避坑清单
| 坑 | 表现 | 对策 |
|---|---|---|
| 未 ACK 堆积 | PEL 膨胀、内存上涨 | 处理完立即 XACK，监控 XPENDING |
| 消费组 ID 用 $ | 历史消息读不到 | 新组用 0 从头，续跑用具体 ID |
| 不裁剪流 | 内存无限增长 | 写时加 MAXLEN ~ N，或定期 XTRIM |
| 单条消息过大 | 网络/序列化慢 | 控制字段大小，大负载走对象存储 |
九、小结
Redis Stream 用”只追加日志 + 消费组 + PEL 确认”三件套，把消息队列的可靠性一次性做齐：XADD 写、XREADGROUP 消费、XACK 确认、XPENDING 兜底重投、XTRIM 控容量。相比 List/Pub-Sub 它更可靠，相比 Kafka/RabbitMQ 它更轻量。配合Redis 缓存设计中的高可用思路，足以撑起绝大多数中小业务的异步解耦需求。当你下次想用 Redis 做队列时，先问一句：需要消费组和可靠确认吗？如果需要，Stream 就是答案。
