mqtt: update

module/mqtt/mqtt_handle.go

@@ -35,23 +35,25 @@ var (
 
 // InitReceiverMapUpdater 2小时更新接收器数据
 func (e *Entry) InitReceiverMapUpdater() {
-	go func() {
-		ticker := time.NewTicker(2 * time.Hour)
-		defer ticker.Stop()
-		zaplog.Info("ReceiverMap updated successfully1")
-		for {
-			select {
-			case <-ticker.C:
-				receiverMutex.Lock()
-				receiverMap = e.FindAppPastureReceiver()
-				receiverMutex.Unlock()
-				zaplog.Info("ReceiverMap updated successfully2")
-			}
+	ticker := time.NewTicker(2 * time.Hour)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			receiverMutex.Lock()
+			receiverMap = e.FindAppPastureReceiver()
+			receiverMutex.Unlock()
+			zaplog.Info("ReceiverMap updated successfully")
 		}
-	}()
+	}
 }
 
 func (e *Entry) NeckRingHandle(data []byte) {
+	if len(receiverMap) <= 0 {
+		receiverMutex.Lock()
+		receiverMap = e.FindAppPastureReceiver()
+		receiverMutex.Unlock()
+	}
 	newReceiverMap := receiverMap
 	newData := e.MsgDataFormat2(data, newReceiverMap)
 	if newData == nil {

service/mqtt/interface.go

@@ -145,6 +145,6 @@ func (s *IMqttClient) Run(enter handleMqtt.Entry) {
 	}
 
 	// 启动数据处理
-	enter.InitReceiverMapUpdater()
+	go enter.InitReceiverMapUpdater()
 	s.Consumer(enter.NeckRingHandle)
 }