Browse Source

fix 跨月的时候 数据可能卡住的问题

dev
lucas 3 days ago
parent
commit
d495ab4c73
  1. 31
      consumers/consumerSinoGnssMySQL.go

31
consumers/consumerSinoGnssMySQL.go

@ -90,16 +90,33 @@ func (the *consumerSinoGnssMySQL) onData() {
log.Printf("读取 缓存异常,err=%v", err.Error()) log.Printf("读取 缓存异常,err=%v", err.Error())
return return
} }
// 先切换月表,避免旧表查空或查询失败后一直无法切表。
newTableName := tableNameNow()
if recordInfo.TableName != newTableName {
log.Printf("跨月切表: %s => %s, id重置为0", recordInfo.TableName, newTableName)
recordInfo.TableName = newTableName
recordInfo.Id = 0
cacheStr, _ := json.Marshal(recordInfo)
if err := utils.SaveCache2File(string(cacheStr), "cache.inout"); err != nil {
log.Printf("保存跨月游标失败: %v", err)
return
}
}
sql := fmt.Sprintf(`select d.id,d.station_name,p.group_name,d.time,d.x,d.y,d.h from %s as d sql := fmt.Sprintf(`select d.id,d.station_name,p.group_name,d.time,d.x,d.y,d.h from %s as d
LEFT JOIN datasolution as p LEFT JOIN datasolution as p
ON d.station_name=p.sn ON d.station_name=p.sn
where p.group_name is not null where p.group_name is not null
and d.id > %d and d.id <= %d and d.id > %d
ORDER BY p.group_name;`, recordInfo.TableName, recordInfo.Id, recordInfo.Id+200) ORDER BY d.id
LIMIT 200;`, recordInfo.TableName, recordInfo.Id)
var GnssDatas []SinoGnssMySQL.GnssData var GnssDatas []SinoGnssMySQL.GnssData
err = the.InDB.Query(&GnssDatas, sql) err = the.InDB.Query(&GnssDatas, sql)
if err != nil || len(GnssDatas) == 0 { if err != nil {
log.Printf("当前批次无数据,跳过") log.Printf("查询GNSS数据失败, table=%s, id=%d, err=%v", recordInfo.TableName, recordInfo.Id, err)
return
}
if len(GnssDatas) == 0 {
log.Printf("当前批次无数据, table=%s, id=%d", recordInfo.TableName, recordInfo.Id)
return return
} }
maxId := MaxId(GnssDatas) maxId := MaxId(GnssDatas)
@ -111,12 +128,6 @@ ORDER BY p.group_name;`, recordInfo.TableName, recordInfo.Id, recordInfo.Id+200)
the.ch <- needPush the.ch <- needPush
} }
fileName := "cache.inout" fileName := "cache.inout"
//发现新月新纪录
newTableName := tableNameNow()
if recordInfo.TableName != newTableName {
recordInfo.TableName = newTableName
recordInfo.Id = 0
}
cacheStr, _ := json.Marshal(recordInfo) cacheStr, _ := json.Marshal(recordInfo)
err = utils.SaveCache2File(string(cacheStr), fileName) err = utils.SaveCache2File(string(cacheStr), fileName)
if err != nil { if err != nil {

Loading…
Cancel
Save