package cache import ( "context" "encoding/json" "fmt" "time" "github.com/redis/go-redis/v9" ) const ( curveLockTTL = 10 * time.Second curveLockRetryDelay = 100 * time.Millisecond curveLockMaxRetries = 50 ) // CurveCache 曲线数据缓存(Redis Hash) type CurveCache struct { rdb *redis.Client } func NewCurveCache(rdb *redis.Client) *CurveCache { return &CurveCache{rdb: rdb} } // Get 从 Redis Hash 中获取曲线缓存 func (c *CurveCache) Get(ctx context.Context, brand, name, target string) (string, error) { key := brand + " " + name val, err := c.rdb.HGet(ctx, key, target).Result() if err == redis.Nil { return "", nil } if err != nil { return "", fmt.Errorf("hget curve cache: %w", err) } return val, nil } // Set 将曲线数据存入 Redis Hash func (c *CurveCache) Set(ctx context.Context, brand, name, target, data string) error { key := brand + " " + name return c.rdb.HSet(ctx, key, target, data).Err() } // GetFR 直接获取独立存储的 fr 数据(用于 modelCurve 接口) func (c *CurveCache) GetFR(ctx context.Context, brand, name string) (string, error) { key := brand + " " + name val, err := c.rdb.HGet(ctx, key, "__fr").Result() if err == redis.Nil { return "", nil } if err != nil { return "", fmt.Errorf("hget curve fr: %w", err) } return val, nil } // GetWithFR 获取 target 缓存数据并合并独立存储的 fr 数据 // 兼容旧缓存:若 target 数据中已有 fr 字段则直接返回 func (c *CurveCache) GetWithFR(ctx context.Context, brand, name, target string) (string, error) { key := brand + " " + name targetData, err := c.rdb.HGet(ctx, key, target).Result() if err == redis.Nil { return "", nil } if err != nil { return "", fmt.Errorf("hget curve target: %w", err) } if targetData == "" { return "", nil } // 旧缓存兼容:已有 fr 字段无需合并 var check map[string]json.RawMessage if err := json.Unmarshal([]byte(targetData), &check); err == nil { if _, hasFR := check["fr"]; hasFR { return targetData, nil } } // 新缓存:从 __fr 字段读取 fr 数据并合并 frData, err := c.rdb.HGet(ctx, key, "__fr").Result() if err == redis.Nil || frData == "" { return targetData, nil } if err != nil { return targetData, nil } var result map[string]json.RawMessage if err := json.Unmarshal([]byte(targetData), &result); err != nil { return targetData, nil } result["fr"] = json.RawMessage(frData) merged, err := json.Marshal(result) if err != nil { return targetData, nil } return string(merged), nil } // AcquireLock 获取分布式锁(SETNX),防止并发请求同一个曲线数据 func (c *CurveCache) AcquireLock(ctx context.Context, brand, name, target string) (bool, error) { lockKey := brand + " " + name + ":" + target + ":lock" return c.rdb.SetNX(ctx, lockKey, "locked", curveLockTTL).Result() } // ReleaseLock 释放分布式锁 func (c *CurveCache) ReleaseLock(ctx context.Context, brand, name, target string) error { lockKey := brand + " " + name + ":" + target + ":lock" return c.rdb.Del(ctx, lockKey).Err() } // GetWithLock 获取缓存数据,缓存不存在时尝试加锁后重新获取 // 返回值: (data, acquired, error) // - data: 缓存数据(空字符串表示无数据) // - acquired: 是否成功获取锁(缓存不存在时需要加锁) func (c *CurveCache) GetWithLock(ctx context.Context, brand, name, target string) (data string, acquired bool, err error) { // 先查缓存 data, err = c.Get(ctx, brand, name, target) if err != nil { return "", false, err } if data != "" { return data, false, nil // 缓存命中 } // 缓存不存在,尝试加锁 acquired, err = c.AcquireLock(ctx, brand, name, target) if err != nil { return "", false, fmt.Errorf("acquire curve lock: %w", err) } if acquired { // 加锁成功,再次检查缓存(双重检查) data, err = c.Get(ctx, brand, name, target) if err != nil { return "", true, err } return data, true, nil } // 未获取锁,等待后重试 return "", false, nil }