package kafka import ( "dataSource" "encoding/json" "gitea.anxinyun.cn/container/common_models" "gitea.anxinyun.cn/container/common_utils" "gitea.anxinyun.cn/container/common_utils/configLoad" "log" "strings" "time" ) type AggDataHandler struct { configHelper *common_utils.ConfigHelper } func NewAggDataHandler() *AggDataHandler { redisAddr := configLoad.LoadConfig().GetString("redis.address") return &AggDataHandler{ configHelper: common_utils.NewConfigHelper(redisAddr), } } func (h AggDataHandler) HandleMessage(message string) bool { // aggDataMsg: {"date":"2024-09-19T09:39:59.999+0000","sensorId":106,"structId":1,"factorId":11,"aggTypeId":2006,"aggMethodId":3004,"agg":{"strain":-19.399999618530273},"changed":{"strain":-3}} // aggDataMsg 中的时间为UTC格式 2024-04-19T01:10:59.999+0000, // 在进行 json.Unmarshal() 时报错 // 解决方案:先将 +0000 -> Z,然后再将 UTC 时间转换为中国时区时间("Asia/Shanghai") // 2024-09-28T23:59:59.999+0800 // 将 2024-04-19T01:10:59.999+0000 -> 2024-04-19T01:10:59.999Z utcTimeStr := strings.Replace(message, "+0800", "+08:00", 1) utcTimeStr = strings.Replace(utcTimeStr, "+0000", "Z", 1) aggData := common_models.AggData{} err := json.Unmarshal([]byte(utcTimeStr), &aggData) if err != nil { log.Printf("json parse error: %v", err) return false } // 转换为中国时区时间("Asia/Shanghai") loc, _ := time.LoadLocation("Asia/Shanghai") chinaTime := aggData.Date.In(loc) aggData.Date = chinaTime //log.Printf("message:%v\n, cvt: %+v", message, aggData) if aggData.ThingId == "" { structure, err := h.configHelper.GetStructure(aggData.StructId) if err != nil { log.Printf("redis 中无 key = structure:%d 的缓存数据.", aggData.StructId) return false } aggData.ThingId = structure.ThingId } log.Printf("handler 处理sensorId[%d]消息", aggData.SensorId) dataSource.GetChannels().AggDataChan <- aggData return true }