Redis Streams 更推荐用于大规模事件总线，因其具备持久化存储、消费者组偏移管理、PEL未确认队列及XACK确认机制，保障“至少一次”投递；而Pub/Sub纯内存广播、无持久化、无ACK、断连即丢消息，仅适用于瞬时通知场景。
为什么不用 Redis Pub/Sub 做事件总线
直接用 Redis Pub/Sub 发布订阅，看起来最省事，但上线后大概率会丢消息——它不持久、无确认、无重试，客户端断连期间所有消息彻底消失。企业级事件总线要求“至少一次投递”，这点 Pub/Sub 从底层就不支持。
而 Stream 是 Redis 5.0+ 引入的持久化日志结构，天然支持：消息写入即落盘（RDB/AOF）、消费者组（consumer group）、消息确认（XACK）、未处理消息自动进入 PENDING 列表、支持按 ID 回溯消费。这才是可落地的事件总线底座。
- 别把
Pub/Sub当总线用，它只适合瞬时通知（如服务心跳广播） -
Stream的每个消息都有唯一 ID（如1640995200000-0），便于幂等和追踪 - 一个
Stream可被多个消费者组独立消费，比如order-processing和analytics组互不影响
如何用 StackExchange.Redis 封装 Stream 事件总线
官方推荐的 .NET Redis 客户端 StackExchange.Redis 对 Stream 支持完整，但原生 API 较底层，需封装关键逻辑：
- 发布事件：用
StreamAddAsync写入，key 建议统一前缀如event:orders，避免 key 冲突 - 创建消费者组：首次消费前必须调用
StreamCreateConsumerGroupAsync，指定起始 ID（0表示从头开始，$表示只消费新消息） - 阻塞读取：用
StreamReadGroupAsync+count=1+block=5000实现低延迟拉取，避免空轮询 - 手动确认：处理成功后必须调用
StreamAcknowledgeAsync，否则消息会一直留在PENDING
示例关键调用：
await db.StreamAddAsync("event:orders", new NameValueEntry("type", "OrderCreated"), new NameValueEntry("data", json));
await db.StreamCreateConsumerGroupAsync("event:orders", "order-processing", "0");
var messages = await db.StreamReadGroupAsync("event:orders", "order-processing", "worker-01", count: 1, block: 5000);
await db.StreamAcknowledgeAsync("event:orders", "order-processing", messages[0].MessageId);
消费者组设计与故障恢复要点
消费者组不是“开箱即用”的高可用方案，几个关键点不处理就会卡死：
Redis Skill - 高性能缓存管理
下载
Redis 缓存和数据结构管理技能。通过自然语言操作 Redis，支持 String、Hash、List、Set、ZSet、Stream 等数据结构操作。当用户提到 Redis、缓存、消息队列、会话存储时使用此技能。
- 消费者进程崩溃后，其
PENDING消息不会自动释放——必须设置超时时间，用StreamPendingAsync扫描并转移给其他 worker（例如idle > 300000即 5 分钟未处理） - 同一个消费者组内，不同 worker 必须使用**唯一 consumer name**（如
worker-01,worker-02），否则XACK会混乱 - 不要在单个消费者里并发处理多条消息——
Stream的PENDING状态是 per-consumer 的，乱序ACK会导致后续消息无法被正确标记 - 重启消费者时，先调用
StreamClaimAsync抢占超时的PENDING消息，再开始正常读取
事件 Schema 与类型安全怎么落地
.NET 生态里最容易踩的坑，是把 JSON 字符串当事件体直接扔进 Stream，结果后期改字段名、加必填项、类型变更全靠人工对齐。
建议强制走 System.Text.Json 序列化 + 接口契约：
- 定义基类
IEvent，含Id（Guid）、Timestamp、Type（字符串标识，如"OrderCreated"） - 每个具体事件继承该接口，用
[JsonPolymorphic]或手动switch (type)反序列化，避免运行时JsonException - 在消费者启动时校验
Stream中历史消息是否仍能反序列化，失败则告警而非静默跳过
这样做的代价是序列化稍慢一点，换来的是事件演进可控、错误可定位、上下游契约清晰——企业系统里，这点性能损失远小于一次线上事件解析失败导致的数据不一致。
真正难的不是把消息发出去，而是确保每一条都可追溯、可重放、可验证。Stream 提供了能力，但怎么组织消费者生命周期、怎么管理 schema 变更、怎么监控 pending 积压，这些才是压垮团队的细节。
