更新K线最新的点

workers/sneakerWorkers/kLineWorker.go

@@ -73,6 +73,7 @@ func calculateInDB(marketId int, period, begin, end int64) (k KLine) {
 	}
 	backupDB.Save(&k)
 	backupDB.DbCommit()
+	synToRedis(&k)
 	return
 }
 
@@ -102,6 +103,7 @@ func calculateInBackupDB(marketId int, period, begin, end int64) (k KLine) {
 	backupDB.Model(KLine{}).Where("market_id = ?", marketId).Where("period = ?", lastPeriod).Where("? <= timestamp AND timestamp < ?", begin, end).Order("timestamp DESC").Limit(1).Select("close as close").Scan(&k)
 	backupDB.Save(&k)
 	backupDB.DbCommit()
+	synToRedis(&k)
 	return
 }
 
@@ -148,3 +150,13 @@ func synFromRedis(marketId int, period, begin, end int64) (k KLine) {
 	}
 	return
 }
+
+func synToRedis(k *KLine) {
+	kRedis := utils.GetRedisConn("k")
+	defer kRedis.Close()
+
+	b, _ := json.Marshal((*k).Data())
+	kRedis.Send("ZREMRANGEBYSCORE", (*k).RedisKey(), (*k).Timestamp)
+	kRedis.Do("ZADD", k.RedisKey(), (*k).Timestamp, string(b))
+
+}

workers/sneakerWorkers/rebuildkLineToRedisWorker.go

@@ -20,7 +20,7 @@ func (worker *Worker) RebuildKLineToRedisWorker(payloadJson *[]byte) (queueName
 	if mainDB.Where("market_id = ?", payload.MarketId).Where("period = ?", payload.Period).Find(&ks).RecordNotFound() {
 		return
 	}
-
+	mainDB.DbRollback()
 	kRedis := utils.GetRedisConn("k")
 	defer kRedis.Close()
 	for i, k := range ks {