Redis Streams 是 Redis 5.0 引入的流数据类型，专门用来构建消息队列与事件流系统。它既有 Kafka 般的可持久化、消费者组能力，又保留了 Redis 的极低延迟，非常适合做轻量级实时消息总线。
一、为什么需要 Redis Streams
在做异步解耦、事件通知时，很多团队第一时间想到 Kafka，但它部署与运维成本不低。而项目里往往已经跑着 Redis 做缓存——如果能用 Redis 直接承载中小规模的消息流，就省掉一套独立中间件。问题在于：Redis 早期的数据结构都不太适合做”正经”的队列。
1.1 老方案的局限
用 List 的 LPUSH / BRPOP 可以做队列，但缺少消费者组与确认机制；Pub/Sub 更是”发后既忘”，订阅者离线消息直接丢失。Streams 正是为填这个坑而生。
| 方案 | 持久化 | 消费者组 | 回溯读取 | 主要局限 |
|---|---|---|---|---|
| List + LPUSH/BRPOP | 有 | 无（需自管） | 难 | 无法多消费者分摊，无 ACK 确认 |
| Pub/Sub | 无 | 无 | 不支持 | 订阅者离线即丢消息 |
| Streams | 有 | 有 | 支持 | 单流写吞吐低于 Kafka |
二、核心命令速查
Streams 的命令围绕”追加—读取—确认—裁剪”四个动作展开，下面这张表覆盖了日常 90% 的场景。
| 命令 | 作用 |
|---|---|
| XADD | 追加一条消息，返回消息 ID |
| XREAD | 读取消息，支持阻塞等待新消息 |
| XRANGE | 按 ID 范围读取（用于回溯历史） |
| XGROUP | 创建 / 管理消费者组 |
| XREADGROUP | 消费者组内读取，自动标记 pending |
| XACK | 确认消息已处理 |
| XPENDING | 查看未确认（pending）消息 |
| XCLAIM | 把超时消息转移给其它消费者 |
| XTRIM | 裁剪流长度，控制内存 |
三、写入数据：XADD
XADD 向流追加一条消息，字段用 key-value 对表示；ID 传 * 时 Redis 自动生成”毫秒时间戳-序号”形式的 ID，保证单调递增。
消息 ID 形如 1727174400000-0：前半是毫秒时间戳，后半是同一毫秒内的自增序号。因为它随写入单调递增，天然支持按时间排序与范围回溯，这也是 Streams 能做”事件溯源”式回放的基础——你随时可以用 XRANGE 把任意时间段的事件重新读出来重放。
# 追加一条订单创建事件，* 让 Redis 自动生成 ID
XADD orders * event "created" order_id 10231 user_id 8848 amount 199.00
# 返回示例：1727174400000-0
# 读取整条流（按 ID 范围）
XRANGE orders - +
四、读取数据：XREAD 与阻塞读取
普通 XREAD 适合回溯历史；加上 BLOCK 后则变成”有消息立即返回，否则等待”，等价于阻塞队列的 BRPOP，是实时消费的关键。
# 从开头读取最多 2 条（用于历史回放）
XREAD COUNT 2 STREAMS orders 0
# 阻塞读取最新消息，$ 表示当前尾部；最多等待 5000ms
XREAD BLOCK 5000 COUNT 10 STREAMS orders $
五、消费者组：竞争消费与可靠确认
单个 XREAD 无法满足”多实例分摊 + 不丢消息”。消费者组（Consumer Group）解决了这个问题：组内每条消息只投递给一个消费者，处理完必须 XACK 确认，未确认的消息会进入 pending 列表，可重新投递。
这里有个关键点：XREADGROUP 拉走的消息不会立刻从流中删除，而是进入该消费者的 pending 列表。只有当 XACK 被调用，消息才真正”消费完成”。如果消费者进程中途崩溃，未 ACK 的消息会在下次被重新投递给其他健康消费者，从而做到”至少一次”投递语义——代价是业务必须幂等，这正是第八节要强调的。
# 创建消费者组 cg1，MKSTREAM 在流不存在时自动建流
XGROUP CREATE orders cg1 $ MKSTREAM
# 消费者 consumer-1 拉取新消息（> 表示"尚未投递过"的消息）
XREADGROUP GROUP cg1 consumer-1 COUNT 5 STREAMS orders >
# 业务处理成功后确认，消息才从 pending 移除
XACK orders cg1 1727174400000-0
六、完整示例：Python 订单事件流
下面用 redis-py 给出一个最小可跑的生产者 / 消费者示例。注意先 XGROUP CREATE 建组，消费时再 XACK，形成”至少一次”投递。
import redis
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
STREAM = "orders"
# 生产者：追加订单事件
def produce():
r.xadd(STREAM, {"event": "created", "order_id": 10231, "amount": 199.0})
print("published")
# 消费者：基于消费者组拉取，处理完再 ACK
def consume():
while True:
resp = r.xreadgroup("cg1", "consumer-1", {STREAM: ">"}, count=5, block=5000)
if not resp:
continue
for _, messages in resp:
for msg_id, fields in messages:
print("handle", msg_id, fields)
r.xack(STREAM, "cg1", msg_id) # 成功处理才确认
if __name__ == "__main__":
produce()
consume()
真实项目里建议在 consume() 外层包一层 try/except：处理异常时故意不调用 XACK，让消息留在 pending 等待重试；再配合第八节的 XPENDING / XCLAIM，就能构建出健壮的”自动重试闭环”，避免一条坏消息把整个消费循环拖死。
七、Redis Streams 还是 Kafka？
两者并非互斥。Streams 胜在”零额外组件、低延迟”，适合中小规模、已有 Redis 的内部事件；Kafka 胜在超大规模与强持久，适合核心交易链路。选型时先看吞吐与持久化诉求，再决定用哪把锤子。关于 Kafka 的选型与可靠性设计，可延伸阅读 Kafka 入门实战：消息队列选型与可靠性保障。
更常见的做法是”分级事件总线”：用 Kafka 扛支付、订单落库等强持久核心链路，用 Redis Streams 承载通知、审计、内部状态同步等对延迟敏感、可容忍少量重放的旁路事件，两者各取所长。这样既能守住核心链路的可靠性，又不必为每一个小需求都背上 Kafka 的运维负担。
| 维度 | Redis Streams | Kafka |
|---|---|---|
| 部署复杂度 | 极低（已有 Redis 即可） | 较高（Broker + 协调组件） |
| 单流吞吐 | 数万 QPS | 数十万~百万 QPS |
| 持久化 | AOF/RDB，可配置 | 分段文件，强持久 |
| 典型场景 | 轻量通知、内部事件流 | 核心链路、多租户日志 |
八、三个常见坑与避坑清单
8.1 消息无限堆积导致内存膨胀
Streams 默认只增不删，长期运行会撑爆内存。务必设置长度上限：写入时用 MAXLEN，或定期 XTRIM 裁剪。
# 写入时直接限制长度（~ 表示近似裁剪，性能更好）
XADD orders MAXLEN ~ 10000 * event "created" order_id 10232
# 或事后裁剪，只保留最近 10000 条
XTRIM orders MAXLEN 10000
8.2 重复消费与幂等
消费者组保证”至少一次”投递：进程崩溃会让消息留在 pending 中，恢复后重新投递。业务侧必须按消息 ID 做幂等（如用 order_id 去重），否则同一订单可能被处理两次。Redis 缓存层面的穿透/击穿防护思路也值得参考，详见 Redis 缓存设计：穿透、击穿、雪崩与最佳实践。
8.3 pending 堆积成”死信”
消费者异常退出会让消息永远 pending。XPENDING 监控、XCLAIM 转移超时消息是标准解法；重试超过阈值就搬到独立死信流，避免阻塞主流程。
# 查看超过 30s 未确认的消息
XPENDING orders cg1 IDLE 30000
# 把超时消息转移给 consumer-2 重试
XCLAIM orders cg1 consumer-2 30000 1727174400000-0
九、生产落地建议
- 始终为流设置
MAXLEN上限或定期XTRIM，守住内存水位。 - 用消费者组 +
XACK实现至少一次语义，业务侧务必做幂等。 - 用
XPENDING监控 pending，定期XCLAIM清理卡住的消息。 - 与 Kafka 不互斥：核心链路用 Kafka，轻量内部事件用 Streams。
- 若顾虑 Redis 许可与持久化，可考虑迁移到其开源分支 Valkey，迁移路径见 Valkey 实战：Redis 许可变更后的开源替代与平滑迁移；需要实时双向推送时还能配合 WebSocket 实时通信实战 把流事件推到前端。
十、小结
Redis Streams 把”消息队列 + 事件流”的能力以极低门槛带进了你本来就有的 Redis，核心就是 XADD 写入、XREADGROUP 消费、XACK 确认、XTRIM 控长四板斧。中小规模场景下，它比引入 Kafka 更划算；用对消费者组与幂等，就能稳定撑起实时事件总线。
