Redis Pub/Sub重连后恢复订阅关系的实现方法

Redis Pub/Sub 重连的核心不是“找回旧连接”，而是保存订阅意图，在新连接建立后重新发送 SUBSCRIBE，并等到订阅回执后再继续消费。需要先接受一个边界：Pub/Sub 采用 at-most-once 语义，客户端断线期间错过的消息不会自动补发。

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

如果业务只需要瞬时通知，重连时恢复频道即可；如果消息必须可追溯、可重放，就不要用 Pub/Sub 充当可靠队列，应改评估 Redis Streams。

本文结论：把频道集合当作长期配置，把连接当作可替换资源；重连流程按“建立连接—恢复订阅—确认回执—开始消费”执行，并用退避控制重试。

• 适用：缓存刷新、在线状态、轻量广播等瞬时消息。
• 不适用：订单事件、任务队列、审计日志等要求补偿的消息。
• 判断 ready 的依据：目标频道收到 subscribe 回执，而不是 TCP 连接刚建立。

先划清 Pub/Sub 的消息边界

SUBSCRIBE 成功后，Redis 会把发布到频道的消息推送给当前订阅者。网络断开时，服务端不会为普通 Pub/Sub 订阅保存一份等待客户端补领的历史队列，所以重连只能恢复“以后接收什么”，不能恢复“刚才漏了什么”。

因此，重连组件应把两个动作分开：一是恢复频道关系，二是处理断线窗口。如果业务能接受窗口丢失，记录日志并继续即可；如果不能接受，就在消息模型层改用 Streams，而不是继续堆重试次数。

保存订阅意图而不是依赖旧连接

不要只把频道写在一次连接的初始化代码里。维护一个受锁保护的频道集合，新增或取消订阅时先修改集合，再由当前连接发送命令。连接失败后集合仍然存在，新连接使用快照恢复，这样不会因为旧连接对象被关闭而丢掉订阅配置。

实践中还要避免并发重连：同一订阅器只允许一个 goroutine 负责建立连接，其他读取循环发现错误后退出并交给重连循环处理。

实现带退避的重连循环

下面是 Go 风格的最小骨架。Client、Dial 和 ReadMessage 代表具体客户端库的适配层，重点在流程顺序和资源清理，不绑定某一个第三方包。

type Subscriber struct {
    mu       sync.RWMutex
    channels map[string]struct{}
}

func (s *Subscriber) run(ctx context.Context) error {
    delay := time.Second
    for {
        // 每次循环只维护一个连接，避免旧读取协程与新连接并行消费。
        conn, err := dial(ctx)
        if err != nil {
            if ctx.Err() != nil { return ctx.Err() }
            time.Sleep(delay)
            if delay 

生产代码应把订阅失败作为本轮连接失败处理，不能在半恢复状态下继续消费。退避上限、最大运行时长和关闭信号也应可配置，避免 Redis 不可用时形成紧密重试风暴。

用订阅回执确认恢复完成

Redis 的订阅回复包含消息类型、频道名和当前订阅数量。收到目标频道的 subscribe 回执后，客户端才知道服务端已经建立了对应关系。实现时可维护一个待确认集合：发送订阅时加入集合，收到回执时删除，集合为空才把连接标记为 ready。

取消订阅同样要更新本地意图。若同时使用频道订阅和模式订阅，要分别保存两类配置；同一条消息可能因同时匹配频道和模式而收到多次，消费端需要用业务事件 ID 做幂等，而不能把重复交付误判成重连故障。

按可靠性要求选择 Pub/Sub 或 Streams

可以用下面的判断快速选型：

重连恢复订阅并不等于恢复业务数据。把这两个目标分别建模，才能避免“连接看起来恢复了，但业务事件已经丢失”的误判。

常见问题

重连后需要重新执行 SUBSCRIBE 吗？

需要。订阅关系属于连接上下文，新的连接应根据本地保存的频道和模式集合重新发送订阅，并等待回执。

增加重试次数能找回断线消息吗？

不能。重试只能提高重新建立连接的机会，不能改变 Pub/Sub 的 at-most-once 语义；需要补发时应更换消息模型。