func GetUser(ctx context.Context, id int64) (*User, error) {
    key := fmt.Sprintf("user:%d", id)

    // 1. 读缓存
    val, err := rdb.Get(ctx, key).Result()
    if err == nil {
        var u User
        json.Unmarshal([]byte(val), &u)
        return &u, nil
    }
    if err != redis.Nil {
        return nil, err     // Redis 故障，可以选择降级直接读 DB
    }

    // 2. 缓存未命中，读数据库
    u, err := db.GetUser(ctx, id)
    if err != nil {
        return nil, err
    }

    // 3. 回写缓存（带随机 TTL 抖动）
    data, _ := json.Marshal(u)
    ttl := time.Hour + time.Duration(rand.Intn(300))*time.Second
    rdb.Set(ctx, key, data, ttl)
    return u, nil
}

func UpdateUser(ctx context.Context, u *User) error {
    // 1. 先更新数据库
    if err := db.UpdateUser(ctx, u); err != nil {
        return err
    }
    // 2. 再删除缓存
    rdb.Del(ctx, fmt.Sprintf("user:%d", u.ID))
    return nil
}

// 高频写只落 Redis
rdb.HIncrBy(ctx, "article:views", articleID, 1)

// 定时任务（如每分钟）批量刷到数据库
func FlushViews() {
    views, _ := rdb.HGetAll(ctx, "article:views").Result()
    // 批量 UPDATE 到数据库
    db.BatchUpdateViews(views)
    // 注意：这里有并发问题，正确做法是用 HGETALL + DEL 的 Lua 或者按时间分 key
}

func GetUser(ctx context.Context, id int64) (*User, error) {
    key := fmt.Sprintf("user:%d", id)

    val, err := rdb.Get(ctx, key).Result()
    if err == nil {
        if val == "" || val == "NULL" {
            return nil, ErrNotFound      // 【命中空值缓存】，直接返回
        }
        var u User
        json.Unmarshal([]byte(val), &u)
        return &u, nil
    }

    u, err := db.GetUser(ctx, id)
    if err == sql.ErrNoRows {
        // 【关键】数据不存在也写缓存，但 TTL 要短（如 60 秒）
        rdb.Set(ctx, key, "NULL", 60*time.Second)
        return nil, ErrNotFound
    }
    if err != nil {
        return nil, err
    }

    data, _ := json.Marshal(u)
    rdb.Set(ctx, key, data, time.Hour+randomJitter())
    return u, nil
}

func CreateUser(ctx context.Context, u *User) error {
    if err := db.CreateUser(ctx, u); err != nil {
        return err
    }
    rdb.Del(ctx, fmt.Sprintf("user:%d", u.ID))   // 删掉可能存在的空值缓存
    return nil
}

# 安装模块后
BF.RESERVE users 0.001 100000000       # 误判率 0.1%，预期 1 亿元素
BF.ADD users 1001                       # 添加
BF.MADD users 1002 1003 1004            # 批量添加
BF.EXISTS users 1001                    # 1 = 可能存在，0 = 一定不存在
BF.MEXISTS users 1001 9999              # 批量查询
BF.INFO users                           # 查看容量、已插入数、误判率等

# 可扩展布隆过滤器（超出容量自动扩容，但误判率会上升）
BF.RESERVE users 0.001 1000000 EXPANSION 2
# NONSCALING 表示不扩容，满了报错

func GetUser(ctx context.Context, id int64) (*User, error) {
    // 【第一道防线】布隆过滤器
    exists, err := rdb.Do(ctx, "BF.EXISTS", "bf:users", id).Bool()
    if err == nil && !exists {
        return nil, ErrNotFound     // 【一定不存在，直接返回，不查 DB】
    }

    // 【第二道防线】空值缓存 + 正常缓存
    // ...（同 2.2）
}

// 新增数据时要加入布隆过滤器
func CreateUser(ctx context.Context, u *User) error {
    if err := db.CreateUser(ctx, u); err != nil {
        return err
    }
    rdb.Do(ctx, "BF.ADD", "bf:users", u.ID)
    rdb.Del(ctx, fmt.Sprintf("user:%d", u.ID))
    return nil
}

-- 添加：用多个哈希函数把 element 映射到 bitmap 的多个位
-- 查询：GETBIT 全为 1 才"可能存在"

// 1. 参数合法性校验（挡住明显非法的请求）
if id <= 0 || id > maxValidID {
    return nil, ErrInvalidParam
}

