package mqtt import ( "errors" "fmt" "runtime" "runtime/debug" "server/utils/logger" "strings" "sync" "time" ) func InitMqtt() { MqttService = GetHandler() MqttService.SubscribeTopics() go MqttService.Handler() } var MqttService *MqttHandler type MqttHandler struct { queue *MlQueue } var _handlerOnce sync.Once var _handlerSingle *MqttHandler func GetHandler() *MqttHandler { _handlerOnce.Do(func() { _handlerSingle = &MqttHandler{ queue: NewQueue(10000), } }) return _handlerSingle } func (o *MqttHandler) SubscribeTopics() { GetMQTTMgr().Subscribe("smartIntersection/#", AtLeastOnce, o.HandlerData) } func (o *MqttHandler) HandlerData(m Message) { for { ok, cnt := o.queue.Put(&m) if ok { break } else { logger.Logger.Errorf("HandlerData:查询队列失败,队列消息数量:%d", cnt) runtime.Gosched() } } } func (o *MqttHandler) Handler() interface{} { defer func() { if err := recover(); err != nil { go GetHandler().Handler() logger.Logger.Errorf("MqttHandler.Handler:发生异常:%s", string(debug.Stack())) } }() for { msg, ok, quantity := o.queue.Get() if !ok { time.Sleep(10 * time.Millisecond) continue } else if quantity > 1000 { logger.Logger.Warnf("数据队列累积过多,请注意优化,当前队列条数:%d", quantity) } m, ok := msg.(*Message) if !ok { continue } deviceCode, topic, err := parseTopic(m.Topic()) if err != nil { logger.Logger.Errorf("parseTopic err") continue } switch topic { case TopicDeviceCamera: fmt.Println("code...", deviceCode) fmt.Println("数据...", m.PayloadString()) //TODO 待更新数据库 } } } func (o *MqttHandler) Publish(topic string, data interface{}) error { return GetMQTTMgr().Publish(topic, data, AtLeastOnce) } func (o *MqttHandler) GetTopic(deviceCode, protocol string) string { return fmt.Sprintf("smartIntersection/%s/%s", deviceCode, protocol) } func parseTopic(topic string) (string, string, error) { strList := strings.Split(topic, "/") if len(strList) < 4 { return "", "", errors.New("不支持的topic") } topic = strings.Join(strList[2:], "/") return strList[1], topic, nil } const ( TopicDeviceCamera = "device/camera" )