@author liuqing

@commit mqtt整合

cloud/ipolesvr/mqttclient.go

@@ -1,141 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientId, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientId,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		err := o.Subscribe(k, v)
-		if err != nil {
-			return
-		}
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		err := o.PublishString(topic, str, 0)
-		if err != nil {
-			return
-		}
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Connect(ctx)
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

cloud/ipolesvr/mqttmgr.go

@@ -17,18 +17,17 @@ func GetMQTTMgr() *MQTTMgr {
 }
 
 type MQTTMgr struct {
-	Cloud *MqttClient
+	Cloud *mqtt.MqttClient
 }
 
 func _newMQTTMgr() *MQTTMgr {
 	return &MQTTMgr{
-		Cloud: NewMqttClient(MQTTConfig.Mqtt_Server,
+		Cloud: mqtt.NewMqttClient(MQTTConfig.Mqtt_Server,
 			MQTTConfig.Mqtt_Id,
 			MQTTConfig.Mqtt_User,
 			MQTTConfig.Mqtt_Password,
-			3000, &EmptyMqttOnline{}),
+			3000, &mqtt.EmptyMqttOnline{}),
 	}
-
 }
 
 func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler) {

cloud/ipolesvr/mqttmgr_hl.go

@@ -18,18 +18,17 @@ func GetHlMqttMgr() *HlMqttMgr {
 }
 
 type HlMqttMgr struct {
-	Cloud *MqttClient
+	Cloud *mqtt.MqttClient
 }
 
 func _newHlMqttMgr() *HlMqttMgr {
 	return &HlMqttMgr{
-		Cloud: NewMqttClient(MqttconfigHl.Mqtt_Server,
+		Cloud: mqtt.NewMqttClient(MqttconfigHl.Mqtt_Server,
 			MqttconfigHl.Mqtt_Id,
 			MqttconfigHl.Mqtt_User,
 			MqttconfigHl.Mqtt_Password,
-			3000, &EmptyMqttOnline{}),
+			3000, &mqtt.EmptyMqttOnline{}),
 	}
-
 }
 
 func (o *HlMqttMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler) {

cloud/ipolesvr/ymlampcontrollermgr.go

@@ -41,10 +41,10 @@ type YMLampControllerMgr struct {
 }
 
 func (o *YMLampControllerMgr) SubscribeTopics() {
-	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_DATA), mqtt.AtMostOnce, o.HandlerData)
-	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_ALARM), mqtt.AtMostOnce, o.HandlerData)
-	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_SET_SWITCH_ACK), mqtt.AtMostOnce, o.HandlerData)
-	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_SET_ONOFFTIME_ACK), mqtt.AtMostOnce, o.HandlerData)
+	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_DATA), mqtt.AtLeastOnce, o.HandlerData)
+	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_ALARM), mqtt.AtLeastOnce, o.HandlerData)
+	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_SET_SWITCH_ACK), mqtt.AtLeastOnce, o.HandlerData)
+	GetMQTTMgr().Subscribe(GetVagueTopic(protocol.DT_LAMPCONTROLLER, protocol.TP_YM_SET_ONOFFTIME_ACK), mqtt.AtLeastOnce, o.HandlerData)
 }
 
 func (o *YMLampControllerMgr) HandlerData(m mqtt.Message) {
@@ -114,6 +114,7 @@ func (o *YMLampControllerMgr) handleQueue() uint32 {
 
 	Tenant, _, DID, topic, err := ParseTopic(m.Topic())
 	if err != nil {
+		logrus.Errorf("YMLampControllerMgr.handleQueue:ParseTopic失败,topic=%s,err=%v", m.Topic(), err)
 		return quantity
 	}
 	pymlc, ok := o.mapYMLampController[DID]
@@ -129,6 +130,8 @@ func (o *YMLampControllerMgr) handleQueue() uint32 {
 		o.handleALARM(pymlc, m)
 	case protocol.TP_YM_SET_SWITCH_ACK, protocol.TP_YM_SET_ONOFFTIME_ACK:
 		o.handleACK(m)
+	default:
+		logrus.Errorf("YMLampControllerMgr.handleQueue:未知topic=%s,全topic=%s", topic, m.Topic())
 	}
 	return quantity
 }
@@ -136,6 +139,7 @@ func (o *YMLampControllerMgr) handleQueue() uint32 {
 func (o *YMLampControllerMgr) handleDATA(lp *YMLampController, m *mqtt.Message) {
 	var obj protocol.Pack_CHZB_UploadData
 	if err := obj.DeCode(m.PayloadString()); err != nil {
+		logrus.Errorf("YMLampControllerMgr.handleDATA:DeCode失败,DID=%s,topic=%s,err=%v", lp.did, m.Topic(), err)
 		return
 	}
 	t, err := util.MlParseTime(obj.Time)
@@ -151,6 +155,7 @@ func (o *YMLampControllerMgr) handleDATA(lp *YMLampController, m *mqtt.Message)
 func (o *YMLampControllerMgr) handleALARM(lp *YMLampController, m *mqtt.Message) {
 	var obj protocol.Pack_CHZB_LampAlarm
 	if err := obj.DeCode(m.PayloadString()); err != nil {
+		logrus.Errorf("YMLampControllerMgr.handleALARM:DeCode失败,DID=%s,topic=%s,err=%v", lp.did, m.Topic(), err)
 		return
 	}
 	lp.HandleAlarm(obj.Data)
@@ -159,6 +164,7 @@ func (o *YMLampControllerMgr) handleALARM(lp *YMLampController, m *mqtt.Message)
 func (o *YMLampControllerMgr) handleACK(m *mqtt.Message) {
 	var obj protocol.Pack_Ack
 	if err := obj.DeCode(m.PayloadString()); err != nil {
+		logrus.Errorf("YMLampControllerMgr.handleACK:DeCode失败,topic=%s,err=%v", m.Topic(), err)
 		return
 	}
 	oo := models.DeviceCmdRecord{ID: obj.Seq, State: 1, Resp: obj.Data.Error}

cloud/mqttforward/mqttclient.go

@@ -1,141 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"lc/common/mqtt"
-
-	"github.com/sirupsen/logrus"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientid,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		err := o.Subscribe(k, v)
-		if err != nil {
-			return
-		}
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		err := o.PublishString(topic, str, 0)
-		if err != nil {
-			return
-		}
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Connect(ctx)
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

cloud/mqttforward/mqttmgr.go

@@ -16,22 +16,22 @@ func GetMQTTMgr() *MQTTMgr {
 }
 
 type MQTTMgr struct {
-	FromCloud *MqttClient
-	ToCloud   *MqttClient
+	FromCloud *mqtt.MqttClient
+	ToCloud   *mqtt.MqttClient
 }
 
 func _newMQTTMgr() *MQTTMgr {
 	return &MQTTMgr{
-		FromCloud: NewMqttClient(FromMQTTConfig.Mqtt_Server,
+		FromCloud: mqtt.NewMqttClient(FromMQTTConfig.Mqtt_Server,
 			FromMQTTConfig.Mqtt_Id,
 			FromMQTTConfig.Mqtt_User,
 			FromMQTTConfig.Mqtt_Password,
-			3000, &EmptyMqttOnline{}),
-		ToCloud: NewMqttClient(ToMQTTConfig.Mqtt_Server,
+			3000, &mqtt.EmptyMqttOnline{}),
+		ToCloud: mqtt.NewMqttClient(ToMQTTConfig.Mqtt_Server,
 			ToMQTTConfig.Mqtt_Id,
 			ToMQTTConfig.Mqtt_User,
 			ToMQTTConfig.Mqtt_Password,
-			3000, &EmptyMqttOnline{}),
+			3000, &mqtt.EmptyMqttOnline{}),
 	}
 }
 

cloud/websvr/controllers/clampcontroller.go

@@ -714,6 +714,8 @@ func (o *LampController) Switch() {
 				continue
 			}
 			topic := GetTopic(obj.Tenant, protocol.DT_LAMPCONTROLLER, v.ID, protocol.TP_YM_SET_SWITCH)
+			beego.Info(fmt.Sprintf("YM Switch 发布: topic=%s, payload=%s, SwitchOn=%d, Brightness=%d",
+				topic, str, obj.SwitchOn, obj.Brightness))
 			err = GetMqttHandler().PublishString(topic, str, mqtt.AtLeastOnce)
 			if err != nil {
 				beego.Error(fmt.Sprintf("LampController.Switch:消息发布错误:%s", err.Error()))
@@ -839,16 +841,17 @@ func (o *LampController) Switch() {
 }
 
 // GetTopic 下发的控制命令
-//topic格式：{tenant}/{GID}/{did}/{TP}
-//func GetCtrlTopic(tenant, gid, did, tp string) string {
-//	return tenant + "/" + gid + "/" + did + "/" + tp
-//}
+// topic格式：{tenant}/{GID}/{did}/{TP}
+//
+//	func GetCtrlTopic(tenant, gid, did, tp string) string {
+//		return tenant + "/" + gid + "/" + did + "/" + tp
+//	}
 func GetTopic(tenant, devtype, id, topic string) string {
 	return tenant + "/" + devtype + "/" + id + "/" + topic
 }
 
 // GethltopicDown 下发的控制命令先固定,再从配置文件读
-//topic格式：hllk/113/lamp/longchi/down
+// topic格式：hllk/113/lamp/longchi/down
 func GethltopicDown(tenant, devtype, id, topic string) string {
 	return "hllk/113/lamp/longchi/down"
 }

edge/its/mqttclient.go

@@ -1,4 +1,4 @@
-package main
+package mqtt
 
 import (
 	"context"
@@ -7,77 +7,80 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
 )
 
+// BaseMqttOnline 上线消息和遗嘱消息接口
 type BaseMqttOnline interface {
 	GetOnlineMsg() (string, string)
 	GetWillMsg() (string, string)
 }
 
-type EmptyMqttOnline struct {
-}
+// EmptyMqttOnline 空实现：不上线、无遗嘱
+type EmptyMqttOnline struct{}
 
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
+func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) { return "", "" }
+func (o *EmptyMqttOnline) GetWillMsg() (string, string)   { return "", "" }
 
+// MqttClient 封装 common/mqtt.Client，提供超时控制、主题记忆、断线自动重订阅
 type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
+	mqtt       *Client
+	mu         sync.Mutex
+	mapTopics  map[string]QOS
+	timeout    uint
+	MqttOnline BaseMqttOnline
 }
 
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
+// NewMqttClient 创建并连接 MQTT 客户端
+func NewMqttClient(server, clientID, user, password string, timeout uint, online BaseMqttOnline) *MqttClient {
+	o := &MqttClient{
+		mapTopics:  make(map[string]QOS),
 		timeout:    timeout,
-		MqttOnline: mqttOnline,
+		MqttOnline: online,
 	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
+	client, err := NewClient(ClientOptions{
 		Servers:       []string{server},
-		ClientID:      clientid,
+		ClientID:      clientID,
 		Username:      user,
 		Password:      password,
 		AutoReconnect: true,
-	}, &o)
+	}, o)
 	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:", err.Error()))
-		return nil
+		panic(fmt.Errorf("MQTT错误: %w", err))
 	}
 	o.mqtt = client
 	ctx, cancel := o.Ctx()
 	defer cancel()
-	err = client.Connect(ctx)
-	return &o
+	_ = client.Connect(ctx)
+	return o
 }
 
+// ---- 实现 ConnHandler 接口 ----
+
 func (o *MqttClient) ConnectionLostHandler(err error) {
 	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
 }
 
 func (o *MqttClient) OnConnectHandler() {
 	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
 	for k, v := range o.mapTopics {
-		o.Subscribe(k, v)
+		if err := o.Subscribe(k, v); err != nil {
+			logrus.Errorf("重订阅主题 %s 失败: %v", k, err)
+		}
 	}
 	topic, str := o.MqttOnline.GetOnlineMsg()
 	if topic != "" {
-		o.PublishString(topic, str, 0)
+		if err := o.PublishString(topic, str, 0); err != nil {
+			logrus.Errorf("发布上线消息失败: %v", err)
+		}
 	}
 }
 
-func (o *MqttClient) GetWill() (topic string, payload string) {
+func (o *MqttClient) GetWill() (string, string) {
 	return o.MqttOnline.GetWillMsg()
 }
 
+// ---- 操作代理 ----
+
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
 		ctx, cancel := o.Ctx()
@@ -87,32 +90,32 @@ func (o *MqttClient) Connect() error {
 	return nil
 }
 
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
+func (o *MqttClient) IsConnected() bool { return o.mqtt.IsConnected() }
 
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
+func (o *MqttClient) Publish(topic string, payload []byte, qos QOS) error {
 	ctx, cancel := o.Ctx()
 	defer cancel()
 	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
+
+func (o *MqttClient) PublishString(topic string, payload string, qos QOS) error {
 	ctx, cancel := o.Ctx()
 	defer cancel()
 	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
+
+func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos QOS) error {
 	ctx, cancel := o.Ctx()
 	defer cancel()
 	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
+func (o *MqttClient) Subscribe(topic string, qos QOS) error {
 	o.mu.Lock()
-	defer o.mu.Unlock()
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
+	o.mu.Unlock()
 	ctx, cancel := o.Ctx()
 	defer cancel()
 	return o.mqtt.Subscribe(ctx, topic, qos)
@@ -120,16 +123,16 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 
 func (o *MqttClient) Unsubscribe(topic string) error {
 	o.mu.Lock()
-	defer o.mu.Unlock()
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
+	o.mu.Unlock()
 	ctx, cancel := o.Ctx()
 	defer cancel()
 	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
+func (o *MqttClient) Handle(topic string, handler MessageHandler) Route {
 	return o.mqtt.Handle(topic, handler)
 }
 

common/mqtt/mgr.go

@@ -0,0 +1,199 @@
+package mqtt
+
+import (
+	"runtime/debug"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"lc/common/util"
+)
+
+// OptType 消息发布/订阅方向
+type OptType uint8
+
+const (
+	ToAll   OptType = 0 // 同时发布到云端和边缘端
+	ToCloud OptType = 1 // 仅云端
+	ToEdge  OptType = 2 // 仅边缘端
+)
+
+// MQTTMessage 队列中的消息
+type MQTTMessage struct {
+	topic   string
+	payload string
+	qos     QOS
+	tp      OptType
+}
+
+// RestartFunc 协程 panic 后重建回调
+type RestartFunc func(fn func(args ...interface{}) interface{}, args ...interface{})
+
+// MQTTMgr 管理 Cloud/Edge 双客户端，支持异步队列发布
+type MQTTMgr struct {
+	Cloud     *MqttClient
+	Edge      *MqttClient
+	Queue     *util.MlQueue
+	restartFn RestartFunc
+}
+
+// MQTTMgrConfig 管理器构造参数
+type MQTTMgrConfig struct {
+	CloudServer   string
+	CloudClientID string
+	CloudUser     string
+	CloudPassword string
+	CloudTimeout  uint
+	CloudOnline   BaseMqttOnline // nil 则使用 EmptyMqttOnline
+
+	EdgeServer   string
+	EdgeClientID string
+	EdgeUser     string
+	EdgePassword string
+	EdgeTimeout  uint
+	EdgeOnline   BaseMqttOnline // nil 则使用 EmptyMqttOnline
+}
+
+// NewMQTTMgr 创建 MQTT 管理器
+func NewMQTTMgr(cfg MQTTMgrConfig) *MQTTMgr {
+	mgr := &MQTTMgr{
+		Queue: util.NewQueue(2000),
+	}
+	if cfg.CloudOnline == nil {
+		cfg.CloudOnline = &EmptyMqttOnline{}
+	}
+	if cfg.EdgeOnline == nil {
+		cfg.EdgeOnline = &EmptyMqttOnline{}
+	}
+	if cfg.CloudServer != "" {
+		mgr.Cloud = NewMqttClient(cfg.CloudServer, cfg.CloudClientID,
+			cfg.CloudUser, cfg.CloudPassword, cfg.CloudTimeout, cfg.CloudOnline)
+	}
+	if cfg.EdgeServer != "" {
+		mgr.Edge = NewMqttClient(cfg.EdgeServer, cfg.EdgeClientID,
+			cfg.EdgeUser, cfg.EdgePassword, cfg.EdgeTimeout, cfg.EdgeOnline)
+	}
+	return mgr
+}
+
+// SetRestartFn 设置 panic 恢复后的协程重建回调
+func (o *MQTTMgr) SetRestartFn(fn RestartFunc) {
+	o.restartFn = fn
+}
+
+// Subscribe 订阅主题，支持方向选择
+func (o *MQTTMgr) Subscribe(topic string, qos QOS, handler MessageHandler, tp OptType) {
+	switch tp {
+	case ToAll:
+		if o.Cloud != nil {
+			o.Cloud.Handle(topic, handler)
+			_ = o.Cloud.Subscribe(topic, qos)
+		}
+		if o.Edge != nil {
+			o.Edge.Handle(topic, handler)
+			_ = o.Edge.Subscribe(topic, qos)
+		}
+	case ToCloud:
+		if o.Cloud != nil {
+			o.Cloud.Handle(topic, handler)
+			_ = o.Cloud.Subscribe(topic, qos)
+		}
+	case ToEdge:
+		if o.Edge != nil {
+			o.Edge.Handle(topic, handler)
+			_ = o.Edge.Subscribe(topic, qos)
+		}
+	}
+}
+
+// UnSubscribe 退订主题
+func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
+	switch tp {
+	case ToAll:
+		if o.Cloud != nil {
+			_ = o.Cloud.Unsubscribe(topic)
+		}
+		if o.Edge != nil {
+			_ = o.Edge.Unsubscribe(topic)
+		}
+	case ToCloud:
+		if o.Cloud != nil {
+			_ = o.Cloud.Unsubscribe(topic)
+		}
+	case ToEdge:
+		if o.Edge != nil {
+			_ = o.Edge.Unsubscribe(topic)
+		}
+	}
+}
+
+// Publish 异步发布（入队列）
+func (o *MQTTMgr) Publish(topic, payload string, qos QOS, tp OptType) {
+	o.Queue.Put(&MQTTMessage{topic: topic, payload: payload, qos: qos, tp: tp})
+}
+
+// doPublish 底层同步发布
+func (o *MQTTMgr) doPublish(msg *MQTTMessage) error {
+	var err error
+	switch msg.tp {
+	case ToAll:
+		if o.Cloud != nil {
+			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
+		}
+		if o.Edge != nil {
+			_ = o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
+		}
+	case ToCloud:
+		if o.Cloud != nil {
+			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
+		}
+	case ToEdge:
+		if o.Edge != nil {
+			err = o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
+		}
+	}
+	return err
+}
+
+// MQTTConnectMgr 连接保持协程（每 10 秒重连）
+func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
+	for {
+		time.Sleep(10 * time.Second)
+		if o.Cloud != nil {
+			o.Cloud.Connect()
+		}
+		if o.Edge != nil {
+			o.Edge.Connect()
+		}
+	}
+}
+
+// MQTTMessageHandle 队列消费协程（带 RETRY）
+func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
+	defer func() {
+		if err := recover(); err != nil {
+			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
+			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
+			time.Sleep(time.Second)
+			if o.restartFn != nil {
+				o.restartFn(o.MQTTMessageHandle, args)
+			}
+		}
+	}()
+	for {
+		if m, ok, _ := o.Queue.Get(); ok {
+			if msg, ok := m.(*MQTTMessage); ok {
+				for {
+					if err := o.doPublish(msg); err != nil {
+						logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
+						time.Sleep(time.Second)
+						continue
+					}
+					break
+				}
+			}
+		} else {
+			time.Sleep(200 * time.Millisecond)
+		}
+	}
+}

edge/camera/main.go

@@ -58,6 +58,9 @@ func main() {
 		return
 	}
 	gopool = util.NewPool(5)
+	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
+		gopool.Add(fn, args)
+	})
 	gopool.Add(DiscoveryCamera, 1)                //发现摄像头
 	gopool.Add(GetMQTTMgr().MQTTMessageHandle, 2) //发布进程
 	gopool.Add(GetMQTTMgr().MQTTConnectMgr, 3)    //保持连接

edge/camera/mqtt_init.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"sync"
+
+	"lc/common/mqtt"
+)
+
+type OptType = mqtt.OptType
+type MQTTMessage = mqtt.MQTTMessage
+
+const (
+	ToAll   = mqtt.ToAll
+	ToCloud = mqtt.ToCloud
+	ToEdge  = mqtt.ToEdge
+)
+
+var _mqttMgrOnce sync.Once
+var _mqttMgr *mqtt.MQTTMgr
+
+func GetMQTTMgr() *mqtt.MQTTMgr {
+	_mqttMgrOnce.Do(func() {
+		_mqttMgr = mqtt.NewMQTTMgr(mqtt.MQTTMgrConfig{
+			CloudServer:   appConfig.Cloud.Mqtt.Server,
+			CloudClientID: appConfig.GID + "@" + appname + appversion,
+			CloudUser:     appConfig.Cloud.Mqtt.User,
+			CloudPassword: appConfig.Cloud.Mqtt.Password,
+			CloudTimeout:  appConfig.Cloud.Mqtt.Timeout,
+			EdgeServer:    appConfig.Edge.Mqtt.Server,
+			EdgeClientID:  appConfig.GID + "@" + appname + appversion,
+			EdgeUser:      appConfig.Edge.Mqtt.User,
+			EdgePassword:  appConfig.Edge.Mqtt.Password,
+			EdgeTimeout:   appConfig.Edge.Mqtt.Timeout,
+		})
+	})
+	return _mqttMgr
+}

edge/camera/mqttclient.go

@@ -1,138 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientid,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		o.Subscribe(k, v)
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		o.PublishString(topic, str, 0)
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	if !o.mqtt.IsConnected() {
-		ctx, cancel := o.Ctx()
-		defer cancel()
-		return o.mqtt.Connect(ctx)
-	}
-	return nil
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

edge/camera/mqttmgr.go

@@ -1,188 +0,0 @@
-package main
-
-import (
-	"runtime/debug"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-	"lc/common/util"
-)
-
-type OptType uint8
-
-const (
-	ToAll   OptType = 0 //发布和订阅边缘端与云端的消息
-	ToCloud OptType = 1 //发布和订阅云端的消息
-	ToEdge  OptType = 2 //发布和订阅边缘端的消息
-)
-
-var _mqttMgronce sync.Once
-var _mqttMgrsingle *MQTTMgr
-
-// GetMQTTMgr 单态
-func GetMQTTMgr() *MQTTMgr {
-	_mqttMgronce.Do(func() {
-		_mqttMgrsingle = _newMQTTMgr()
-	})
-	return _mqttMgrsingle
-}
-
-type MQTTMgr struct {
-	Cloud *MqttClient
-	Edge  *MqttClient
-	Queue *util.MlQueue
-}
-
-// 建两个client
-func _newMQTTMgr() *MQTTMgr {
-	mgr := &MQTTMgr{
-		Queue: util.NewQueue(2000),
-	}
-	if appConfig.Edge.Mqtt.Server != "" {
-		mgr.Edge = NewMqttClient(appConfig.Edge.Mqtt.Server,
-			appConfig.GID+"@"+appname+appversion,
-			appConfig.Edge.Mqtt.User,
-			appConfig.Edge.Mqtt.Password,
-			appConfig.Edge.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	if appConfig.Cloud.Mqtt.Server != "" {
-		mgr.Cloud = NewMqttClient(appConfig.Cloud.Mqtt.Server,
-			appConfig.GID+"@"+appname+appversion,
-			appConfig.Cloud.Mqtt.User,
-			appConfig.Cloud.Mqtt.Password,
-			appConfig.Cloud.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	return mgr
-}
-
-// Subscribe 定阅
-func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	}
-}
-
-// UnSubscribe 退定
-func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	}
-}
-
-// Publish 发布进队列
-func (o *MQTTMgr) Publish(topic string, payload string, qos mqtt.QOS, tp OptType) {
-	msg := MQTTMessage{
-		topic:   topic,
-		payload: payload,
-		qos:     qos,
-		tp:      tp,
-	}
-	o.Queue.Put(&msg)
-}
-
-// 发布低
-func (o *MQTTMgr) _publish(msg *MQTTMessage) error {
-	var err error
-	switch msg.tp {
-	case ToAll:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	}
-	return err
-}
-
-// MQTTConnectMgr 连接保持
-func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
-	for {
-		time.Sleep(10 * time.Second)
-		if o.Cloud != nil {
-			o.Cloud.Connect()
-		}
-		if o.Edge != nil {
-			o.Edge.Connect()
-		}
-	}
-}
-
-func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
-			time.Sleep(time.Second)
-			gopool.Add(o.MQTTMessageHandle, args)
-		}
-	}()
-	var err error
-	for { //队列中所有发布
-		if m, ok, _ := o.Queue.Get(); ok {
-			if msg, ok := m.(*MQTTMessage); ok {
-			RETRY:
-				err = o._publish(msg)
-				if err != nil {
-					logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
-					time.Sleep(time.Second)
-					goto RETRY
-				}
-			}
-		} else {
-			time.Sleep(200 * time.Millisecond)
-		}
-	}
-}
-
-type MQTTMessage struct {
-	topic   string
-	payload string
-	qos     mqtt.QOS
-	tp      OptType
-}

