Redis发布订阅实现即时通讯，Stream处理离线消息

Redis的Pub/Sub虽能实现低延迟的实时消息广播，但因其纯内存、无持久化、无消费位点等特性，完全无法保障离线期间的消息可达性；真正可靠的即时通讯系统必须采用分层架构：用Stream（XADD/XREAD）作为持久化消息总线，为每位用户或会话独立建流、按ID序号精准投递离线消息，并配合last_read_id机制实现断线续传与去重；而Pub/Sub仅承担“在线即达”的轻量广播角色，二者协同——先落库（XADD）再广播（PUBLISH）——才能兼顾实时性、可靠性与可扩展性，避免订单通知丢失、客服消息遗漏等线上事故。

Redis 的 Pub/Sub 本身不存消息，离线消息必须靠外部机制补足 —— 直接用 SUBSCRIBE 接收，等于默认放弃所有断连期间的消息。

为什么不能只靠 SUBSCRIBE 做即时通讯

Pub/Sub 是纯内存广播，没有重放、无确认、无消费位点。一旦订阅者断开连接，再重连时：
– 已发布的消息全部丢失
– PUBLISH 返回值为 0 就代表这条消息彻底蒸发
– 订阅者无法知道“自己漏了多少条”，更无法请求补发

常见错误现象：
– 用户切后台/网络抖动后收不到新订单通知
– WebSocket 连接重建后，错过中间 3 条状态变更
– 后台服务重启时，SUBSCRIBE 重新执行，但历史消息全不可见

所以，任何严肃的即时通讯场景（比如外卖订单、客服消息），都必须把 Pub/Sub 当作“实时通道”，而把离线兜底逻辑交给别的结构 —— Stream 就是最自然的选择。

XADD + XREAD 实现带序号的消息持久化

Stream 天然适合做事件日志：每条消息自带唯一 ID（如 1678945623456-0），支持按 ID 范围读取、支持消费者组、支持自动过期（MAXLEN）。

关键设计点：
– 每个用户或会话对应一个独立 Stream key，例如 stream:chat:u1001-u2002
– 所有发给该用户的文本、状态、附件，统一走 XADD stream:chat:u1001-u2002 * content "hi" type "text"
– 用户上线后，先查自己上次读到哪条 ID（存在本地或 Redis String 中），再用 XREAD COUNT 100 STREAMS stream:chat:u1001-u2002 last_id+1 拉取新消息
– 不需要手动维护偏移量？可用 XINFO CONSUMERS 配合消费者组，但对单用户推送场景，直接记 last_id 更轻量

性能提示：
– XADD 是 O(1)，XREAD 是 O(N) 但 N 是返回条数，不是整个 Stream 长度
– 避免用 XREAD BLOCK 在非阻塞上下文里轮询；WebSocket 服务端可开一个长连接专门 XREAD 等待新消息

Pub/Sub 和 Stream 怎么协同工作

两者不是替代关系，而是分层协作：
– PUBLISH 只用于“此刻在线的人”——比如把新消息同时推给当前已连接的 WebSocket 客户端，实现亚秒级触达
– XADD 必须同步执行——哪怕用户离线，也要把消息落库，作为兜底依据
– 不要等 PUBLISH 成功后再写 Stream；应先 XADD，再 PUBLISH；否则出现网络分区时，可能消息进了 Stream 却没广播出去，导致“已存但未推”

典型流程示例：
– 用户 A 发送消息给 B
– 服务端：

• 写入 stream:chat:a-b 和 stream:chat:b-a（双向存）
• 查 B 的在线状态（比如 GET user:online:b）
• 若在线，PUBLISH chat:b "new_msg_id:1678945623456-0"
• 若离线，跳过 PUBLISH，仅依赖后续上线时 XREAD

容易踩的坑：
– 把 Stream 当成 Pub/Sub 的“增强版”直接替换，结果发现 XREAD GROUP 的 ACK 机制在单推场景下反而增加复杂度
– 用同一个 Stream 存所有用户消息（如 stream:all_msgs），导致读取时无法按用户隔离，也难以清理过期数据
– 忘记设置 MAXLEN ~ 10000，长期运行后 Stream 膨胀，XLEN 变慢，XTRIM 代价变高

上线时怎么安全拉取离线消息

用户重连 WebSocket 后，不能简单 XREAD STREAMS ... 0 —— 那会重复推送已读过的消息。必须依赖一个可靠的“最后已读 ID”。

推荐做法：
– 登录成功后，服务端从 user:last_read_id:b 读出上一次记录的 ID（比如 1678945623455-2）
– 执行 XREAD COUNT 200 STREAMS stream:chat:b-a 1678945623455-3（ID 加 1，避免漏掉紧邻的下一条）
– 拉到消息后，批量发送给前端，并更新 user:last_read_id:b 为最新 ID
– 若返回空，说明无新消息，直接进入正常 PUBLISH 接收流

注意：
– 不要用 XRANGE 替代 XREAD，前者是快照读，无法感知新追加消息
– 如果用户长时间未登录，Stream 可能已被 MAXLEN 截断，此时应结合业务逻辑决定是否返回“消息过多，已折叠”提示，而不是静默丢弃

最易被忽略的一点：离线消息的“送达确认”不在 Redis 层完成，它依赖业务层——比如前端收到消息后主动调用 ACK 接口，服务端才更新 last_read_id。否则，页面刷新两次就可能重复渲染同一条消息。

到这里，我们也就讲完了《Redis发布订阅实现即时通讯，Stream处理离线消息》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！