// 2. 对"缓存未命中"的请求做限流
// 正常业务的缓存命中率应该 > 95%，如果某个 IP/用户的未命中率异常高，
// 说明它在扫描不存在的数据 → 限流或拉黑
if missRate(clientIP) > 0.5 && missCount(clientIP) > 100 {
    return nil, ErrTooManyRequests
}

func GetHotData(ctx context.Context, id int64) (*Data, error) {
    key := fmt.Sprintf("data:%d", id)

    // 1. 先读缓存
    if val, err := rdb.Get(ctx, key).Result(); err == nil {
        return unmarshal(val), nil
    }

    // 2. 缓存未命中，尝试获取"重建锁"
    lockKey := "lock:rebuild:" + key
    token := uuid.NewString()
    ok, _ := rdb.SetNX(ctx, lockKey, token, 10*time.Second).Result()

    if ok {
        // 【拿到锁】：负责查库并回填缓存
        defer unlockScript.Run(ctx, rdb, []string{lockKey}, token)

        // 【重要】拿到锁后要再查一次缓存（Double Check）
        // 因为可能在等锁的过程中别人已经填好了
        if val, err := rdb.Get(ctx, key).Result(); err == nil {
            return unmarshal(val), nil
        }

        data, err := db.Get(ctx, id)
        if err != nil {
            return nil, err
        }
        rdb.Set(ctx, key, marshal(data), time.Hour+randomJitter())
        return data, nil
    }

    // 【没拿到锁】：三种策略选一种
    // 策略 A：短暂等待后重试读缓存（推荐）
    for i := 0; i < 5; i++ {
        time.Sleep(50 * time.Millisecond)
        if val, err := rdb.Get(ctx, key).Result(); err == nil {
            return unmarshal(val), nil
        }
    }
    return nil, ErrTimeout

    // 策略 B：直接返回旧值/默认值（降级，可用性最好）
    // 策略 C：直接查数据库（放弃保护，只在 DB 能承受时用）
}

import "golang.org/x/sync/singleflight"

var g singleflight.Group

func GetHotData(ctx context.Context, id int64) (*Data, error) {
    key := fmt.Sprintf("data:%d", id)

    if val, err := rdb.Get(ctx, key).Result(); err == nil {
        return unmarshal(val), nil
    }

    // 【进程内合并】：同一进程内的 N 个并发请求只有 1 个真正执行
    v, err, _ := g.Do(key, func() (interface{}, error) {
        // 这里再用分布式锁做跨进程的合并
        return getWithDistributedLock(ctx, key, id)
    })
    if err != nil {
        return nil, err
    }
    return v.(*Data), nil
}

type CacheValue struct {
    Data      json.RawMessage `json:"data"`
    ExpireAt  int64           `json:"expire_at"`   // 逻辑过期时间戳
}

func GetHotData(ctx context.Context, id int64) (*Data, error) {
    key := fmt.Sprintf("data:%d", id)

    val, err := rdb.Get(ctx, key).Result()
    if err == redis.Nil {
        // 缓存完全没有（首次访问或被淘汰）→ 同步查库（走 3.2 的互斥锁逻辑）
        return getWithLock(ctx, key, id)
    }
    if err != nil {
        return nil, err
    }

    var cv CacheValue
    json.Unmarshal([]byte(val), &cv)

    if time.Now().Unix() < cv.ExpireAt {
        // 【未逻辑过期】，直接返回
        return unmarshal(cv.Data), nil
    }

    // 【已逻辑过期】：尝试获取锁，异步重建
    lockKey := "lock:rebuild:" + key
    if ok, _ := rdb.SetNX(ctx, lockKey, 1, 10*time.Second).Result(); ok {
        go func() {
            defer rdb.Del(context.Background(), lockKey)
            data, err := db.Get(context.Background(), id)
            if err != nil {
                return
            }
            newCV := CacheValue{
                Data:     marshal(data),
                ExpireAt: time.Now().Add(time.Hour).Unix(),
            }
            // 【注意】物理上不设 TTL 或设很长的 TTL
            rdb.Set(context.Background(), key, marshal(newCV), 0)
        }()
    }

    // 【关键】无论是否拿到锁，都【立即返回旧数据】
    return unmarshal(cv.Data), nil
}

// 1. 活动开始前预热缓存
func WarmUp(ctx context.Context, hotIDs []int64) {
    for _, id := range hotIDs {
        data, _ := db.Get(ctx, id)
        rdb.Set(ctx, fmt.Sprintf("data:%d", id), marshal(data), 0)   // 不设 TTL
    }
}