edge/ipole/devmgr.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/sirupsen/logrus"
@@ -51,8 +52,51 @@ type DeviceMgr struct {
 	mapDevice map[string]Device
 }
 
+func validateDevInfos(devinfos []protocol.DevInfo) error {
+	for i := range devinfos {
+		v := &devinfos[i]
+		if v.DevCode == "" {
+			return fmt.Errorf("DevCode 不能为空")
+		}
+		if v.TID == 0 {
+			return fmt.Errorf("设备[%s]的 TID 不能为 0，请指定正确的物模型ID", v.DevCode)
+		}
+		if v.ProtocolType > 2 {
+			return fmt.Errorf("设备[%s]的 ProtocolType=%d 不支持，仅支持 0(ModbusRTU) 1(Zigbee) 2(裕明485)", v.DevCode, v.ProtocolType)
+		}
+		// 同批次内 DevCode 唯一性
+		for j := i + 1; j < len(devinfos); j++ {
+			if v.DevCode == devinfos[j].DevCode {
+				return fmt.Errorf("同一文件中 DevCode=%s 重复", v.DevCode)
+			}
+		}
+		// 同批次内同一 Code 下 DevID 冲突检测
+		for j := i + 1; j < len(devinfos); j++ {
+			if v.Code == devinfos[j].Code && v.DevID == devinfos[j].DevID {
+				return fmt.Errorf("同一串口[Code=%d]下 DevID=%d 冲突(设备 %s 和 %s)", v.Code, v.DevID, v.DevCode, devinfos[j].DevCode)
+			}
+		}
+	}
+	return nil
+}
+
+func (o *DeviceMgr) checkDevIDConflict(v *protocol.DevInfo) error {
+	for _, exist := range o.mapDevice {
+		ei := exist.GetDevInfo()
+		if ei.Code == v.Code && ei.DevID == v.DevID && ei.DevCode != v.DevCode {
+			return fmt.Errorf("串口[Code=%d]下 DevID=%d 已被设备[%s]占用，设备[%s]不能使用相同地址",
+				v.Code, v.DevID, ei.DevCode, v.DevCode)
+		}
+	}
+	return nil
+}
+
 // AddDevices 新增
 func (o *DeviceMgr) AddDevices(devinfos []protocol.DevInfo) {
+	if err := validateDevInfos(devinfos); err != nil {
+		logrus.Errorf("AddDevices: 配置校验失败: %v", err)
+		return
+	}
 	o.mu.Lock()
 	defer o.mu.Unlock()
 	for _, v := range devinfos {
@@ -60,6 +104,10 @@ func (o *DeviceMgr) AddDevices(devinfos []protocol.DevInfo) {
 			logrus.Errorf("AddDevices:已添加该设备，不可重复添加:devcode=%s", v.DevCode)
 			continue
 		}
+		if err := o.checkDevIDConflict(&v); err != nil {
+			logrus.Errorf("AddDevices: %v", err)
+			continue
+		}
 		di := v
 		if device := CreateDevice(v.ProtocolType, &di); device != nil {
 			device.Start()
@@ -80,6 +128,12 @@ func (o *DeviceMgr) RemoveDevice(devcode string) {
 
 // UpdateDevices 新增或修改
 func (o *DeviceMgr) UpdateDevices(code uint8, devinfos []protocol.DevInfo) {
+	if len(devinfos) > 0 {
+		if err := validateDevInfos(devinfos); err != nil {
+			logrus.Errorf("UpdateDevices: 配置校验失败: %v", err)
+			return
+		}
+	}
 	//先遍历devinfos
 	mapDevinfos := make(map[string]*protocol.DevInfo)
 	for _, v := range devinfos {
@@ -123,6 +177,11 @@ func (o *DeviceMgr) UpdateDevices(code uint8, devinfos []protocol.DevInfo) {
 				device.Stop()
 				delete(o.mapDevice, v.DevCode)
 			}
+		} else {
+			if err := o.checkDevIDConflict(&v); err != nil {
+				logrus.Errorf("UpdateDevices: %v", err)
+				continue
+			}
 		}
 		if device := CreateDevice(v.ProtocolType, &v); device != nil {
 			device.Start()

edge/ipole/main.go

@@ -164,12 +164,18 @@ func main() {
 
 	//创建设备并启动数据处理流程
 	for k := range serialConfig.Serial {
-		if devinfos, err := LoadDev(k); err == nil {
-			GetDeviceMgr().AddDevices(devinfos)
+		devinfos, err := LoadDev(k)
+		if err != nil {
+			logrus.Warnf("加载串口[%d]的设备配置文件dev/%d.json失败: %v ,该串口无设备将被跳过", k, k, err)
+			continue
 		}
+		GetDeviceMgr().AddDevices(devinfos)
 	}
 
 	gopool = util.NewPool(10)
+	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
+		gopool.Add(fn, args)
+	})
 	gopool.Add(SyncTime, 0)
 	// gopool.Add(WatchGoforever, 0)
 	// gopool.Add(SignalProcess, 8)

edge/ipole/mqtt_init.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"sync"
+
+	"lc/common/mqtt"
+)
+
+// 重导出类型和常量，避免修改所有业务文件
+type OptType = mqtt.OptType
+type MQTTMessage = mqtt.MQTTMessage
+
+const (
+	ToAll   = mqtt.ToAll
+	ToCloud = mqtt.ToCloud
+	ToEdge  = mqtt.ToEdge
+)
+
+var _mqttMgrOnce sync.Once
+var _mqttMgr *mqtt.MQTTMgr
+
+func GetMQTTMgr() *mqtt.MQTTMgr {
+	_mqttMgrOnce.Do(func() {
+		_mqttMgr = mqtt.NewMQTTMgr(mqtt.MQTTMgrConfig{
+			CloudServer:   appConfig.Cloud.Mqtt.Server,
+			CloudClientID: appConfig.GID + "@" + appname + version,
+			CloudUser:     appConfig.Cloud.Mqtt.User,
+			CloudPassword: appConfig.Cloud.Mqtt.Password,
+			CloudTimeout:  appConfig.Cloud.Mqtt.Timeout,
+			EdgeServer:    appConfig.Edge.Mqtt.Server,
+			EdgeClientID:  appConfig.GID + "@" + appname + version,
+			EdgeUser:      appConfig.Edge.Mqtt.User,
+			EdgePassword:  appConfig.Edge.Mqtt.Password,
+			EdgeTimeout:   appConfig.Edge.Mqtt.Timeout,
+		})
+	})
+	return _mqttMgr
+}

edge/ipole/mqttclient.go

@@ -1,138 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientid,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintln("MQTT错误:", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		o.Subscribe(k, v)
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		o.PublishString(topic, str, 0)
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	if !o.mqtt.IsConnected() {
-		ctx, cancel := o.Ctx()
-		defer cancel()
-		return o.mqtt.Connect(ctx)
-	}
-	return nil
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

edge/ipole/mqttmgr.go

@@ -1,184 +0,0 @@
-package main
-
-import (
-	"runtime/debug"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-	"lc/common/util"
-)
-
-type OptType uint8
-
-const (
-	ToAll   OptType = 0 //发布和订阅边缘端与云端的消息
-	ToCloud OptType = 1 //发布和订阅云端的消息
-	ToEdge  OptType = 2 //发布和订阅边缘端的消息
-)
-
-var _mqttMgronce sync.Once
-var _mqttMgrsingle *MQTTMgr
-
-func GetMQTTMgr() *MQTTMgr {
-	_mqttMgronce.Do(func() {
-		_mqttMgrsingle = _newMQTTMgr()
-	})
-	return _mqttMgrsingle
-}
-
-type MQTTMgr struct {
-	Cloud *MqttClient
-	Edge  *MqttClient
-	Queue *util.MlQueue
-}
-
-func _newMQTTMgr() *MQTTMgr {
-	mgr := &MQTTMgr{
-		Queue: util.NewQueue(2000),
-	}
-	if appConfig.Edge.Mqtt.Server != "" {
-		mgr.Edge = NewMqttClient(appConfig.Edge.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Edge.Mqtt.User,
-			appConfig.Edge.Mqtt.Password,
-			appConfig.Edge.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	if appConfig.Cloud.Mqtt.Server != "" {
-		mgr.Cloud = NewMqttClient(appConfig.Cloud.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Cloud.Mqtt.User,
-			appConfig.Cloud.Mqtt.Password,
-			appConfig.Cloud.Mqtt.Timeout,
-			&MqttOnline{})
-	}
-	return mgr
-}
-
-func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	}
-}
-
-func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	}
-}
-
-func (o *MQTTMgr) Publish(topic string, payload string, qos mqtt.QOS, tp OptType) {
-	msg := MQTTMessage{
-		topic:   topic,
-		payload: payload,
-		qos:     qos,
-		tp:      tp,
-	}
-	o.Queue.Put(&msg)
-}
-
-func (o *MQTTMgr) _publish(msg *MQTTMessage) error {
-	var err error
-	switch msg.tp {
-	case ToAll:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	}
-	return err
-}
-
-func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
-	for {
-		time.Sleep(10 * time.Second)
-		//GetMonitorStatus().MQTTConnectMgr = time.Now() //更新状态
-		if o.Cloud != nil {
-			o.Cloud.Connect()
-		}
-		if o.Edge != nil {
-			o.Edge.Connect()
-		}
-	}
-}
-
-func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
-			time.Sleep(time.Second)
-			gopool.Add(o.MQTTMessageHandle, args)
-		}
-	}()
-	var err error
-	for {
-		if m, ok, _ := o.Queue.Get(); ok {
-			if msg, ok := m.(*MQTTMessage); ok {
-			RETRY:
-				err = o._publish(msg)
-				if err != nil {
-					logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
-					time.Sleep(time.Second)
-					goto RETRY
-				}
-			}
-		} else {
-			time.Sleep(200 * time.Millisecond)
-			//更新状态时间
-			//GetMonitorStatus().MQTTMessageHandle = time.Now()
-		}
-	}
-}
-
-type MQTTMessage struct {
-	topic   string
-	payload string
-	qos     mqtt.QOS
-	tp      OptType
-}

