Redis Pub/Sub 断线重连后的消息丢失处理

Redis Pub/Sub 断线期间的消息不会在重连后自动补发。它提供至多一次投递语义；重新订阅只能恢复接收后续消息。若消息只是“数据已变化”的提醒，重连后重新加载权威状态；若每一条事件都必须处理，需要提前把事件写入 Redis Streams 等可回放记录，并保存消费进度。已经只通过 Pub/Sub 发出且没有其他记录的消息，不能靠重连追回。

Redis Pub/Sub 官方文档：https://redis.io/docs/latest/develop/pubsub/

先恢复订阅，再恢复业务状态

对于配置刷新、看板刷新等场景，最小处理方案是：客户端重连并确认订阅建立后，将本地状态标为待同步，从数据库或权威接口重新获取快照，然后恢复正常展示。Redis 只负责变化通知，业务接口负责当前事实。不要因为 TCP 连接恢复就直接把界面标成“数据最新”。

例如消息体只含配置项标识和版本号，订阅端收到提醒后读取配置。断线后即使漏掉多条中间通知，重新加载当前配置仍能恢复正确状态；但这个方案不适用于必须逐条执行业务副作用的事件。

快照加载期间可能仍有新通知到来。应用可先完成订阅，暂存加载期间的更新提示，待快照装载完成再按版本处理，必要时重新读取。若接口不能提供一致快照或可比较版本，就不能声称消除了全部并发窗口，应依据业务一致性要求设计同步协议。

区分订阅成功与消息补发

下面是 Bash 中调用 redis-cli 的演示命令，需在独立命令会话中分别使用。示例没有在本文写作环境执行，不提供伪造的终端输出。

# 订阅会话：等待服务端订阅确认后，才能接收该频道的新消息。
redis-cli SUBSCRIBE demo:config:changed

# 发布会话：发送一个配置变化提示，完整配置仍从权威接口读取。
redis-cli PUBLISH demo:config:changed '{"config_id":"theme","version":42}'

断开订阅会话，在它离线时发布另一条提醒，再重新执行 SUBSCRIBE。重新订阅不会拿到离线期间的历史。PUBLISH 返回的订阅接收者数量不等于业务处理成功数；回调开始执行也不代表配置已经落地。

客户端的自动重连、自动重新订阅属于具体 SDK 的行为，不能用 Redis 协议本身推断。应用应监听连接错误与订阅状态，设置有上限的退避重试，避免持续重连冲击；具体参数以所用客户端文档为准。

必须补齐事件时改用可回放记录

如果是需要逐条处理的库存变动等事件，先明确历史记录的权威来源，再决定如何接入 Streams。最简单的设计是把 Stream 当作事件读取入口，而不是先 PUBLISH、等订阅者发现丢失后再尝试补写。

# 单引号保护星号，避免被 shell 当成文件匹配；event_id 用于业务去重。
redis-cli XADD 'demo:{inventory}:events' '*' event_id evt-9001 sku item-7 delta -1
# 新消费端从保留历史起点读取；COUNT 限制本批返回条数。
redis-cli XREAD COUNT 100 STREAMS 'demo:{inventory}:events' 0-0

XADD 回执中的 Stream ID 标识流中的条目位置，event_id 则是业务事件标识。生产者超时后重试可能写出多个具有相同 event_id 的条目，不能把这两个标识混为一谈。

若业务变更在关系数据库提交，而事件写入 Redis 是另一步，两者存在双写失败窗口。可以通过数据库事务内的 outbox 记录和独立投递任务衔接，消费者仍需去重。单纯并列执行 XADD 与 PUBLISH，不构成跨系统原子提交。Pub/Sub 可以保留为唤醒提示，但断线恢复必须以可回放记录为依据。

用保存的 Stream ID 继续读取

已有消费端恢复时，XREAD 传入最后成功处理并持久化的 Stream ID，它返回该位置之后的条目。下面的数字只是格式示例，实际必须替换为自己的已保存进度，不能照抄当作当前游标。

# 示例游标来自消费端的持久化记录；读取超时无新条目时保持它不变。
saved_id='1794540000000-0'
redis-cli XREAD COUNT 100 BLOCK 5000 STREAMS 'demo:{inventory}:events' "$saved_id"

不要在每次重连时传入 $。它表示以当时流的尾部为起点等待新增内容，会跳过已经存在的积压。0-0 适合首次从保留历史开始读取，但不是每次重连的固定参数，否则会反复读历史。

该例使用普通 XREAD，不涉及消费者组和确认机制。需要任务分摊与待处理消息管理时可以另行采用消费者组，但仍要设计确认、超时接管和业务幂等，不能仅替换命令就声称实现端到端恰好一次。

处理重复执行与历史保留

先保存游标再做业务，会在中途崩溃时跳过尚未完成的事件；先做业务再保存游标，则可能在崩溃恢复后重复执行。实际处理应把业务事件去重记录与业务变更放到同一事务中，再更新可恢复进度，或设计等价的可靠提交协议。外部接口副作用还需要稳定幂等键和结果查询能力。

示例中的 event_id 去重不能只存在进程内存里，重启后仍应有效；去重记录的期限也应覆盖可能重放的事件范围。遇到一条失败事件，不要直接把全局游标推进到它后面的成功条目，除非已经为它保存了可靠的单独恢复记录。

Streams 可保留记录，但保留期、裁剪和 Redis 持久化策略共同决定可恢复范围。离线时间超过历史保留窗口，或故障导致尚未可靠持久化的数据丢失，都不能凭消费游标自动恢复。规划最大离线时间、积压容量和归档方案，并监控最旧未处理事件的年龄。

排查重连后的常见误区

打开 AOF 就能让 Pub/Sub 补发吗？

不能。Redis 的数据持久化不把普通频道变成带消费游标的历史日志；需要补发能力就必须事先采用相应存储模型。

重连成功为什么仍没有新消息？

确认订阅是否重新建立、频道名称与环境前缀是否一致、回调是否阻塞。普通 Pub/Sub 不按 Redis 数据库编号隔离频道，不能把 SELECT 当作频道环境隔离方案。

只在消息中加递增编号能找回丢失内容吗？

编号可以帮助发现缺口，但找回内容还需要保存原始事件的来源；只有编号而没有历史内容，依然无法回放。

恢复后该展示“已连接”还是“已同步”？

分别记录连接、订阅和数据同步状态。只有快照或回放达到业务定义的同步点，才把数据状态更新为已同步。

Streams 官方文档：https://redis.io/docs/latest/develop/data-types/streams/。XREAD 参数：https://redis.io/docs/latest/commands/xread/。