// 2. 后台定时任务主动刷新（而不是等它过期）
func RefreshHotKeys(ctx context.Context) {
    ticker := time.NewTicker(5 * time.Minute)
    for range ticker.C {
        for _, id := range getHotIDs() {
            data, err := db.Get(ctx, id)
            if err == nil {
                rdb.Set(ctx, fmt.Sprintf("data:%d", id), marshal(data), 0)
            }
        }
    }
}

// ❌ 所有 key 的 TTL 完全相同
rdb.Set(ctx, key, val, time.Hour)

// ✅ 加随机抖动，把过期时间打散
func randomTTL(base time.Duration) time.Duration {
    // 基础 1 小时 + 0~10 分钟随机
    return base + time.Duration(rand.Intn(600))*time.Second
}
rdb.Set(ctx, key, val, randomTTL(time.Hour))

import "github.com/dgraph-io/ristretto"   // 或 bigcache / freecache / go-cache

var localCache *ristretto.Cache

func GetData(ctx context.Context, id int64) (*Data, error) {
    key := fmt.Sprintf("data:%d", id)

    // L1：本地缓存
    if v, found := localCache.Get(key); found {
        return v.(*Data), nil
    }

    // L2：Redis
    val, err := rdb.Get(ctx, key).Result()
    if err == nil {
        data := unmarshal(val)
        // 回填本地缓存，TTL 要【短】（因为无法及时感知失效）
        localCache.SetWithTTL(key, data, 1, 30*time.Second)
        return data, nil
    }

    // L3：数据库（用 singleflight + 分布式锁保护）
    data, err := getFromDBWithProtection(ctx, id)
    if err != nil {
        return nil, err
    }
    rdb.Set(ctx, key, marshal(data), randomTTL(time.Hour))
    localCache.SetWithTTL(key, data, 1, 30*time.Second)
    return data, nil
}

CLIENT TRACKING ON            # 开启键追踪（需要 RESP3）
# 客户端本地缓存 GET 到的数据
# 服务端在 key 被修改时【主动推送 invalidate 消息】给缓存了它的客户端

// 1. 限流：限制到数据库的并发量
var dbSemaphore = make(chan struct{}, 100)   // 最多 100 个并发查库

func getFromDB(ctx context.Context, id int64) (*Data, error) {
    select {
    case dbSemaphore <- struct{}{}:
        defer func() { <-dbSemaphore }()
        return db.Get(ctx, id)
    case <-time.After(50 * time.Millisecond):
        return nil, ErrSystemBusy       // 【快速失败，保护数据库】
    }
}

// 2. 熔断：数据库错误率过高时直接返回降级数据
// 用 sony/gobreaker 或 alibaba/sentinel-golang
var cb = gobreaker.NewCircuitBreaker(gobreaker.Settings{
    Name:        "db-query",
    MaxRequests: 10,
    Interval:    10 * time.Second,
    Timeout:     30 * time.Second,
    ReadyToTrip: func(counts gobreaker.Counts) bool {
        return counts.Requests >= 20 &&
            float64(counts.TotalFailures)/float64(counts.Requests) > 0.5
    },
})

result, err := cb.Execute(func() (interface{}, error) {
    return db.Get(ctx, id)
})
if err != nil {
    return getDefaultData(), nil    // 【降级：返回兜底数据】
}

// 服务启动时预热核心数据
func WarmUpCache(ctx context.Context) error {
    hotIDs, err := db.GetHotDataIDs(ctx, 10000)    // 取最热的 1 万条
    if err != nil {
        return err
    }

    // 分批预热，避免瞬间压垮数据库
    for i := 0; i < len(hotIDs); i += 100 {
        end := min(i+100, len(hotIDs))
        batch := hotIDs[i:end]

        items, _ := db.BatchGet(ctx, batch)
        pipe := rdb.Pipeline()
        for _, item := range items {
            pipe.Set(ctx, fmt.Sprintf("data:%d", item.ID),
                marshal(item), randomTTL(time.Hour))
        }
        pipe.Exec(ctx)

        time.Sleep(50 * time.Millisecond)      // 限速
    }
    return nil
}

func Update(ctx context.Context, u *User) error {
    if err := db.Update(ctx, u); err != nil {    // 1. 先更新 DB
        return err
    }
    rdb.Del(ctx, key(u.ID))                       // 2. 再删缓存
    return nil
}

func Update(ctx context.Context, u *User) error {
    rdb.Del(ctx, key(u.ID))                       // 1. 先删缓存
    if err := db.Update(ctx, u); err != nil {     // 2. 更新数据库
        return err
    }
    // 3. 【延迟】再删一次，清掉可能在步骤 1、2 之间被写入的旧值
    time.AfterFunc(500*time.Millisecond, func() {
        rdb.Del(context.Background(), key(u.ID))
    })
    return nil
}