edge/ipole/ym485.go

@@ -141,7 +141,7 @@ func (o *YmLampController) UpdateModel2(mi *ModelInfo) {
 		logrus.Errorf("YmLampController.UpdateModel2:加载模型[%d]文件错误:%s", mi.TID, err.Error())
 		return
 	}
-	if iot.Protocol == ModbusRtuProtocol { //合法的物模型
+	if iot.Protocol == YmProtocol { //合法的物模型
 		o.model = iot
 		logrus.Infof("YmLampController.UpdateModel2:更新设备[%s]的物模型[%d]成功", o.devinfo.DevCode, mi.TID)
 	} else {
@@ -193,7 +193,10 @@ func (o *YmLampController) HandleData() {
 					t := util.MlNow()
 
 					o.QueryDeviceState() //获取灯的数据
-					o.ConfirmState(t)    //获取设置灯的开/关
+					// QueryDeviceState失败时State为过期数据，跳过ConfirmState避免误操作
+					if len(o.State) > 0 {
+						o.ConfirmState(t) //获取设置灯的开/关
+					}
 
 					nextFillTime = util.MlNow().Add(time.Duration(o.devinfo.SendCloud) * time.Millisecond)
 				}
@@ -210,8 +213,12 @@ func (o *YmLampController) SendRecvData(aduRequest []byte, retry int) (aduRespon
 	// time.Sleep(time.Millisecond * 10)
 	serial := GetSerialMgr().GetSerialPort(o.devinfo.Code)
 	if serial == nil {
+		logrus.Errorf("YM串口未找到: Code=%d", o.devinfo.Code) // ← 加这行
 		return nil, ErrClosedConnection
 	}
+	if !serial.IsConnected() {
+		logrus.Errorf("YM串口未连接: Code=%d", o.devinfo.Code) // ← 加这行
+	}
 	if retry <= 0 {
 		retry = 1
 	}
@@ -305,7 +312,7 @@ func (o *YmLampController) QueryDeviceState() {
 	}
 	recvbuf, err := o.SendRecvData(buf.Bytes(), 1) //根据地址得到数据
 	if err != nil {
-		logrus.Error("单灯状态数据查询错误")
+		logrus.Errorf("单灯状态数据查询错误,DevCode=%s,err=%v", o.devinfo.DevCode, err)
 		return
 	}
 
@@ -352,7 +359,9 @@ func (o *YmLampController) QueryDeviceState() {
 	mapData[o.devinfo.DevCode] = &data
 	var ret1 protocol.Pack_CHZB_UploadData
 	if str, err := ret1.EnCode(o.devinfo.DevCode, appConfig.GID, GetNextUint64(), o.devinfo.TID, mapData); err == nil {
-		GetMQTTMgr().Publish(GetTopic(protocol.DT_LAMPCONTROLLER, o.devinfo.DevCode, protocol.TP_YM_DATA), str, mqtt.AtMostOnce, ToCloud)
+		topic := GetTopic(protocol.DT_LAMPCONTROLLER, o.devinfo.DevCode, protocol.TP_YM_DATA)
+		logrus.Infof("QueryDeviceState:发布灯控数据,topic=%s,payload=%s", topic, str)
+		GetMQTTMgr().Publish(topic, str, mqtt.AtLeastOnce, ToCloud)
 	}
 }
 
@@ -369,15 +378,17 @@ func (o *YmLampController) TurnOnOff(flag uint8) error {
 	}
 	recvbuf, err := o.SendRecvData(buf.Bytes(), 3)
 	if err != nil {
+		logrus.Errorf("TurnOnOff:串口通信失败,DevCode=%s,flag=%d,err=%v", o.devinfo.DevCode, flag, err)
 		return err
 	}
 	var ret ym485.PackingResult
 	err = ret.DeCode(recvbuf)
 	if err == nil {
-		if ret.Result == 89 { //字母'Y'
+		if ret.Result == 89 { //字母Y
 			return nil
 		}
 	}
+	logrus.Errorf("TurnOnOff:设备返回失败,DevCode=%s,flag=%d,Result=0x%02X", o.devinfo.DevCode, flag, ret.Result)
 	return errors.New(protocol.FAILED_STR)
 }
 
@@ -500,7 +511,9 @@ func (o *YmLampController) HandleTpYmSetSwitch(m mqtt.Message) {
 		brightness = obj.Data.Brightness
 	}
 	err := o.Switch(obj.Data.Switch, obj.Data.Brightness)
-	if err == nil {
+	if err != nil {
+		logrus.Errorf("HandleTpYmSetSwitch:开关灯失败,DevCode=%s,Switch=%d,err=%v", o.devinfo.DevCode, obj.Data.Switch, err)
+	} else {
 		o.tswitch = util.MlNow()
 		mapRedisTempLampsOOT := make(map[string]interface{}) //临时开关灯记录，用于排除异常亮灯正常亮灯的情况
 		ltr := LampTimeRange{

edge/its/main.go

@@ -8,12 +8,12 @@ import (
 	"lc/common/util"
 )
 
-//这个网页是新建的， 建好之后，有4点需要处理：admin 168
+// 这个网页是新建的， 建好之后，有4点需要处理：admin 168
 // 1. 网关build的目录下有对应的index.html，
 // 2. 设制网关外网地址（如： http://192.168.0.63:9180/index.html）
 // 3. 新建的vsn文件重新发布；
 // 4. 节目管理中排优先播放
-//不是很确定，主要是云端这个以前设置的，而现在又没有这样的接口，可以先将云端的svn修改一下名称，再刚才的本地工具删除云端的节目再次发布一下本地的试试看
+// 不是很确定，主要是云端这个以前设置的，而现在又没有这样的接口，可以先将云端的svn修改一下名称，再刚才的本地工具删除云端的节目再次发布一下本地的试试看
 var IDGen util.IdWorker
 var gopool *util.Pool
 
@@ -29,7 +29,7 @@ func GetNextUint64() uint64 {
 	return uint64(u64)
 }
 
-//cgo学习资料：https://www.cnblogs.com/binHome/p/12984064.html
+// cgo学习资料：https://www.cnblogs.com/binHome/p/12984064.html
 func main() {
 	defer uninit()
 
@@ -58,6 +58,9 @@ func main() {
 	GetITSDeviceMgr().LoadITSDevice()
 
 	gopool = util.NewPool(10)
+	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
+		gopool.Add(fn, args)
+	})
 	gopool.Add(GetMQTTMgr().MQTTMessageHandle, 1)
 	gopool.Add(GetITSDeviceMgr().Handle, 2)
 	gopool.Add(GetWebSvr().StartSvr, 3)

edge/its/mqtt_init.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"sync"
+
+	"lc/common/mqtt"
+)
+
+type OptType = mqtt.OptType
+type MQTTMessage = mqtt.MQTTMessage
+
+const (
+	ToAll   = mqtt.ToAll
+	ToCloud = mqtt.ToCloud
+	ToEdge  = mqtt.ToEdge
+)
+
+var _mqttMgrOnce sync.Once
+var _mqttMgr *mqtt.MQTTMgr
+
+func GetMQTTMgr() *mqtt.MQTTMgr {
+	_mqttMgrOnce.Do(func() {
+		_mqttMgr = mqtt.NewMQTTMgr(mqtt.MQTTMgrConfig{
+			CloudServer:   appConfig.Cloud.Mqtt.Server,
+			CloudClientID: appConfig.GID + "@" + appname + version,
+			CloudUser:     appConfig.Cloud.Mqtt.User,
+			CloudPassword: appConfig.Cloud.Mqtt.Password,
+			CloudTimeout:  appConfig.Cloud.Mqtt.Timeout,
+			EdgeServer:    appConfig.Edge.Mqtt.Server,
+			EdgeClientID:  appConfig.GID + "@" + appname + version,
+			EdgeUser:      appConfig.Edge.Mqtt.User,
+			EdgePassword:  appConfig.Edge.Mqtt.Password,
+			EdgeTimeout:   appConfig.Edge.Mqtt.Timeout,
+		})
+	})
+	return _mqttMgr
+}

edge/its/mqttmgr.go

@@ -1,181 +0,0 @@
-package main
-
-import (
-	"runtime/debug"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-	"lc/common/util"
-)
-
-type OptType uint8
-
-const (
-	ToAll   OptType = 0 //发布和订阅边缘端与云端的消息
-	ToCloud OptType = 1 //发布和订阅云端的消息
-	ToEdge  OptType = 2 //发布和订阅边缘端的消息
-)
-
-var _mqttMgronce sync.Once
-var _mqttMgrsingle *MQTTMgr
-
-func GetMQTTMgr() *MQTTMgr {
-	_mqttMgronce.Do(func() {
-		_mqttMgrsingle = _newMQTTMgr()
-	})
-	return _mqttMgrsingle
-}
-
-type MQTTMgr struct {
-	Cloud *MqttClient
-	Edge  *MqttClient
-	Queue *util.MlQueue
-}
-
-func _newMQTTMgr() *MQTTMgr {
-	mgr := &MQTTMgr{
-		Queue: util.NewQueue(2000),
-	}
-	if appConfig.Edge.Mqtt.Server != "" {
-		mgr.Edge = NewMqttClient(appConfig.Edge.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Edge.Mqtt.User,
-			appConfig.Edge.Mqtt.Password,
-			appConfig.Edge.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	if appConfig.Cloud.Mqtt.Server != "" {
-		mgr.Cloud = NewMqttClient(appConfig.Cloud.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Cloud.Mqtt.User,
-			appConfig.Cloud.Mqtt.Password,
-			appConfig.Cloud.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	return mgr
-}
-
-func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	}
-}
-
-func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	}
-}
-
-func (o *MQTTMgr) Publish(topic string, payload string, qos mqtt.QOS, tp OptType) {
-	msg := MQTTMessage{
-		topic:   topic,
-		payload: payload,
-		qos:     qos,
-		tp:      tp,
-	}
-	o.Queue.Put(&msg)
-}
-
-func (o *MQTTMgr) _publish(msg *MQTTMessage) error {
-	var err error
-	switch msg.tp {
-	case ToAll:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	}
-	return err
-}
-
-func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
-	for {
-		time.Sleep(10 * time.Second)
-		if o.Cloud != nil {
-			o.Cloud.Connect()
-		}
-		if o.Edge != nil {
-			o.Edge.Connect()
-		}
-	}
-}
-
-func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
-			time.Sleep(time.Second)
-			gopool.Add(o.MQTTMessageHandle, args)
-		}
-	}()
-	var err error
-	for {
-		if m, ok, _ := o.Queue.Get(); ok {
-			if msg, ok := m.(*MQTTMessage); ok {
-			RETRY:
-				err = o._publish(msg)
-				if err != nil {
-					logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
-					time.Sleep(time.Second)
-					goto RETRY
-				}
-			}
-		} else {
-			time.Sleep(200 * time.Millisecond)
-		}
-	}
-}
-
-type MQTTMessage struct {
-	topic   string
-	payload string
-	qos     mqtt.QOS
-	tp      OptType
-}

edge/its_win/main.go

@@ -43,6 +43,9 @@ func main() {
 	GetWebSvr().MQTTSubscribe()
 
 	gopool = util.NewPool(10)
+	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
+		gopool.Add(fn, args)
+	})
 	gopool.Add(GetMQTTMgr().MQTTMessageHandle, 1)
 	gopool.Add(GetWebSvr().StartSvr, 3)
 	gopool.Add(GetWebSvr().Handle, 4)

