fix bug of real time data injection shell

deploy/redis-test-data/real-time-compute/compute_data_injection.go

@@ -13,10 +13,17 @@ import (
 
 	util "modelRT/deploy/redis-test-data/util"
 
+	"github.com/redis/go-redis/v9"
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 )
 
+const (
+	redisAddr = "localhost:6379"
+)
+
+var globalRedisClient *redis.Client
+
 var (
 	highEnd, highStart, lowStart, lowEnd int
 	totalLength                          int
@@ -30,8 +37,8 @@ func selectRandomInt() int {
 	return options[randomIndex]
 }
 
-// GenerateMixedData 生成满足特定条件的一组浮点数数据
-func GenerateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase float64) []string {
+// generateMixedData define func to generate a set of floating-point data that meets specific conditions
+func generateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase float64) []float64 {
 	totalLength = 500
 	highSegmentLength = 20
 	lowSegmentLength = 20
@@ -40,7 +47,7 @@ func GenerateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase
 	source := rand.NewSource(seed)
 	r := rand.New(source)
 
-	data := make([]string, totalLength)
+	data := make([]float64, totalLength)
 	highStart = rand.Intn(totalLength - highSegmentLength - lowSegmentLength - 1)
 	highEnd = highStart + highSegmentLength
 	lowStart = rand.Intn(totalLength-lowSegmentLength-highEnd) + highEnd
@@ -50,15 +57,15 @@ func GenerateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase
 		if i >= highStart && i < highStart+highSegmentLength {
 			// 数据值均大于 55.0，在 [55.5, 60.0] 范围内随机
 			// rand.Float64() 生成 [0.0, 1.0) 范围的浮点数
-			data[i] = strconv.FormatFloat(highMin+r.Float64()*(highBase), 'f', 2, 64)
+			data[i] = highMin + r.Float64()*(highBase)
 		} else if i >= lowStart && i < lowStart+lowSegmentLength {
 			// 数据值均小于 45.0，在 [40.0, 44.5] 范围内随机
-			data[i] = strconv.FormatFloat(lowMin+r.Float64()*(lowBase), 'f', 2, 64)
+			data[i] = lowMin + r.Float64()*(lowBase)
 		} else {
 			// 数据在 [45.0, 55.0] 范围内随机 (baseValue ± 5)
 			// 50 + rand.Float64() * 10 - 5
 			change := normalBase - r.Float64()*normalBase*2
-			data[i] = strconv.FormatFloat(baseValue+change, 'f', 2, 64)
+			data[i] = baseValue + change
 		}
 	}
 	return data
