本文详解如何使用 Eclipse Paho Go 客户端检测 MQTT 连接丢失、自动重连，并确保断线重连后能正确恢复订阅，避免消息丢失。
本文详解如何使用 Eclipse Paho Go 客户端检测 MQTT 连接丢失、自动重连，并确保断线重连后能正确恢复订阅，避免消息丢失。
在基于 github.com/eclipse/paho.mqtt.golang 构建的 MQTT 订阅客户端中，一个常见误区是：将 Subscribe() 调用放在 Connect() 之后的主流程中，而非连接建立后的回调里。这会导致客户端虽具备自动重连能力（默认开启 AutoReconnect = true），但重连成功后未重新订阅主题，从而无法接收新消息——即使 Broker 已恢复、发布端持续发消息，订阅端也“静默失联”。
根本原因在于：MQTT 协议本身不维护客户端的订阅状态；连接断开后，Broker 会清理该 ClientID 对应的订阅关系（除非启用持久会话且 CleanSession = false 并正确配置 SessionExpiryInterval）。因此，每次成功连接（包括重连）后，都必须显式重新订阅所需主题。
✅ 正确做法：利用连接生命周期钩子
Paho Go 提供了两个关键回调函数：
-
SetConnectionLostHandler()：连接意外中断时触发，可用于记录日志、告警或清理资源； -
OnConnect（函数字段）：每次成功建立连接（含首次连接和自动重连）后立即调用，是执行订阅逻辑的唯一可靠位置。
同时，需注意：
-
AutoReconnect默认为true，无需手动开启； -
CleanSession: false（即持久会话）仅对 QoS 1/2 消息的服务端排队有效，不能替代客户端主动重订阅； - 若需接收断连期间发布的消息，还需配合
QoS > 0+Retained消息或 Broker 级别消息持久化（如 MQTT 5.0 的Session Expiry Interval）。
✅ 完整可运行示例
package main
import (
"fmt"
"os"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func messageHandler(c mqtt.Client, msg mqtt.Message) {
fmt.Printf("✅ TOPIC: %s\n", msg.Topic())
fmt.Printf("✅ PAYLOAD: %s\n", msg.Payload())
}
func connLostHandler(c mqtt.Client, err error) {
fmt.Printf("⚠️ Connection lost: %v\n", err)
// 可在此添加重试计数、熔断、通知等逻辑
}
func main() {
opts := mqtt.NewClientOptions().
AddBroker("tcp://localhost:1883").
SetClientID("group-one").
SetDefaultPublishHandler(messageHandler).
SetConnectionLostHandler(connLostHandler).
// 关键：所有订阅逻辑移入 OnConnect 回调
OnConnect: func(c mqtt.Client) {
fmt.Println("? Connected successfully. Subscribing to 'test/topic'...")
if token := c.Subscribe("test/topic", 0, nil); token.Wait() && token.Error() != nil {
fmt.Printf("❌ Subscription failed: %v\n", token.Error())
os.Exit(1)
}
fmt.Println("✅ Subscribed to 'test/topic'")
}
client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
panic(fmt.Sprintf("❌ Initial connect failed: %v", token.Error()))
}
// 防止主 goroutine 退出；实际项目中建议用 context 控制生命周期
select {}
}⚠️ 注意事项与最佳实践
-
永远不要在
Connect()后直接Subscribe()：该调用只对首次连接生效，重连后无效； -
OnConnect是订阅的黄金位置：它保证每次连接就绪后立即重建订阅； -
日志与可观测性：务必实现
SetConnectionLostHandler，便于定位网络波动、Broker 故障等问题； -
优雅关闭：程序退出前调用
client.Disconnect(250)，确保清理资源； -
超时控制：生产环境建议为
token.Wait()设置超时（如token.WaitTimeout(5 * time.Second)），避免阻塞。
通过以上改造，你的 MQTT 客户端即可稳健应对网络抖动、Broker 重启等场景，在断线后自动恢复连接与订阅，真正实现高可用消息消费。