edge/its_win/mqtt_init.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"sync"
+
+	"lc/common/mqtt"
+)
+
+type OptType = mqtt.OptType
+type MQTTMessage = mqtt.MQTTMessage
+
+const (
+	ToAll   = mqtt.ToAll
+	ToCloud = mqtt.ToCloud
+	ToEdge  = mqtt.ToEdge
+)
+
+var _mqttMgrOnce sync.Once
+var _mqttMgr *mqtt.MQTTMgr
+
+func GetMQTTMgr() *mqtt.MQTTMgr {
+	_mqttMgrOnce.Do(func() {
+		_mqttMgr = mqtt.NewMQTTMgr(mqtt.MQTTMgrConfig{
+			CloudServer:   appConfig.Cloud.Mqtt.Server,
+			CloudClientID: appConfig.GID + "@" + appname + version,
+			CloudUser:     appConfig.Cloud.Mqtt.User,
+			CloudPassword: appConfig.Cloud.Mqtt.Password,
+			CloudTimeout:  appConfig.Cloud.Mqtt.Timeout,
+			EdgeServer:    appConfig.Edge.Mqtt.Server,
+			EdgeClientID:  appConfig.GID + "@" + appname + version,
+			EdgeUser:      appConfig.Edge.Mqtt.User,
+			EdgePassword:  appConfig.Edge.Mqtt.Password,
+			EdgeTimeout:   appConfig.Edge.Mqtt.Timeout,
+		})
+	})
+	return _mqttMgr
+}

