Go Paho MQTT Go 怎样在重连回调中恢复业务订阅

Go 程序使用 Paho MQTT 客户端时，最容易被忽略的场景不是第一次连接，而是网络短暂中断后客户端重新连上了，业务订阅却没有按预期恢复。处理这类问题的关键是：把“连接成功后的业务初始化”统一放进 OnConnectHandler，让首次连接和自动重连走同一份订阅代码。

官方地址：https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang

本文使用 MQTT 3.1.1 客户端 API 说明这一套写法。SetOnConnectHandler 的回调既会在初次连接时触发，也会在自动重连时触发；SetReconnectingHandler 则发生在客户端准备尝试重连之前，两者职责不要混在一起。

先分清重连前后的两个回调

一个实际场景是：服务启动后订阅 device/+/state 和 device/+/alarm，网络断开几秒，Paho 自动重连成功，但业务没有再收到消息。原因通常不是回调没有触发，而是订阅动作只写在了启动流程里。

ReconnectingHandler 适合写“即将重连”的记录、指标或状态切换；它不是连接已经建立的通知。恢复订阅应该放在 OnConnectHandler，因为这个回调代表客户端从未连接或已断开状态变成已连接状态。

可以先把职责写成一句判断：重连前做观测，连接成功后做业务初始化。这样即使连接过程经历多次失败，真正的订阅入口仍然只有一个。

把业务订阅集中成一个函数

订阅主题最好集中保存，而不是在 main 里散落多次调用。Paho 提供 SubscribeMultiple，可以把主题过滤器和 QoS 放进一个 map，统一提交给 Broker。

package main

import (
    "log"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

// restoreSubscriptions 只保存业务订阅声明，首次连接和重连都复用它。
func restoreSubscriptions(client mqtt.Client) error {
    filters := map[string]byte{
        "device/+/state": 1, // 状态消息使用 QoS 1，允许至少一次送达。
        "device/+/alarm": 1, // 告警主题沿用同一个消息处理器。
    }

    token := client.SubscribeMultiple(filters, func(_ mqtt.Client, msg mqtt.Message) {
        // 回调只做轻量分发，耗时业务应交给独立的 goroutine 或队列。
        log.Printf("topic=%s payload=%s", msg.Topic(), string(msg.Payload()))
    })
    if token.Wait() && token.Error() != nil {
        // token 是异步操作结果，必须把 Broker 返回的错误传给上层。
        return token.Error()
    }
    return nil
}

这里的重点不是把所有业务逻辑塞进回调，而是让“订阅哪些主题”成为可重复调用的声明。后续如果需要给不同主题分配不同处理器，可以改成多个 Subscribe 调用，仍然由同一个恢复函数统一管理。

在 OnConnect 中恢复订阅

接下来把恢复函数安装到客户端选项中。安装后，第一次调用 Connect 成功和自动重连成功都会进入同一个入口。

func newClient() mqtt.Client {
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://broker.example:1883") // 示例地址只表示 Broker 配置位置。
    opts.SetClientID("telemetry-service")       // 客户端 ID 应由业务保证稳定且唯一。
    opts.SetAutoReconnect(true)                  // 让客户端在连接丢失后自动尝试恢复。

    opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
        // 这里只记录“准备重连”，不要在未连接时提交业务订阅。
        log.Println("mqtt connection is being retried")
    })
    opts.SetOnConnectHandler(func(client mqtt.Client) {
        // 首次连接和自动重连都走这里，订阅逻辑不再分叉。
        if err := restoreSubscriptions(client); err != nil {
            log.Printf("restore subscriptions failed: %v", err)
        }
    })

    return mqtt.NewClient(opts)
}

这个回调的错误处理要保留。Paho 的订阅方法返回 token，调用 Wait 后再检查 Error，才能知道提交是否被 Broker 接受。生产代码还可以把失败次数、主题集合和客户端 ID 写入结构化日志，便于区分“连接成功但订阅失败”。

CleanSession 与 ResumeSubs 不能替代业务恢复

看到 SetResumeSubs 后，很多人会以为它可以自动恢复所有业务订阅。实际要区分两件事：客户端内部保存的订阅消息恢复，以及应用层想要的业务订阅声明。

SetResumeSubs 针对的是存储中的订阅或取消订阅消息；文档还说明，它在连接时有效，但在 CleanSession 为 false 的自动重连场景下并不是重新执行所有业务订阅的替代品。业务真正依赖哪些主题，仍应由应用自己的订阅声明负责。

因此，更稳妥的判断方式如下：

避免重复订阅和阻塞回调

重连回调可能被多次触发，所以恢复函数要保持幂等思路：主题集合来自固定配置，订阅代码不依赖上一次回调留下的临时变量，也不要在回调中无限追加主题。调用成功后记录一次“订阅集合版本”，遇到重复触发时仍然提交同一份声明即可。

另一个边界是消息处理器。Paho 文档提醒，默认顺序处理模式下，消息回调不应阻塞，也不应直接调用可能阻塞的客户端操作；耗时任务应投递到队列或新 goroutine。否则看似是重连订阅问题，实际可能是消息处理器阻塞导致连接保活异常。

func handleMessage(jobs chan

如果不同主题需要不同处理器，可以在恢复函数中分别注册；如果共用处理器，则至少把主题名传入下游，避免重连后所有消息都被当成同一种事件。

一份可落地的采用清单

• 订阅声明集中在一个函数或配置对象中，首次连接和自动重连不写两套逻辑。
• ReconnectingHandler 只做断线前后的观测，OnConnectHandler 负责连接成功后的初始化。
• 每次订阅都检查 token 错误，把“已连接”和“已成功订阅”分开记录。
• 根据 CleanSession、Broker 会话策略和 Store 配置判断恢复语义，不把 ResumeSubs 当成万能开关。
• 消息回调保持轻量，耗时处理交给队列或独立 goroutine，并明确队列满时的处理方式。

如果你的目标只是保证业务主题在每次连接成功后都存在，最小可靠方案就是：稳定的主题声明 + OnConnectHandler + token 错误处理。会话持久化可以进一步减少恢复成本，但不能取代应用对业务订阅的明确表达。

相关问题

OnConnectHandler 会只在自动重连时触发吗？

不会。官方文档说明它既用于初次连接，也用于自动重连，所以适合放置两种场景都需要的初始化动作。

为什么已经设置了 ResumeSubs 还要重新声明订阅？

因为 ResumeSubs 面向存储的订阅消息恢复，业务代码仍需要表达自己当前依赖的主题集合，尤其是在会话被清理或业务订阅配置发生变化时。

订阅 token 一定要 Wait 吗？

不一定要同步等待，但必须在异步完成时读取错误。示例用 Wait 是为了让恢复函数在记录成功状态前得到明确结果。