CL-Softwares
/
modelRT
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

optimize redis real time data injection func

This commit is contained in:
douxu 2025-11-21 17:02:07 +08:00
parent 36f267aec7
commit 5e311a7071
2 changed files with 126 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

300
deploy/redis-test-data/data_injection.go
View File

@ -6,6 +6,10 @@ import (
"fmt"
"log"
"math/rand"
"os"
"os/signal"
"strconv"
"syscall"
"time"
"modelRT/orm"
@ -17,10 +21,11 @@ import (
// Redis配置
const (
RedisAddr = "localhost:6379" // 您的Redis地址
KeyName = "telemetry:sensor_101" // 存储实时数据的 ZSet 键名
RedisAddr = "localhost:6379"
)
var globalRedisClient *redis.Client
func initRedisClient() *redis.Client {
rdb := redis.NewClient(&redis.Options{
Addr: RedisAddr,
@ -38,67 +43,17 @@ func initRedisClient() *redis.Client {
return rdb
}
// simulateDataWrite 定时生成并写入模拟数据到 Redis ZSet
func simulateDataWrite(ctx context.Context, rdb *redis.Client) {
// 设定写入频率：每秒写入一次
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
// 用于生成模拟数据的初始值
currentValue := 25.0
// 模拟数据变化的方向
direction := 0.1
for {
select {
case <-ctx.Done():
fmt.Printf("\n[%s] 写入程序已停止。\n", KeyName)
return // 收到停止信号，退出 goroutine
case <-ticker.C:
// 1. 生成新的模拟数据
currentValue += direction
if currentValue > 30.0 || currentValue < 20.0 {
direction *= -1 // 达到边界，反转变化方向
}
// 2. 准备写入数据
// ZSet的 Score 使用当前时间戳（秒级）
score := float64(time.Now().Unix())
// ZSet的 Member 使用值（转换为字符串）
// 在您的实时数据处理场景中，Member 通常存储值，Score 存储时间戳
memberValue := fmt.Sprintf("%.2f", currentValue)
z := redis.Z{
Score: score,
Member: memberValue,
}
// 3. 执行 ZADD 写入操作
err := rdb.ZAdd(ctx, KeyName, z).Err()
if err != nil {
log.Printf("写入 Redis 失败 (%s)：%v", KeyName, err)
} else {
fmt.Printf("[%s] 成功写入：时间戳=%.0f, 值=%s\n", KeyName, score, memberValue)
}
}
}
}
func processMeasurements(measurements []orm.Measurement) map[string]calculationResult {
results := make(map[string]calculationResult, len(measurements))
for _, measurement := range measurements {
// 检查 DataSource 是否存在且 type 为 1
if measurement.DataSource == nil {
fmt.Println(11111)
continue
}
// 检查 type 是否为 1
dataType, typeExists := measurement.DataSource["type"]
if !typeExists {
fmt.Println(22222)
continue
}
@ -116,20 +71,17 @@ func processMeasurements(measurements []orm.Measurement) map[string]calculationR
}
if typeValue != 1 {
fmt.Println(33333)
continue
}
// 获取 io_address
ioAddressRaw, ioExists := measurement.DataSource["io_address"]
if !ioExists {
fmt.Println(44444)
continue
}
ioAddress, ok := ioAddressRaw.(map[string]any)
if !ok {
fmt.Println(44444)
continue
}
@ -139,23 +91,19 @@ func processMeasurements(measurements []orm.Measurement) map[string]calculationR
result := fmt.Sprintf("%s:%s:phasor:%s", station, device, channel)
if measurement.EventPlan == nil {
fmt.Println(55555)
continue
}
causeValue, causeExist := measurement.EventPlan["cause"]
if !causeExist {
fmt.Println(66666)
continue
}
causeMap, ok := causeValue.(map[string]any)
if !ok {
fmt.Println(77777)
continue
}
calResult, err := calculateBaseValueEnhanced(causeMap)
if err != nil {
fmt.Println(88888, err)
continue
}
calResult.Size = measurement.Size
@ -381,8 +329,8 @@ func calculateBaseValueEnhanced(data map[string]any) (calculationResult, error)
}
}
// OutlierConfig 异常段配置
type OutlierConfig struct {
// outlierConfig 异常段配置
type outlierConfig struct {
Enabled bool // 是否启用异常段
Count int // 异常段数量 (0=随机, 1-5=指定数量)
MinLength int // 异常段最小长度
@ -397,7 +345,7 @@ type OutlierConfig struct {
// size: 生成的切片长度
// variationType: 变化类型
// outlierConfig: 异常段配置
func generateFloatSliceWithOutliers(baseValue float64, changes []float64, size int, variationType string, outlierConfig OutlierConfig) ([]float64, error) {
func generateFloatSliceWithOutliers(baseValue float64, changes []float64, size int, variationType string, outlierConfig outlierConfig) ([]float64, error) {
// 先生成正常数据
data, err := generateFloatSlice(baseValue, changes, size, variationType)
if err != nil {
@ -413,7 +361,7 @@ func generateFloatSliceWithOutliers(baseValue float64, changes []float64, size i
}
// 插入异常段
func insertOutliers(data []float64, baseValue float64, changes []float64, config OutlierConfig) []float64 {
func insertOutliers(data []float64, baseValue float64, changes []float64, config outlierConfig) []float64 {
if len(data) == 0 || !config.Enabled {
return data
}
@ -421,7 +369,7 @@ func insertOutliers(data []float64, baseValue float64, changes []float64, config
// 获取变化范围的边界
minBound, maxBound := getChangeBounds(baseValue, changes)
// TODO delete
fmt.Printf("获取变化范围的边界,min:%.4f,max:%.4f\n", minBound, maxBound)
log.Printf("获取变化范围的边界,min:%.4f,max:%.4f\n", minBound, maxBound)
// 确定异常段数量
outlierCount := config.Count
@ -439,7 +387,7 @@ func insertOutliers(data []float64, baseValue float64, changes []float64, config
// 生成异常段位置
segments := generateOutlierSegments(len(data), config.MinLength, config.MaxLength, outlierCount, config.Distribution)
// TODO 调试信息待删除
fmt.Printf("生成异常段位置:%+v\n", segments)
log.Printf("生成异常段位置:%+v\n", segments)
// 插入异常数据
for _, segment := range segments {
data = insertOutlierSegment(data, segment, minBound, maxBound, config)
@ -534,7 +482,7 @@ func generateOutlierSegments(totalSize, minLength, maxLength, count int, distrib
return segments
}
func insertOutlierSegment(data []float64, segment OutlierSegment, minBound, maxBound float64, config OutlierConfig) []float64 {
func insertOutlierSegment(data []float64, segment OutlierSegment, minBound, maxBound float64, config outlierConfig) []float64 {
rangeWidth := maxBound - minBound
// 确定整个异常段的方向
@ -660,29 +608,125 @@ func generateRandomData(baseValue float64, changes []float64, size int) []float6
return data
}
// simulateDataWrite 定时生成并写入模拟数据到 Redis ZSet
func simulateDataWrite(ctx context.Context, rdb *redis.Client, redisKey string, config outlierConfig, measInfo calculationResult) {
log.Printf("启动数据写入程序, Redis Key: %s, 基准值: %.4f, 变化范围: %+v\n", redisKey, measInfo.BaseValue, measInfo.Changes)
ticker := time.NewTicker(10 * time.Second)
defer ticker.Stop()
pipe := rdb.Pipeline()
for {
select {
case <-ctx.Done():
log.Printf("\n[%s] 写入程序已停止\n", redisKey)
return
case <-ticker.C:
minBound, maxBound := getChangeBounds(measInfo.BaseValue, measInfo.Changes)
log.Printf("计算边界: [%.4f, %.4f]\n", minBound, maxBound)
// 根据基准值类型决定如何处理
switch measInfo.BaseType {
case "TI":
// 边沿触发类型，生成特殊处理的数据
log.Printf("边沿触发类型，跳过异常数据生成\n")
return
case "TE":
// 正常上下限类型，生成包含异常的数据
if len(measInfo.Changes) == 0 {
log.Printf("无变化范围数据,跳过\n")
return
}
// 根据变化范围数量调整异常配置
if len(measInfo.Changes) == 2 {
// 只有上下限
config.Distribution = "both"
} else if len(measInfo.Changes) == 4 {
// 有上下限和预警上下限
config.Distribution = "both"
config.Intensity = 2.0 // 增强异常强度
}
// 生成包含异常的数据
data, err := generateFloatSliceWithOutliers(
measInfo.BaseValue,
measInfo.Changes,
measInfo.Size,
"random",
config,
)
if err != nil {
log.Printf("生成异常数据失败:%v\n", err)
continue
}
segments := detectOutlierSegments(data, measInfo.BaseValue, measInfo.Changes, config.MinLength)
log.Printf("检测到异常段数量:%d\n", len(segments))
for i, segment := range segments {
log.Printf("异常段%d: 位置[%d-%d], 长度=%d, 类型=%s\n",
i+1, segment.Start, segment.Start+segment.Length-1, segment.Length, segment.Type)
}
redisZs := make([]redis.Z, 0, len(data))
for i := range len(data) {
z := redis.Z{
Score: data[i],
Member: strconv.FormatInt(time.Now().UnixNano(), 10),
}
redisZs = append(redisZs, z)
}
pipe.ZAdd(ctx, redisKey, redisZs...)
_, err = pipe.Exec(ctx)
if err != nil {
log.Printf("redis pipeline execution failed: %v", err)
}
}
}
}
}
func gracefulShutdown() {
if globalRedisClient != nil {
if err := globalRedisClient.Close(); err != nil {
log.Printf("关闭 Redis 客户端失败:%v", err)
} else {
log.Println("关闭 Redis 客户端成功")
}
}
time.Sleep(500 * time.Millisecond)
os.Exit(0)
}
func main() {
ctx := context.Background()
rootCtx := context.Background()
pgURI := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s", "192.168.1.101", 5432, "postgres", "coslight", "demo")
db, err := gorm.Open(postgres.Open(pgURI))
postgresDBClient, err := gorm.Open(postgres.Open(pgURI))
if err != nil {
panic(err)
}
defer func() {
sqlDB, err := postgresDBClient.DB()
if err != nil {
panic(err)
}
sqlDB.Close()
}()
cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
cancelCtx, cancel := context.WithTimeout(rootCtx, 5*time.Second)
defer cancel()
var measurements []orm.Measurement
result := db.WithContext(cancelCtx).Find(&measurements)
result := postgresDBClient.WithContext(cancelCtx).Find(&measurements)
if result.Error != nil {
panic(result.Error)
}
fmt.Println("总共读取到测量点数量:", len(measurements))
log.Println("总共读取到测量点数量:", len(measurements))
measInfos := processMeasurements(measurements)
// 测量点数据生成(包含异常数据)
// 配置异常段参数
outlierConfig := OutlierConfig{
outlierConfig := outlierConfig{
Enabled: true, // 是否产生异常段数据
Count: 2, // 异常段数量
MinLength: 10, // 异常段最小连续长度
@ -691,112 +735,16 @@ func main() {
Distribution: "both", // 分布类型
}
// 为每个测量点生成包含异常的数据
generatedData := make(map[string][]float64)
detectedSegments := make(map[string][]OutlierSegment)
globalRedisClient = initRedisClient()
rCancelCtx, cancel := context.WithCancel(rootCtx)
defer cancel()
for key, measInfo := range measInfos {
fmt.Printf("\n处理测量点: %s\n", key)
fmt.Printf(" 基准值: %.4f, 变化范围: %v, 数据长度: %d, 类型: %s\n",
measInfo.BaseValue, measInfo.Changes, measInfo.Size, measInfo.BaseType)
minBound, maxBound := getChangeBounds(measInfo.BaseValue, measInfo.Changes)
fmt.Printf(" 计算边界: [%.4f, %.4f]\n", minBound, maxBound)
// 根据基准值类型决定如何处理
switch measInfo.BaseType {
case "TI":
// 边沿触发类型，生成特殊处理的数据
fmt.Printf(" 边沿触发类型，跳过异常数据生成\n")
continue
case "TE":
// 正常上下限类型，生成包含异常的数据
if len(measInfo.Changes) == 0 {
fmt.Printf(" 无变化范围数据，跳过\n")
continue
}
// 根据变化范围数量调整异常配置
config := outlierConfig
if len(measInfo.Changes) == 2 {
// 只有上下限
config.Distribution = "both"
} else if len(measInfo.Changes) == 4 {
// 有上下限和预警上下限
config.Distribution = "both"
config.Intensity = 2.0 // 增强异常强度
}
// 生成包含异常的数据
data, err := generateFloatSliceWithOutliers(
measInfo.BaseValue,
measInfo.Changes,
measInfo.Size,
"random",
config,
)
if err != nil {
fmt.Printf(" 生成异常数据失败: %v\n", err)
continue
}
// 保存生成的数据
generatedData[key] = data
// 检测异常段
segments := detectOutlierSegments(data, measInfo.BaseValue, measInfo.Changes, config.MinLength)
detectedSegments[key] = segments
fmt.Printf(" 成功生成数据,长度: %d\n", len(data))
fmt.Printf(" 检测到异常段数量: %d\n", len(segments))
// 显示异常段详情
for i, segment := range segments {
fmt.Printf(" 异常段%d: 位置[%d-%d], 长度=%d, 类型=%s\n",
i+1, segment.Start, segment.Start+segment.Length-1, segment.Length, segment.Type)
}
}
go simulateDataWrite(rCancelCtx, globalRedisClient, key, outlierConfig, measInfo)
}
// 统计信息
fmt.Println("\n=== 生成结果统计 ===")
fmt.Printf("成功处理的测量点数量: %d/%d\n", len(generatedData), len(measInfos))
totalOutliers := 0
for key, segments := range detectedSegments {
totalOutliers += len(segments)
fmt.Printf(" %s: %d个异常段\n", key, len(segments))
}
fmt.Printf("总共检测到异常段: %d\n", totalOutliers)
// 示例：显示特定测量点的详细数据
if len(generatedData) > 0 {
fmt.Println("\n=== 示例数据详情 ===")
// 取第一个测量点显示详细数据
for key, data := range generatedData {
fmt.Printf("测量点: %s\n", key)
fmt.Printf("前30个数据值:\n")
minBound, maxBound := getChangeBounds(measInfos[key].BaseValue, measInfos[key].Changes)
for i := 0; i < min(30, len(data)); i++ {
status := "正常"
if data[i] > maxBound {
status = "↑异常"
} else if data[i] < minBound {
status = "↓异常"
}
fmt.Printf(" [%d] %.4f (%s)\n", i, data[i], status)
}
break // 只显示第一个
}
}
// Redis写入代码
// rdb := initRedisClient()
// gCtx, cancel := context.WithCancel(ctx)
// defer cancel()
// go simulateDataWrite(gCtx, rdb)
// TODO 添加singal监听，优雅退出
// cancel()
// time.Sleep(2 * time.Second)
// fmt.Println("程序已退出。")
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
<-sigChan
gracefulShutdown()
}

3
util/token.go
View File

@ -7,6 +7,7 @@ import (
"encoding/base64"
"fmt"
"os"
"strconv"
"strings"
"time"
)
@ -22,7 +23,7 @@ func GenerateClientToken(host string, serviceName string, secretKey string) (str
return "", fmt.Errorf("TOKEN_SECRET_KEY environment variable not set and no key provided in parameters")
}
uniqueID := fmt.Sprintf("%d", time.Now().UnixNano())
uniqueID := strconv.FormatInt(time.Now().UnixNano(), 10)
clientInfo := fmt.Sprintf("host=%s;service=%s;id=%s", host, serviceName, uniqueID)
mac := hmac.New(sha256.New, []byte(finalSecretKey))