edge/its_win/mqttclient.go

@@ -1,138 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientid,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:%v", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		o.Subscribe(k, v)
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		o.PublishString(topic, str, 0)
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	if !o.mqtt.IsConnected() {
-		ctx, cancel := o.Ctx()
-		defer cancel()
-		return o.mqtt.Connect(ctx)
-	}
-	return nil
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

edge/its_win/mqttmgr.go

@@ -1,181 +0,0 @@
-package main
-
-import (
-	"runtime/debug"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-	"lc/common/util"
-)
-
-type OptType uint8
-
-const (
-	ToAll   OptType = 0 //发布和订阅边缘端与云端的消息
-	ToCloud OptType = 1 //发布和订阅云端的消息
-	ToEdge  OptType = 2 //发布和订阅边缘端的消息
-)
-
-var _mqttMgronce sync.Once
-var _mqttMgrsingle *MQTTMgr
-
-func GetMQTTMgr() *MQTTMgr {
-	_mqttMgronce.Do(func() {
-		_mqttMgrsingle = _newMQTTMgr()
-	})
-	return _mqttMgrsingle
-}
-
-type MQTTMgr struct {
-	Cloud *MqttClient
-	Edge  *MqttClient
-	Queue *util.MlQueue
-}
-
-func _newMQTTMgr() *MQTTMgr {
-	mgr := &MQTTMgr{
-		Queue: util.NewQueue(2000),
-	}
-	if appConfig.Edge.Mqtt.Server != "" {
-		mgr.Edge = NewMqttClient(appConfig.Edge.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Edge.Mqtt.User,
-			appConfig.Edge.Mqtt.Password,
-			appConfig.Edge.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	if appConfig.Cloud.Mqtt.Server != "" {
-		mgr.Cloud = NewMqttClient(appConfig.Cloud.Mqtt.Server,
-			appConfig.GID+"@"+appname+version,
-			appConfig.Cloud.Mqtt.User,
-			appConfig.Cloud.Mqtt.Password,
-			appConfig.Cloud.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	return mgr
-}
-
-func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	}
-}
-
-func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	}
-}
-
-func (o *MQTTMgr) Publish(topic string, payload string, qos mqtt.QOS, tp OptType) {
-	msg := MQTTMessage{
-		topic:   topic,
-		payload: payload,
-		qos:     qos,
-		tp:      tp,
-	}
-	o.Queue.Put(&msg)
-}
-
-func (o *MQTTMgr) _publish(msg *MQTTMessage) error {
-	var err error
-	switch msg.tp {
-	case ToAll:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			err = o.Cloud.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.PublishString(msg.topic, msg.payload, msg.qos)
-		}
-	}
-	return err
-}
-
-func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
-	for {
-		time.Sleep(10 * time.Second)
-		if o.Cloud != nil {
-			o.Cloud.Connect()
-		}
-		if o.Edge != nil {
-			o.Edge.Connect()
-		}
-	}
-}
-
-func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
-			time.Sleep(time.Second)
-			gopool.Add(o.MQTTMessageHandle, args)
-		}
-	}()
-	var err error
-	for {
-		if m, ok, _ := o.Queue.Get(); ok {
-			if msg, ok := m.(*MQTTMessage); ok {
-			RETRY:
-				err = o._publish(msg)
-				if err != nil {
-					logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
-					time.Sleep(time.Second)
-					goto RETRY
-				}
-			}
-		} else {
-			time.Sleep(200 * time.Millisecond)
-		}
-	}
-}
-
-type MQTTMessage struct {
-	topic   string
-	payload string
-	qos     mqtt.QOS
-	tp      OptType
-}