rdb.Set(ctx, key, val, randomTTL(time.Hour))

func Update(ctx context.Context, u *User) error {
    if err := db.Update(ctx, u); err != nil {
        return err
    }

    if err := rdb.Del(ctx, key(u.ID)).Err(); err != nil {
        // 删除失败 → 丢进消息队列异步重试
        mq.Publish("cache-invalidate", key(u.ID))
        log.Errorf("删除缓存失败，已投递重试: %v", err)
    }
    return nil
}

// 低峰期跑，抽样比对缓存和数据库
func Reconcile(ctx context.Context) {
    for _, id := range sampleIDs(1000) {
        cached, _ := rdb.Get(ctx, key(id)).Result()
        actual, _ := db.Get(ctx, id)
        if cached != "" && !equal(cached, actual) {
            metrics.InconsistentCounter.Inc()
            rdb.Del(ctx, key(id))               // 修复
            log.Warnf("发现不一致: id=%d", id)
        }
    }
}

func rebuildCache(ctx context.Context, id int64) {
    data, _ := db.Master().Get(ctx, id)     // 【强制读主库】
    rdb.Set(ctx, key(id), marshal(data), randomTTL(time.Hour))
}

# 1. 采样扫描（走 SCAN，线上可用）
redis-cli --bigkeys                    # 每种类型的 Top 1
redis-cli --memkeys                    # 按实际内存排序（更准但更慢）
redis-cli --memkeys --memkeys-samples 0   # 0 = 扫全部（谨慎）

# 2. 单个 key 精确查看
MEMORY USAGE mykey
DEBUG OBJECT mykey                     # serializedlength 是序列化后的长度
STRLEN / HLEN / LLEN / SCARD / ZCARD mykey

# 3. 离线分析 RDB（最全面，不影响线上）
# rdb_bigkeys / redis-rdb-tools
rdb --command memory dump.rdb --bytes 10240 -f memory.csv

# ---------- 方案一：拆分（最根本）----------
# 一个 100 万字段的 hash → 按 hash(field) % 100 拆成 100 个 hash
HSET user:bucket:0 f1 v1
HSET user:bucket:1 f2 v2
# 好处：每个小 hash 保持 listpack 编码，内存还能省 5~10 倍（第 4 篇）

# 一个 100 万元素的 list → 按业务维度拆
LPUSH msg:user:1001 ...        # 而不是 LPUSH msg:all ...

# ---------- 方案二：只取需要的部分 ----------
HMGET bigHash f1 f2 f3         # 而不是 HGETALL
HSCAN bigHash 0 COUNT 100      # 分批遍历
LRANGE bigList 0 99            # 分页而不是 0 -1
ZRANGE bigZset 0 9             # Top N 而不是全量

# ---------- 方案三：压缩 value ----------
# 应用层用 gzip/snappy/zstd 压缩后再存（对 JSON 通常能压到 20%~30%）
# 代价：CPU 开销 + 无法在服务端做部分操作

# ---------- 方案四：不该放 Redis 的别放 ----------
# 大文件、图片、大文档 → 对象存储（OSS/S3）+ Redis 只存 URL

# ---------- 删除大 key ----------
UNLINK bigkey                  # 异步删除（4.0+）
# 或者分批删（更安全）
HSCAN + HDEL 批量
LTRIM 逐步截断
SPOP 批量弹出
ZREMRANGEBYRANK 批量删

# 1. Redis 自带（需要 maxmemory-policy 为 lfu 系列）
redis-cli --hotkeys

# 2. MONITOR 抓样本（线上慎用，有明显性能开销）
redis-cli monitor | head -10000 | awk '{print $4}' | sort | uniq -c | sort -rn | head

# 3. 从命令统计推断
INFO commandstats                # 看哪类命令调用最多
OBJECT FREQ mykey                # LFU 模式下的访问频率

# 4. 【推荐】客户端埋点统计
#    在应用层统计每个 key 的访问次数，上报到监控系统
#    这是最准确且无侵入的方式

# 5. 云厂商的热 key 分析
#    阿里云 Tair、腾讯云 Redis 都提供内置的热 key 探测（基于抽样）

