使用 Eclipse Paho Go 库时，无法通过 client ID 从服务端查询已有订阅；重复创建客户端会导致订阅丢失，正确做法是在连接成功后通过 OnConnect 回调主动重订阅。
使用 eclipse paho go 库时，无法通过 client id 从服务端查询已有订阅；重复创建客户端会导致订阅丢失，正确做法是在连接成功后通过 `onconnect` 回调主动重订阅。
MQTT 协议本身不提供“查询当前客户端已订阅主题”的机制——服务端不会向客户端暴露其历史订阅列表，也不会在重建连接时主动同步订阅关系。当你以相同 clientID 重新连接（且设置了 CleanSession=false），Broker 确实会保留该 client ID 对应的会话状态（包括未确认的 QoS 1/2 消息、遗嘱消息等），但不会自动恢复订阅。Paho Go 客户端自身也不维护订阅元数据，所有 Subscribe() 调用仅影响本地消息路由逻辑：只有显式调用 Subscribe() 并注册 MessageHandler 后，收到匹配主题的消息才会被分发到对应处理器；否则，消息将落入默认处理器（如未设置则被静默丢弃）。
因此，关键在于：连接建立后，必须主动重新执行 Subscribe()。推荐方式是利用 OnConnect 回调，在连接成功瞬间完成订阅：
func CreateMQTTClient(clientID string, topics map[string]byte) (client MQTT.Client) {
username := viper.GetString("messaging.rabbitmq.username")
password := viper.GetString("messaging.rabbitmq.password")
host := viper.GetString("messaging.rabbitmq.host")
mqttPort := viper.GetString("messaging.rabbitmq.mqqtPort")
mqttURL := "tcp://" + host + ":" + mqttPort
opts := MQTT.NewClientOptions().
AddBroker(mqttURL).
SetClientID(clientID).
SetUsername(username).
SetPassword(password).
SetCleanSession(false).
SetOnConnectHandler(func(c MQTT.Client) {
// 连接成功后立即重订阅
for topic, qos := range topics {
if token := c.Subscribe(topic, qos, nil); token.Wait() && token.Error() != nil {
log.Printf("Failed to subscribe to %s: %v", topic, token.Error())
} else {
log.Printf("Subscribed to %s with QoS %d", topic, qos)
}
}
})
cli := MQTT.NewClient(opts)
if !cli.IsConnected() {
log.Println("Connecting MQTT client:", clientID)
if token := cli.Connect(); token.Wait() && token.Error() != nil {
log.Printf("MQTT connection failed for %s: %v", clientID, token.Error())
}
}
return cli
}⚠️ 注意事项：
-
topics参数应为预定义的静态主题集合（如map[string]byte{"sensor/+/temp": 1, "cmd/#": 2}），动态生成的主题需在业务层额外管理并传入； -
SetCleanSession(false)是前提，否则服务端会清除会话状态，QoS 1/2 消息可能丢失； -
OnConnectHandler仅在首次连接或重连成功时触发，确保每次连接都执行订阅； - 若需支持运行时动态增删订阅，应封装
Subscribe/Unsubscribe方法，并维护本地订阅映射表，避免重复订阅。
总结：MQTT 的“会话恢复”是服务端行为，而“订阅恢复”是客户端责任。不要依赖服务端记忆订阅，而应将订阅逻辑视为连接后的必需初始化步骤——这是符合 MQTT 规范且被主流 SDK 推荐的最佳实践。