@@ -91,6 +98,10 @@ func main() {
 	log.Println("总共读取到测量点数量:", len(measurements))
 	measInfos := util.ProcessMeasurements(measurements)
 
+	globalRedisClient = util.InitRedisClient(redisAddr)
+	rCancelCtx, cancel := context.WithCancel(rootCtx)
+	defer cancel()
+
 	for key, measInfo := range measInfos {
 		var highMin, highBase float64
 		var lowMin, lowBase float64
@@ -107,7 +118,6 @@ func main() {
 			normalBase = changes[0]
 		} else {
 			randomIndex := selectRandomInt()
-			fmt.Println(randomIndex)
 			highMin = baseValue + changes[randomIndex]
 			lowMin = baseValue + changes[randomIndex+1]
 			highBase = changes[randomIndex]
@@ -115,42 +125,57 @@ func main() {
 			normalBase = changes[0]
 		}
 
-		datas := GenerateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase)
+		datas := generateMixedData(highMin, lowMin, highBase, lowBase, baseValue, normalBase)
+		// log.Printf("key:%s\n datas:%v\n", key, datas)
 
-		fmt.Printf("key:%s\n datas:%v\n", key, datas)
-		// 检查高值段是否满足 > 55
 		allHigh := true
-
 		for i := highStart; i < highEnd; i++ {
-			value, _ := strconv.ParseFloat(datas[i], 10)
-			if value <= highMin {
+			if datas[i] <= highMin {
 				allHigh = false
 				break
 			}
 		}
-		fmt.Printf("\n// 验证结果 (高值段在 %d-%d): 所有值是否 >  %.2f? %t\n", highStart, highEnd-1, highMin, allHigh)
+		log.Printf("\n// 验证结果 (高值段在 %d-%d): 所有值是否 >  %.2f? %t\n", highStart, highEnd-1, highMin, allHigh)
 
-		// 检查低值段是否满足 < 45
 		allLow := true
 		for i := lowStart; i < lowEnd; i++ {
-			value, _ := strconv.ParseFloat(datas[i], 10)
-			if value >= lowMin {
+			if datas[i] >= lowMin {
 				allLow = false
 				break
 			}
 		}
-		fmt.Printf("// 验证结果 (低值段在 %d-%d): 所有值是否 < %.2f? %t\n", lowStart, lowEnd-1, lowMin, allLow)
+		log.Printf("// 验证结果 (低值段在 %d-%d): 所有值是否 < %.2f? %t\n", lowStart, lowEnd-1, lowMin, allLow)
 
 		allTrue := true
 		for i := 0; i < totalLength-1; i++ {
-			value, _ := strconv.ParseFloat(datas[i], 10)
+			value := datas[i]
 			if i < highStart || (i >= highEnd && i < lowStart) || i >= lowEnd {
-				fmt.Printf("index:%d, value:%.2f\n", i, value)
+				log.Printf("index:%d, value:%.2f\n", i, value)
 				if value >= highMin && value <= lowMin {
 					allTrue = false
 				}
 			}
 		}
-		fmt.Printf("// 验证结果 (正常段在 %d-%d): 所有值是否 <= %.2f或>= %.2f? %t\n", 0, totalLength-1, highMin, lowMin, allTrue)
+		log.Printf("// 验证结果 (正常段在 %d-%d): 所有值是否 <= %.2f或>= %.2f %t\n", 0, totalLength-1, highMin, lowMin, allTrue)
+
+		log.Printf("启动数据写入程序, Redis Key: %s, 基准值: %.4f, 变化范围: %+v\n", key, measInfo.BaseValue, measInfo.Changes)
+
+		pipe := globalRedisClient.Pipeline()
+		redisZs := make([]redis.Z, 0, totalLength)
+		currentTime := time.Now().UnixNano()
+		for i := range totalLength {
+			sequentialTime := currentTime + int64(i)
+			z := redis.Z{
+				Score:  datas[i],
+				Member: strconv.FormatInt(sequentialTime, 10),
+			}
+			redisZs = append(redisZs, z)
+		}
+		log.Printf("启动数据写入程序, Redis Key: %s, 写入数据量: %d\n", key, len(redisZs))
+		pipe.ZAdd(rCancelCtx, key, redisZs...)
+		_, err = pipe.Exec(rCancelCtx)
+		if err != nil {
+			log.Printf("redis pipeline execution failed: %v\n", err)
+		}
 	}
 }

deploy/redis-test-data/real-time-subpull/sub_data_injection.go

@@ -22,28 +22,11 @@ import (
 
 // Redis配置
 const (
-	RedisAddr = "localhost:6379"
+	redisAddr = "localhost:6379"
 )
 
 var globalRedisClient *redis.Client
 
-func initRedisClient() *redis.Client {
-	rdb := redis.NewClient(&redis.Options{
-		Addr:     RedisAddr,
-		Password: "", // 如果有密码，请填写
-		DB:       0,  // 使用默认数据库
-	})
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	_, err := rdb.Ping(ctx).Result()
-	if err != nil {
-		return nil
-	}
-	return rdb
-}
-
 // outlierConfig 异常段配置
 type outlierConfig struct {
 	Enabled      bool    // 是否启用异常段
@@ -451,7 +434,7 @@ func main() {
 		Distribution: "both", // 分布类型
 	}
 
-	globalRedisClient = initRedisClient()
+	globalRedisClient = util.InitRedisClient(redisAddr)
 	rCancelCtx, cancel := context.WithCancel(rootCtx)
 	defer cancel()
 

deploy/redis-test-data/util/redis.go

@@ -0,0 +1,25 @@
+package util
+
+import (
+	"context"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func InitRedisClient(redisAddr string) *redis.Client {
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     redisAddr,
+		Password: "",
+		DB:       0,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := rdb.Ping(ctx).Result()
+	if err != nil {
+		return nil
+	}
+	return rdb
+}