package cache import ( "context" "errors" "time" "fwh-toolbox-server/cache/redisx" "go.uber.org/zap" ) // LoaderFunc 定义 type LoaderFunc[T any] func(ctx context.Context) (*T, error) // GetOrSet: cache-aside 模式,带分布式锁防击穿与异步写缓存 func GetOrSet[T any]( ctx context.Context, key string, loader LoaderFunc[T], opts ...Option, ) (*T, error) { o := defaultOptions() for _, op := range opts { op(o) } full := fullKey(o.prefix, key) // 1. 尝试缓存读取(JSON) var out T if err := redisx.GetJSON(ctx, full, &out); err == nil { onHit(full) return &out, nil } else { onMiss(full) } // 2. 未命中:尝试加分布式锁(如果启用) if o.lockEnabled { lockKey := full + ":lock" lock, _, err := redisx.TryLock(ctx, lockKey, o.lockTTL) if err == nil && lock != nil { // acquired lock defer func() { if err := lock.Release(ctx); err != nil { // log but don't break business zap.L().Error("cachex: failed to release lock", zap.String("key", lockKey), zap.String("err", err.Error())) } }() // double check cache (防止其它线程已写入) if err := redisx.GetJSON(ctx, full, &out); err == nil { onHit(full) return &out, nil } // 调用 loader 加载真实数据 data, err := loader(ctx) if err != nil { return nil, err } if data == nil { // 业务决定:我们不写空值,直接返回 not found return nil, errors.New("cachex: loader returned nil") } // 写入缓存(可异步) if o.async { go func() { if err := redisx.SetJSON(context.Background(), full, data, o.ttl); err != nil { zap.L().Error("cachex: async set failed", zap.String("full", full), zap.String("err", err.Error())) } }() } else { if err := redisx.SetJSON(ctx, full, data, o.ttl); err != nil { zap.L().Error("cachex: set failed", zap.String("full", full), zap.String("err", err.Error())) } } return data, nil } // 未能获得锁:改为短轮询等待缓存(其他 worker 可能正在写) start := time.Now() for { // 尝试从缓存读取 var r T if err := redisx.GetJSON(ctx, full, &r); err == nil { onHit(full) return &r, nil } if time.Since(start) >= o.waitForCache { // 超时则 fallback 直接调用 loader(避免长时间阻塞) break } time.Sleep(50 * time.Millisecond) } } // 最后兜底:直接调用 loader(不持锁) data, err := loader(ctx) if err != nil { return nil, err } if data == nil { return nil, errors.New("cachex: loader returned nil") } // 写入缓存(异步) if o.async { go func() { if err := redisx.SetJSON(context.Background(), full, data, o.ttl); err != nil { zap.L().Error("cachex: async set fallback failed", zap.String("full", full), zap.String("err", err.Error())) } }() } else { _ = redisx.SetJSON(ctx, full, data, o.ttl) } return data, nil } // GetOrSetMany: 批量模式,keys 与 loaderFunc 的约定:loader 需要知道哪些 keys 要加载 // 这里提供一个通用版本:loader 接收 keys 并返回 map[key]*T type BatchLoaderFunc[T any] func(ctx context.Context, keys []string) (map[string]*T, error) // GetOrSetMany: 先 MGET 缓存,找出缺失 keys,调用 loader 批量加载并回写缓存 func GetOrSetMany[T any]( ctx context.Context, keys []string, loader BatchLoaderFunc[T], opts ...Option, ) (map[string]*T, error) { o := defaultOptions() for _, op := range opts { op(o) } // prefix + full keys fullKeys := make([]string, len(keys)) for i, k := range keys { fullKeys[i] = fullKey(o.prefix, k) } // MGET via redisx.getClient().MGet out, missing, err := redisx.MGetJSON[T](ctx, fullKeys...) if err != nil { return nil, err } if len(missing) == 0 { // 全命中 return out, nil } // 批量 loader loaded, err := loader(ctx, missing) if err != nil { return nil, err } fullKeysLoaded := make(map[string]*T) for k, v := range loaded { // prefix + full keys fullKeysLoaded[fullKey(o.prefix, k)] = v out[k] = v } // 回写缓存(批量 MSET) if err := redisx.MSetJSON[T](ctx, fullKeysLoaded, o.ttl); err != nil { zap.L().Error("cachex: batch mset exec failed: %v", zap.String("err", err.Error())) } return out, nil }