入库es的时间格式改成2025-09-29T14:11:39.347+0800格式

adaptors/安心云最新主题数据toE5.go

@@ -43,11 +43,16 @@ func (the Adaptor_Anxinyun_LastTheme) Theme2Es(theme models.AXYSavoirTheme) *mod
 	//}
 	//
 	Atime, err := time.Parse("2006-01-02T15:04:05.000+0800", theme.AcqTime)
+	AtimeStr := Atime.Format("2006-01-02T15:04:05.000+0800")
+	log.Printf("AtimeStr[%s]", AtimeStr)
 	if err != nil {
 		log.Printf("安心云 测点[%s] 数据时间 %s 解析错误", theme.Station.Name, theme.AcqTime)
 		return nil
 	}
 
+	//now := time.Now()
+	//formattedTime := now.Format("2006-01-02T15:04:05.000")
+
 	themeData := &models.EsTheme{
 		SensorName:      theme.Station.Name,
 		FactorName:      theme.Station.Factor.Name,
@@ -55,7 +60,7 @@ func (the Adaptor_Anxinyun_LastTheme) Theme2Es(theme models.AXYSavoirTheme) *mod
 		Data:            theme.Data,
 		FactorProtoName: theme.Station.Factor.ProtoName,
 		Factor:          theme.Station.Factor.Id,
-		CollectTime:     Atime,
+		CollectTime:     AtimeStr,
 		Sensor:          theme.Station.Id,
 		Structure:       theme.Station.Structure.Id,
 		IotaDevice:      []string{},

adaptors/知物云最新主题数据toES.go

@@ -43,6 +43,8 @@ func (the Adaptor_Savoir_LastTheme) Theme2Es(theme models.SavoirTheme) *models.E
 	//}
 	//
 	Atime, err := time.Parse("2006-01-02T15:04:05.000+0800", theme.AcqTime)
+	AtimeStr := Atime.Format("2006-01-02T15:04:05.000+0800")
+	log.Printf("AtimeStr[%s]", AtimeStr)
 	if err != nil {
 		log.Printf("知物云 测点[%s] 数据时间 %s 解析错误", theme.Station.Name, theme.AcqTime)
 		return nil
@@ -55,7 +57,7 @@ func (the Adaptor_Savoir_LastTheme) Theme2Es(theme models.SavoirTheme) *models.E
 		Data:            theme.Data,
 		FactorProtoName: theme.Station.Factor.ProtoName,
 		Factor:          theme.Station.Factor.Id,
-		CollectTime:     Atime,
+		CollectTime:     AtimeStr,
 		Sensor:          theme.Station.Id,
 		Structure:       theme.Station.Structure.Id,
 		IotaDevice:      []string{},

configFiles/config_安心云测点数据_最新同步.yaml

@@ -11,8 +11,6 @@ ioConfig:
     es:
       address:
         - "http://10.8.30.150:30847"
-      #        - "http://10.8.30.155:5601"
-      #        - "http://10.8.30.160:30092"
       index: "anxincloud_last_theme" #es存测点最后一条数据的新索引
       auth:
         userName: post

consumers/consumerAXYThemeToES.go

@@ -80,6 +80,7 @@ func (the *consumerAXYThemeToES) outputInitial() error {
 
 func (the *consumerAXYThemeToES) sinkTask() {
 	intervalSec := the.Info.IoConfig.Out.Es.Interval
+	log.Printf("读取配置文件intervalSec[%d]", intervalSec)
 	ticker := time.NewTicker(time.Duration(intervalSec) * time.Second)
 	defer ticker.Stop()
 	for {
@@ -108,7 +109,7 @@ func (the *consumerAXYThemeToES) toSink() {
 	})
 	if len(themes) > 0 {
 		index := the.Info.IoConfig.Out.Es.Index
-		log.Printf("写入es [%s] %d条", index, len(themes))
+		log.Printf("写入es [%s] %d条，%s", index, len(themes), themes)
 		the.OutEs.BulkWriteThemes2Es(index, themes)
 		the.sinkMap.Clear()
 	}
@@ -146,7 +147,9 @@ func (the *consumerAXYThemeToES) onData(topic string, msg string) bool {
 		the.dataCache <- needPush
 	} else {
 		s, _ := json.Marshal(needPush)
+
 		if needPush != nil {
+			log.Printf("onData 测点[%d] needPush=  %s", needPush.Sensor, s)
 			if needPush.Sensor == the.logTagId {
 				log.Printf("onData 测点[%d] 异常needPush=  %s", needPush.Sensor, s)
 			}

models/esTheme.go

@@ -9,7 +9,7 @@ type EsTheme struct {
 	Data            map[string]any `json:"data"`
 	FactorProtoName string         `json:"factor_proto_name"`
 	Factor          int            `json:"factor"`
-	CollectTime     time.Time      `json:"collect_time"`
+	CollectTime     string         `json:"collect_time"`
 	Sensor          int            `json:"sensor"`
 	Structure       int            `json:"structure"`
 	IotaDevice      []string       `json:"iota_device"`