edge/led_screen/led_device.go

@@ -113,7 +113,7 @@ func (o *LedDevice) CheckOnline(dev *LedDevice, online uint8) {
 	if str, err := obj.EnCode(dev.SN, appConfig.GID, GetNextUint64(), util.MlNow(), online); err == nil {
 		topic := appConfig.Tenant + "/cltled/" + dev.SN + "/up/" + protocol.TP_LED_STATE + "/" + strconv.Itoa(int(GetNextUint64()))
 		//fmt.Printf("topic = %v str=%v \n", topic, str)
-		GetMQTTMgr().Publish(topic, []byte(str), mqtt.AtMostOnce, ToCloud)
+		GetMQTTMgr().Publish(topic, str, mqtt.AtMostOnce, ToCloud)
 	}
 }
 
@@ -149,7 +149,7 @@ func (o *LedDevice) LedData(dev *LedDevice) {
 	var obj protocol.Pack_LedCltledData
 	if str, err := obj.EnCode(dev.SN, appConfig.GID, GetNextUint64(), &data); err == nil {
 		topic := appConfig.Tenant + "/cltled/" + dev.SN + "/up/" + protocol.TP_LED_DATA + "/" + strconv.Itoa(int(GetNextUint64()))
-		GetMQTTMgr().Publish(topic, []byte(str), mqtt.AtMostOnce, ToCloud)
+		GetMQTTMgr().Publish(topic, str, mqtt.AtMostOnce, ToCloud)
 	}
 }
 