func WarmUpSeckill(ctx context.Context, activityID, skuID int64, stock int) error {
    // 预热库存
    if err := rdb.Set(ctx, fmt.Sprintf("seckill:stock:%d", skuID),
        stock, 24*time.Hour).Err(); err != nil {
        return err
    }
    // 清理上次活动的用户集合
    rdb.Del(ctx, fmt.Sprintf("seckill:users:%d", skuID))
    return nil
}

-- KEYS[1] = 库存 key      KEYS[2] = 已购用户集合 key
-- ARGV[1] = 用户 ID       ARGV[2] = 购买数量
-- 返回：1 成功  -1 已购买过  -2 库存不足

-- 【所有校验在前】（Lua 不能回滚，见第 8 篇）
if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then
    return -1
end

local stock = tonumber(redis.call('GET', KEYS[1]))
if stock == nil or stock < tonumber(ARGV[2]) then
    return -2
end

-- 【所有写入在后】
redis.call('DECRBY', KEYS[1], ARGV[2])
redis.call('SADD', KEYS[2], ARGV[1])
return 1

stockKey := fmt.Sprintf("seckill:{%d}:stock", skuID)
usersKey := fmt.Sprintf("seckill:{%d}:users", skuID)
// 都用 skuID 做 hash tag，保证同 slot；且粒度足够细，不会倾斜（第 12 篇）

var seckillScript = redis.NewScript(`...上面的 Lua...`)

// 进程内的"已售完"标记，避免售完后还去打 Redis
var soldOut sync.Map

func Seckill(ctx context.Context, skuID, userID int64) error {
    // 1. 【本地缓存快速失败】：已售完直接返回
    if v, ok := soldOut.Load(skuID); ok && v.(bool) {
        return ErrSoldOut
    }

    stockKey := fmt.Sprintf("seckill:{%d}:stock", skuID)
    usersKey := fmt.Sprintf("seckill:{%d}:users", skuID)

    // 2. Redis 原子扣减
    result, err := seckillScript.Run(ctx, rdb,
        []string{stockKey, usersKey}, userID, 1).Int()
    if err != nil {
        return fmt.Errorf("秒杀失败: %w", err)
    }

    switch result {
    case -1:
        return ErrAlreadyBought
    case -2:
        soldOut.Store(skuID, true)      // 【标记售完，后续请求本地拦截】
        return ErrSoldOut
    }

    // 3. 扣减成功 → 【异步】创建订单
    msg := OrderMessage{SkuID: skuID, UserID: userID, Qty: 1}
    if err := mq.Publish(ctx, "seckill-order", msg); err != nil {
        // 【关键】发消息失败必须【回滚 Redis 扣减】，否则库存白扣了
        rollbackScript.Run(ctx, rdb, []string{stockKey, usersKey}, userID, 1)
        return fmt.Errorf("下单失败: %w", err)
    }

    return nil   // 立即返回"抢购成功，订单处理中"
}

-- 消息投递失败时回滚 Redis 的扣减
if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then
    redis.call('SREM', KEYS[2], ARGV[1])
    redis.call('INCRBY', KEYS[1], ARGV[2])
    return 1
end
return 0

func ConsumeOrder(ctx context.Context, msg OrderMessage) error {
    // 【幂等】：用消息 ID 或 (skuID, userID) 唯一索引防重复
    return db.Transaction(func(tx *gorm.DB) error {
        // 【最终兜底】：数据库层面的库存扣减，即使 Redis 出错也不会超卖
        result := tx.Exec(
            "UPDATE stock SET count = count - ? WHERE sku_id = ? AND count >= ?",
            msg.Qty, msg.SkuID, msg.Qty)
        if result.Error != nil {
            return result.Error
        }
        if result.RowsAffected == 0 {
            // 数据库库存不足（说明 Redis 和 DB 不一致，要告警）
            log.Errorf("DB 库存不足但 Redis 扣减成功: sku=%d", msg.SkuID)
            return ErrOutOfStock
        }

        // 创建订单（唯一索引 (sku_id, user_id) 保证一人一单）
        order := &Order{SkuID: msg.SkuID, UserID: msg.UserID, Qty: msg.Qty}
        if err := tx.Create(order).Error; err != nil {
            if isDuplicateKey(err) {
                return nil       // 【重复消息，幂等返回成功】
            }
            return err
        }
        return nil
    })
}

val, err := rdb.Get(ctx, key).Result()
if err != nil && err != redis.Nil {
    // Redis 故障 → 降级
    // 选项 A：直接查数据库（但要限流保护 DB！）
    // 选项 B：返回本地缓存的旧数据
    // 选项 C：返回默认值/兜底数据
    // 选项 D：直接返回错误（非核心功能）
}