优化缓存结构,优化缓存机制
This commit is contained in:
Vendored
+61
-2
@@ -2,6 +2,7 @@ package cache
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -9,7 +10,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
curveLockTTL = 10 * time.Second
|
curveLockTTL = 10 * time.Second
|
||||||
curveLockRetryDelay = 100 * time.Millisecond
|
curveLockRetryDelay = 100 * time.Millisecond
|
||||||
curveLockMaxRetries = 50
|
curveLockMaxRetries = 50
|
||||||
)
|
)
|
||||||
@@ -42,6 +43,64 @@ func (c *CurveCache) Set(ctx context.Context, brand, name, target, data string)
|
|||||||
return c.rdb.HSet(ctx, key, target, data).Err()
|
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),防止并发请求同一个曲线数据
|
// AcquireLock 获取分布式锁(SETNX),防止并发请求同一个曲线数据
|
||||||
func (c *CurveCache) AcquireLock(ctx context.Context, brand, name, target string) (bool, error) {
|
func (c *CurveCache) AcquireLock(ctx context.Context, brand, name, target string) (bool, error) {
|
||||||
lockKey := brand + " " + name + ":" + target + ":lock"
|
lockKey := brand + " " + name + ":" + target + ":lock"
|
||||||
@@ -85,4 +144,4 @@ func (c *CurveCache) GetWithLock(ctx context.Context, brand, name, target string
|
|||||||
|
|
||||||
// 未获取锁,等待后重试
|
// 未获取锁,等待后重试
|
||||||
return "", false, nil
|
return "", false, nil
|
||||||
}
|
}
|
||||||
|
|||||||
+108
-31
@@ -55,6 +55,26 @@ func (h *CurveHandler) ModelCurve(c *gin.Context) {
|
|||||||
|
|
||||||
ctx := c.Request.Context()
|
ctx := c.Request.Context()
|
||||||
|
|
||||||
|
// 快速路径:直接读取独立存储的 fr 数据(缓存命中时只需一次 Redis 调用)
|
||||||
|
frData, err := h.cache.GetFR(ctx, brand, name)
|
||||||
|
if err != nil {
|
||||||
|
h.log.Warn("get fr from cache failed", zap.Error(err))
|
||||||
|
}
|
||||||
|
|
||||||
|
if frData != "" {
|
||||||
|
h.log.Info("model curve fr cache hit", zap.String("brand", brand), zap.String("name", name))
|
||||||
|
var fr any
|
||||||
|
if err := json.Unmarshal([]byte(frData), &fr); err == nil {
|
||||||
|
h.writeCurveResponse(c, base64Resp, gin.H{
|
||||||
|
"code": 200,
|
||||||
|
"msg": "ok",
|
||||||
|
"fr": fr,
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 慢速路径:缓存未命中,走完整流程获取数据(同时会写入 __fr 缓存)
|
||||||
result, err := h.getCurvePoint(ctx, brand, name, defaultModelCurveTarget)
|
result, err := h.getCurvePoint(ctx, brand, name, defaultModelCurveTarget)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
h.log.Error("get model curve failed", zap.Error(err))
|
h.log.Error("get model curve failed", zap.Error(err))
|
||||||
@@ -76,7 +96,21 @@ func (h *CurveHandler) ModelCurve(c *gin.Context) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
h.writeCurveResponse(c, base64Resp, resp)
|
// 只返回 fr,不包含 parametric_eq
|
||||||
|
fr, hasFR := resp["fr"]
|
||||||
|
if !hasFR {
|
||||||
|
h.writeCurveResponse(c, base64Resp, gin.H{
|
||||||
|
"code": 0,
|
||||||
|
"msg": "无曲线数据",
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
h.writeCurveResponse(c, base64Resp, gin.H{
|
||||||
|
"code": 200,
|
||||||
|
"msg": "ok",
|
||||||
|
"fr": fr,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetCurve 获取目标曲线
|
// GetCurve 获取目标曲线
|
||||||
@@ -151,47 +185,90 @@ func (h *CurveHandler) writeCurveResponse(c *gin.Context, base64Resp bool, data
|
|||||||
}
|
}
|
||||||
|
|
||||||
// getCurvePoint 获取曲线数据:先查缓存,缓存不存在则请求 EQ 接口并缓存结果
|
// getCurvePoint 获取曲线数据:先查缓存,缓存不存在则请求 EQ 接口并缓存结果
|
||||||
|
// 缓存结构优化:fr 数据(与 target 无关)单独存储在 __fr 字段,避免每个 target 重复存储
|
||||||
func (h *CurveHandler) getCurvePoint(ctx context.Context, brand, name, target string) (string, error) {
|
func (h *CurveHandler) getCurvePoint(ctx context.Context, brand, name, target string) (string, error) {
|
||||||
data, acquired, err := h.cache.GetWithLock(ctx, brand, name, target)
|
// 使用 GetWithFR,命中缓存时自动合并 __fr 数据
|
||||||
|
data, err := h.cache.GetWithFR(ctx, brand, name, target)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
|
if data != "" {
|
||||||
// 缓存命中
|
|
||||||
if data != "" && !acquired {
|
|
||||||
h.log.Info("curve cache hit", zap.String("brand", brand), zap.String("name", name), zap.String("target", target))
|
h.log.Info("curve cache hit", zap.String("brand", brand), zap.String("name", name), zap.String("target", target))
|
||||||
return data, nil
|
return data, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// 获取了锁,缓存仍然为空,需要请求 EQ 接口
|
// 缓存未命中,获取分布式锁
|
||||||
if acquired {
|
acquired, lockErr := h.cache.AcquireLock(ctx, brand, name, target)
|
||||||
h.log.Info("curve cache miss, requesting eq api", zap.String("brand", brand), zap.String("name", name), zap.String("target", target))
|
if lockErr != nil {
|
||||||
result, eqErr := h.getCurvePointFromPEQ(ctx, brand, name, target)
|
return "", lockErr
|
||||||
if eqErr != nil {
|
|
||||||
// 释放锁
|
|
||||||
if lockErr := h.cache.ReleaseLock(ctx, brand, name, target); lockErr != nil {
|
|
||||||
h.log.Warn("release curve lock failed", zap.Error(lockErr))
|
|
||||||
}
|
|
||||||
return "", eqErr
|
|
||||||
}
|
|
||||||
// 释放锁
|
|
||||||
if lockErr := h.cache.ReleaseLock(ctx, brand, name, target); lockErr != nil {
|
|
||||||
h.log.Warn("release curve lock failed", zap.Error(lockErr))
|
|
||||||
}
|
|
||||||
|
|
||||||
if result != "" {
|
|
||||||
// 缓存结果
|
|
||||||
if cacheErr := h.cache.Set(ctx, brand, name, target, result); cacheErr != nil {
|
|
||||||
h.log.Warn("curve cache set failed", zap.Error(cacheErr))
|
|
||||||
}
|
|
||||||
return result, nil
|
|
||||||
}
|
|
||||||
return "", nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 未获取锁(其他请求正在处理),等待后重试
|
// 未获取锁(其他请求正在处理),等待后重试
|
||||||
time.Sleep(100 * time.Millisecond)
|
if !acquired {
|
||||||
return h.getCurvePoint(ctx, brand, name, target)
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
return h.getCurvePoint(ctx, brand, name, target)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 获取了锁,双重检查缓存
|
||||||
|
h.log.Info("curve cache miss, requesting eq api", zap.String("brand", brand), zap.String("name", name), zap.String("target", target))
|
||||||
|
data, err = h.cache.GetWithFR(ctx, brand, name, target)
|
||||||
|
if err != nil {
|
||||||
|
h.releaseLock(brand, name, target)
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
if data != "" {
|
||||||
|
h.releaseLock(brand, name, target)
|
||||||
|
return data, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// 请求 EQ 接口
|
||||||
|
result, eqErr := h.getCurvePointFromPEQ(ctx, brand, name, target)
|
||||||
|
if eqErr != nil {
|
||||||
|
h.releaseLock(brand, name, target)
|
||||||
|
return "", eqErr
|
||||||
|
}
|
||||||
|
|
||||||
|
if result == "" {
|
||||||
|
h.releaseLock(brand, name, target)
|
||||||
|
return "", nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// 将 fr 数据剥离,单独存储在 __fr 字段,避免每个 target 重复存储
|
||||||
|
dataToCache := result
|
||||||
|
var frToCache string
|
||||||
|
var parsed map[string]json.RawMessage
|
||||||
|
if err := json.Unmarshal([]byte(result), &parsed); err == nil {
|
||||||
|
if fr, ok := parsed["fr"]; ok {
|
||||||
|
frToCache = string(fr)
|
||||||
|
delete(parsed, "fr")
|
||||||
|
if stripped, err := json.Marshal(parsed); err == nil {
|
||||||
|
dataToCache = string(stripped)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 先写缓存,再释放锁(避免其他服务器在缓存写入前抢到锁后重复调用 EQ API)
|
||||||
|
if cacheErr := h.cache.Set(ctx, brand, name, target, dataToCache); cacheErr != nil {
|
||||||
|
h.log.Warn("curve cache set failed", zap.Error(cacheErr))
|
||||||
|
}
|
||||||
|
if frToCache != "" {
|
||||||
|
existingFR, _ := h.cache.Get(ctx, brand, name, "__fr")
|
||||||
|
if existingFR == "" {
|
||||||
|
if cacheErr := h.cache.Set(ctx, brand, name, "__fr", frToCache); cacheErr != nil {
|
||||||
|
h.log.Warn("curve fr cache set failed", zap.Error(cacheErr))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
h.releaseLock(brand, name, target)
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// releaseLock 释放分布式锁并记录警告日志
|
||||||
|
func (h *CurveHandler) releaseLock(brand, name, target string) {
|
||||||
|
if err := h.cache.ReleaseLock(context.Background(), brand, name, target); err != nil {
|
||||||
|
h.log.Warn("release curve lock failed", zap.Error(err))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// getCurvePointFromPEQ 从 EQ 接口获取曲线数据
|
// getCurvePointFromPEQ 从 EQ 接口获取曲线数据
|
||||||
|
|||||||
Reference in New Issue
Block a user