@@ -243,7 +243,7 @@ func (l *LedDevice) GetVolume(m mqtt.Message) {
 func (l *LedDevice) SnapShot(m mqtt.Message) {
 	url := l.GetCltUrl(clt_client.SNAPSHOT)
 	shot := clt_client.GetClient().SnapShot(url)
-	GetMQTTMgr().Publish(getUpTopic(m.Topic()), shot, mqtt.AtMostOnce, ToCloud)
+	GetMQTTMgr().Publish(getUpTopic(m.Topic()), string(shot), mqtt.AtMostOnce, ToCloud)
 }
 
 func (l *LedDevice) DoCommonGet(url, funcname string, m mqtt.Message) {
@@ -251,7 +251,7 @@ func (l *LedDevice) DoCommonGet(url, funcname string, m mqtt.Message) {
 	if err != nil {
 		logrus.Errorf(funcname+" error: %v", err)
 	}
-	GetMQTTMgr().Publish(getUpTopic(m.Topic()), resp, mqtt.AtMostOnce, ToCloud)
+	GetMQTTMgr().Publish(getUpTopic(m.Topic()), string(resp), mqtt.AtMostOnce, ToCloud)
 }
 
 //================================PUT

edge/led_screen/main.go

@@ -28,6 +28,9 @@ func main() {
 		return
 	}
 	gopool = util.NewPool(5)
+	GetMQTTMgr().SetRestartFn(func(fn func(args ...interface{}) interface{}, args ...interface{}) {
+		gopool.Add(fn, args)
+	})
 	gopool.Add(GetMQTTMgr().MQTTMessageHandle, 1)
 	gopool.Add(GetMQTTMgr().MQTTConnectMgr, 2)
 	gopool.Run()

edge/led_screen/mqtt_init.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"sync"
+
+	"lc/common/mqtt"
+)
+
+type OptType = mqtt.OptType
+type MQTTMessage = mqtt.MQTTMessage
+
+const (
+	ToAll   = mqtt.ToAll
+	ToCloud = mqtt.ToCloud
+	ToEdge  = mqtt.ToEdge
+)
+
+var _mqttMgrOnce sync.Once
+var _mqttMgr *mqtt.MQTTMgr
+
+func GetMQTTMgr() *mqtt.MQTTMgr {
+	_mqttMgrOnce.Do(func() {
+		_mqttMgr = mqtt.NewMQTTMgr(mqtt.MQTTMgrConfig{
+			CloudServer:   appConfig.Cloud.Mqtt.Server,
+			CloudClientID: appConfig.GID + "@" + appname + appversion,
+			CloudUser:     appConfig.Cloud.Mqtt.User,
+			CloudPassword: appConfig.Cloud.Mqtt.Password,
+			CloudTimeout:  appConfig.Cloud.Mqtt.Timeout,
+			EdgeServer:    appConfig.Edge.Mqtt.Server,
+			EdgeClientID:  appConfig.GID + "@" + appname + appversion,
+			EdgeUser:      appConfig.Edge.Mqtt.User,
+			EdgePassword:  appConfig.Edge.Mqtt.Password,
+			EdgeTimeout:   appConfig.Edge.Mqtt.Timeout,
+		})
+	})
+	return _mqttMgr
+}

edge/led_screen/mqttclient.go

@@ -1,138 +0,0 @@
-package main
-
-import (
-	"context"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-)
-
-type BaseMqttOnline interface {
-	GetOnlineMsg() (string, string)
-	GetWillMsg() (string, string)
-}
-
-type EmptyMqttOnline struct {
-}
-
-func (o *EmptyMqttOnline) GetOnlineMsg() (string, string) {
-	return "", ""
-}
-func (o *EmptyMqttOnline) GetWillMsg() (string, string) {
-	return "", ""
-}
-
-type MqttClient struct {
-	mqtt       *mqtt.Client        //
-	mu         sync.Mutex          //保护mapTopics
-	mapTopics  map[string]mqtt.QOS //订阅的主题
-	timeout    uint                //超时时间,毫秒为单位
-	MqttOnline BaseMqttOnline      //是否发布上线消息&遗嘱消息
-}
-
-func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOnline BaseMqttOnline) *MqttClient {
-	o := MqttClient{
-		mapTopics:  make(map[string]mqtt.QOS),
-		timeout:    timeout,
-		MqttOnline: mqttOnline,
-	}
-	client, err := mqtt.NewClient(mqtt.ClientOptions{
-		Servers:       []string{server},
-		ClientID:      clientid,
-		Username:      user,
-		Password:      password,
-		AutoReconnect: true,
-	}, &o)
-	if err != nil {
-		panic(fmt.Sprintf("MQTT错误:", err.Error()))
-		return nil
-	}
-	o.mqtt = client
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	err = client.Connect(ctx)
-	return &o
-}
-
-func (o *MqttClient) ConnectionLostHandler(err error) {
-	logrus.Errorln("MqttClient.ConnectionLostHandler:MQTT连接已经断开,原因:", err)
-}
-
-func (o *MqttClient) OnConnectHandler() {
-	logrus.Infoln("MqttClient.OnConnectHandler:MQTT连接成功")
-	//连接成功则订阅主题
-	for k, v := range o.mapTopics {
-		o.Subscribe(k, v)
-	}
-	topic, str := o.MqttOnline.GetOnlineMsg()
-	if topic != "" {
-		o.PublishString(topic, str, 0)
-	}
-}
-
-func (o *MqttClient) GetWill() (topic string, payload string) {
-	return o.MqttOnline.GetWillMsg()
-}
-
-func (o *MqttClient) Connect() error {
-	if !o.mqtt.IsConnected() {
-		ctx, cancel := o.Ctx()
-		defer cancel()
-		return o.mqtt.Connect(ctx)
-	}
-	return nil
-}
-
-func (o *MqttClient) IsConnected() bool {
-	return o.mqtt.IsConnected()
-}
-
-func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Publish(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishString(ctx, topic, payload, qos)
-}
-func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
-}
-
-func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; !ok {
-		o.mapTopics[topic] = qos
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Subscribe(ctx, topic, qos)
-}
-
-func (o *MqttClient) Unsubscribe(topic string) error {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	if _, ok := o.mapTopics[topic]; ok {
-		delete(o.mapTopics, topic)
-	}
-	ctx, cancel := o.Ctx()
-	defer cancel()
-	return o.mqtt.Unsubscribe(ctx, topic)
-}
-
-func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
-	return o.mqtt.Handle(topic, handler)
-}
-
-func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
-	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-}

