package mqtt import ( "errors" mqtt "github.com/eclipse/paho.mqtt.golang" "laic-backend/common" "laic-backend/logger" ) var ( client mqtt.Client subTopics = map[string]byte{} subHandler mqtt.MessageHandler ) // InitMQTT 初始化 MQTT 客户端(不依赖 service,订阅由上层注册) func InitMQTT(config *common.MQTT) { if config == nil || config.Broker == "" { logger.WARN("MQTT broker 为空,跳过初始化") return } opts := mqtt.NewClientOptions(). AddBroker(config.Broker). SetClientID(config.ClientId). SetUsername(config.Username). SetPassword(config.Password). SetAutoReconnect(true). SetCleanSession(true). SetOnConnectHandler(func(c mqtt.Client) { logger.INFO("MQTT 已连接,重新订阅") resubscribe(c) }) client = mqtt.NewClient(opts) if token := client.Connect(); token.Wait() && token.Error() != nil { logger.ERROR("MQTT 连接失败", token.Error()) return } logger.INFO("MQTT ready") } // Subscribe 记录订阅关系并立即订阅;连接重建时自动重订阅 func Subscribe(topics map[string]byte, handler mqtt.MessageHandler) { subHandler = handler for topic, qos := range topics { subTopics[topic] = qos } if client != nil && client.IsConnected() { resubscribe(client) } } func resubscribe(c mqtt.Client) { for topic, qos := range subTopics { if token := c.Subscribe(topic, qos, inboundHandler); token.Wait() && token.Error() != nil { logger.ERROR("订阅失败 topic="+topic, token.Error()) } } } func inboundHandler(c mqtt.Client, msg mqtt.Message) { //logger.INFO("MQTT 收到消息", // "topic="+msg.Topic(), // "qos=", msg.Qos(), // "retained=", msg.Retained(), // "payload="+string(msg.Payload()), //) if subHandler != nil { subHandler(c, msg) } } // Publish 发布消息到指定 topic func Publish(topic string, qos byte, retained bool, payload []byte) error { if client == nil || !client.IsConnected() { return errors.New("mqtt client not connected") } logger.INFO("MQTT 发送消息", "topic="+topic, "qos=", qos, "retained=", retained, "payload="+string(payload), ) token := client.Publish(topic, qos, retained, payload) token.Wait() return token.Error() }