edge/led_screen/mqttmgr.go

@@ -1,188 +0,0 @@
-package main
-
-import (
-	"runtime/debug"
-	"sync"
-	"time"
-
-	"github.com/sirupsen/logrus"
-
-	"lc/common/mqtt"
-	"lc/common/util"
-)
-
-type OptType uint8
-
-const (
-	ToAll   OptType = 0 //发布和订阅边缘端与云端的消息
-	ToCloud OptType = 1 //发布和订阅云端的消息
-	ToEdge  OptType = 2 //发布和订阅边缘端的消息
-)
-
-var _mqttMgronce sync.Once
-var _mqttMgrsingle *MQTTMgr
-
-// GetMQTTMgr 单态
-func GetMQTTMgr() *MQTTMgr {
-	_mqttMgronce.Do(func() {
-		_mqttMgrsingle = _newMQTTMgr()
-	})
-	return _mqttMgrsingle
-}
-
-type MQTTMgr struct {
-	Cloud *MqttClient
-	Edge  *MqttClient
-	Queue *util.MlQueue
-}
-
-// 建两个client
-func _newMQTTMgr() *MQTTMgr {
-	mgr := &MQTTMgr{
-		Queue: util.NewQueue(2000),
-	}
-	if appConfig.Edge.Mqtt.Server != "" {
-		mgr.Edge = NewMqttClient(appConfig.Edge.Mqtt.Server,
-			appConfig.GID+"@"+appname+appversion,
-			appConfig.Edge.Mqtt.User,
-			appConfig.Edge.Mqtt.Password,
-			appConfig.Edge.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	if appConfig.Cloud.Mqtt.Server != "" {
-		mgr.Cloud = NewMqttClient(appConfig.Cloud.Mqtt.Server,
-			appConfig.GID+"@"+appname+appversion,
-			appConfig.Cloud.Mqtt.User,
-			appConfig.Cloud.Mqtt.Password,
-			appConfig.Cloud.Mqtt.Timeout,
-			&EmptyMqttOnline{})
-	}
-	return mgr
-}
-
-// Subscribe 定阅
-func (o *MQTTMgr) Subscribe(topic string, qos mqtt.QOS, handler mqtt.MessageHandler, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Handle(topic, handler)
-			o.Cloud.Subscribe(topic, qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Handle(topic, handler)
-			o.Edge.Subscribe(topic, qos)
-		}
-	}
-}
-
-// UnSubscribe 退定
-func (o *MQTTMgr) UnSubscribe(topic string, tp OptType) {
-	switch tp {
-	case ToAll:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			o.Cloud.Unsubscribe(topic)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Unsubscribe(topic)
-		}
-	}
-}
-
-// Publish 发布进队列
-func (o *MQTTMgr) Publish(topic string, payload []byte, qos mqtt.QOS, tp OptType) {
-	msg := MQTTMessage{
-		topic:   topic,
-		payload: payload,
-		qos:     qos,
-		tp:      tp,
-	}
-	o.Queue.Put(&msg)
-}
-
-// 发布低
-func (o *MQTTMgr) _publish(msg *MQTTMessage) error {
-	var err error
-	switch msg.tp {
-	case ToAll:
-		if o.Cloud != nil {
-			err = o.Cloud.Publish(msg.topic, msg.payload, msg.qos)
-		}
-		if o.Edge != nil {
-			o.Edge.Publish(msg.topic, msg.payload, msg.qos)
-		}
-	case ToCloud:
-		if o.Cloud != nil {
-			err = o.Cloud.Publish(msg.topic, msg.payload, msg.qos)
-		}
-	case ToEdge:
-		if o.Edge != nil {
-			o.Edge.Publish(msg.topic, msg.payload, msg.qos)
-		}
-	}
-	return err
-}
-
-// MQTTConnectMgr 连接保持
-func (o *MQTTMgr) MQTTConnectMgr(args ...interface{}) interface{} {
-	for {
-		time.Sleep(10 * time.Second)
-		if o.Cloud != nil {
-			o.Cloud.Connect()
-		}
-		if o.Edge != nil {
-			o.Edge.Connect()
-		}
-	}
-}
-
-func (o *MQTTMgr) MQTTMessageHandle(args ...interface{}) interface{} {
-	defer func() {
-		if err := recover(); err != nil {
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常:%v", err)
-			logrus.Errorf("MQTTMgr.MQTTMessageHandle发生异常,堆栈信息:%s", string(debug.Stack()))
-			time.Sleep(time.Second)
-			gopool.Add(o.MQTTMessageHandle, args)
-		}
-	}()
-	var err error
-	for { //队列中所有发布
-		if m, ok, _ := o.Queue.Get(); ok {
-			if msg, ok := m.(*MQTTMessage); ok {
-			RETRY:
-				err = o._publish(msg)
-				if err != nil {
-					logrus.Errorf("发布主题为%s的消息失败，原因:%s", msg.topic, err.Error())
-					time.Sleep(time.Second)
-					goto RETRY
-				}
-			}
-		} else {
-			time.Sleep(200 * time.Millisecond)
-		}
-	}
-}
-
-type MQTTMessage struct {
-	topic   string
-	payload []byte
-	qos     mqtt.QOS
-	tp      OptType
-}