高可用高性能缓存应用

// Cache-Aside 模式：应用先查缓存，缓存没有再查数据库，然后写入缓存

// GetUserByID 根据用户ID获取用户信息（Cache-Aside 模式）
// 参数：ctx 上下文，userID 用户ID
// 返回：用户信息或错误
func GetUserByID(ctx context.Context, userID int64) (*User, error) {
    // 第一步：构造缓存 key
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 第二步：先查 Redis 缓存
    cached, err := redisClient.Get(ctx, cacheKey).Result()
    if err == nil {
        // 缓存命中！直接返回（不需要查数据库）
        // Redis 命中率越高，数据库压力越小
        var user User
        json.Unmarshal([]byte(cached), &user)
        return &user, nil
    }

    // 第三步：缓存未命中，查询数据库
    user, err := db.QueryUser(ctx, userID)
    if err != nil {
        return nil, err
    }

    // 第四步：将数据库结果写入缓存（设置过期时间，防止数据一直不一致）
    // 这里设置 30 分钟过期，是 "最终一致性" 的体现
    userJSON, _ := json.Marshal(user)
    redisClient.Set(ctx, cacheKey, userJSON, 30*time.Minute)

    return user, nil
}

// Read-Through：应用只和缓存交互，缓存自己去查数据库
// Write-Through：写入时同时写缓存和数据库

// CacheStore 封装了缓存和数据库的协同操作
type CacheStore struct {
    redis    *redis.Client
    db       *sql.DB
}

// Get Read-Through 模式
func (s *CacheStore) Get(ctx context.Context, key string) (string, error) {
    // 先查缓存
    val, err := s.redis.Get(ctx, key).Result()
    if err == nil {
        return val, nil // 命中缓存
    }

    // 缓存未命中，由 CacheStore（而不是应用层）去查数据库
    val, err = s.db.QueryRow(ctx, "SELECT value FROM kv WHERE key = $1", key).String()
    if err != nil {
        return "", err
    }

    // 回填缓存
    s.redis.Set(ctx, key, val, 30*time.Minute)
    return val, nil
}

// Set Write-Through 模式
func (s *CacheStore) Set(ctx context.Context, key, value string) error {
    // 先写缓存
    err := s.redis.Set(ctx, key, value, 30*time.Minute).Err()
    if err != nil {
        return err
    }

    // 再写数据库
    _, err = s.db.Exec(ctx, "UPDATE kv SET value = $1 WHERE key = $2", value, key)
    return err
}

// Redis SDS 结构体（简化版）
struct sdshdr {
    int len;      // 已使用长度（记录字符串实际有多长）
    int free;     // 剩余可用空间（预分配的空间，减少内存重分配次数）
    char buf[];   // 实际存储数据的字符数组
};

// 为什么不用 C 的 char[]？
// 1. C 字符串获取长度要遍历 O(n)，SDS 直接读 len 字段 O(1)
// 2. C 字符串修改时可能溢出（没有边界检查），SDS 有空间预分配
// 3. C 字符串不能存二进制（遇到 \0 就截断），SDS 用 len 判断结束
// 4. SDS 修改字符串时，如果空间够就不需要重新分配内存（空间预分配策略）

// Go 中操作 Redis String 的示例

// 基本操作
val, err := rdb.Set(ctx, "name", "张三", 0).Result()  // 设置，0表示永不过期
val, err = rdb.Get(ctx, "name").Result()               // 获取

// 计数器操作（原子性，适合做点赞数、库存等）
rdb.Incr(ctx, "counter")         // +1
rdb.IncrBy(ctx, "counter", 10)   // +10
rdb.Decr(ctx, "counter")         // -1
rdb.DecrBy(ctx, "counter", 5)    // -5

// 设置过期时间
rdb.SetEX(ctx, "token", "abc123", 3600)  // 1小时后过期

// 不存在才设置（分布式锁的基础）
ok, err := rdb.SetNX(ctx, "lock:order:1", "owner1", 10*time.Second).Result()
// ok=true 表示设置成功（获取到锁）
// ok=false 表示 key 已存在（没抢到锁）

// 场景1：消息队列（生产者-消费者模式）

// 生产者：从左端推入消息
rdb.LPush(ctx, "task_queue", "task1")
rdb.LPush(ctx, "task_queue", "task2")

// 消费者：从右端弹出消息（阻塞式，队列空时等待）
// BRPop 会阻塞直到有消息或超时
result, err := rdb.BRPop(ctx, 30*time.Second, "task_queue").Result()
// result = ["task_queue", "task1"]（最先进去的先出来）

// 场景2：最新文章列表
// 每发一篇文章，用 LPUSH 加入列表，获取最新N篇用 LRANGE
rdb.LPush(ctx, "latest_posts", "article_id_100")
rdb.LPush(ctx, "latest_posts", "article_id_101")

// 获取最新的10篇文章
posts, _ := rdb.LRange(ctx, "latest_posts", 0, 9).Result()

// Hash 操作示例：存储用户信息

// 设置用户的不同字段
rdb.HSet(ctx, "user:1", "name", "张三")
rdb.HSet(ctx, "user:1", "age", "25")
rdb.HSet(ctx, "user:1", "city", "北京")

// 也可以批量设置
rdb.HMSet(ctx, "user:2", map[string]interface{}{
    "name": "李四",
    "age":  "30",
    "city": "上海",
})

// 获取单个字段
name, _ := rdb.HGet(ctx, "user:1", "name").Result()  // "张三"

// 获取所有字段
allFields, _ := rdb.HGetAll(ctx, "user:1").Result()
// map[string]string{"name":"张三", "age":"25", "city":"北京"}

// 场景：购物车
// key = cart:用户ID, field = 商品ID, value = 数量
rdb.HSet(ctx, "cart:1001", "product_001", "2")
rdb.HIncrBy(ctx, "cart:1001", "product_001", 1)  // 数量+1

// Set 操作示例

// 添加元素
rdb.SAdd(ctx, "tags:article1", "go", "redis", "cache")
rdb.SAdd(ctx, "tags:article2", "go", "database", "mysql")

// 求交集（共同标签）
common, _ := rdb.SInter(ctx, "tags:article1", "tags:article2").Result()
// ["go"]

// 求并集
all, _ := rdb.SUnion(ctx, "tags:article1", "tags:article2").Result()
// ["go", "redis", "cache", "database", "mysql"]

// 求差集（article1有但article2没有的）
diff, _ := rdb.SDiff(ctx, "tags:article1", "tags:article2").Result()
// ["redis", "cache"]

// 场景1：共同好友
rdb.SAdd(ctx, "friends:user1", "alice", "bob", "charlie")
rdb.SAdd(ctx, "friends:user2", "bob", "charlie", "david")
commonFriends, _ := rdb.SInter(ctx, "friends:user1", "friends:user2").Result()
// ["bob", "charlie"]

// 场景2：抽奖（随机弹出）
winner, _ := rdb.SPop(ctx, "lottery_participants").Result()

// ZSet 操作示例：排行榜

// 添加玩家分数
rdb.ZAdd(ctx, "game_rank", redis.Z{Score: 3500, Member: "player1"})
rdb.ZAdd(ctx, "game_rank", redis.Z{Score: 2800, Member: "player2"})
rdb.ZAdd(ctx, "game_rank", redis.Z{Score: 4200, Member: "player3"})
rdb.ZAdd(ctx, "game_rank", redis.Z{Score: 1500, Member: "player4"})

// 增加分数（原子操作）
rdb.ZIncrBy(ctx, "game_rank", 500, "player4")  // player4 从1500变成2000

// 获取排行榜 Top 3（按分数从高到低）
top3, _ := rdb.ZRevRangeWithScores(ctx, "game_rank", 0, 2).Result()
// [{player3 4200}, {player1 3500}, {player2 2800}]

// 获取某玩家的排名（从0开始）
rank, _ := rdb.ZRevRank(ctx, "game_rank", "player3").Result()
// 0（第一名）

// 获取某玩家的分数
score, _ := rdb.ZScore(ctx, "game_rank", "player3").Result()
// 4200

// Bitmap 示例：用户签到
// key = sign:用户ID:年月, offset = 日期-1, value = 1表示签到

// 用户1001在2024年1月5日签到
rdb.SetBit(ctx, "sign:1001:202401", 4, 1)  // offset从0开始，5日对应offset=4

// 检查是否签到
signed, _ := rdb.GetBit(ctx, "sign:1001:202401", 4).Result()
// signed = 1 表示已签到

// 统计本月签到次数
count, _ := rdb.BitCount(ctx, "sign:1001:202401", 0, -1).Result()
// count = 签到天数

// HyperLogLog 示例：统计UV
// 每个独立访客访问时
rdb.PFAdd(ctx, "uv:page:home", "user1", "user2", "user3")
// 获取独立访客数（有约0.81%误差，但只需12KB内存）
uv, _ := rdb.PFCount(ctx, "uv:page:home").Result()

// Go 中使用 Redis 事务（通过 Transaction 闭包）

// 场景：转账，从账户A扣钱，给账户B加钱
err := rdb.Watch(ctx, func(tx *redis.Tx) error {
    // 1. 读取账户A的余额
    balanceA, err := tx.Get(ctx, "account:A").Int()
    if err != nil {
        return err
    }

    // 2. 检查余额是否足够
    if balanceA < 100 {
        return errors.New("余额不足")
    }

    // 3. 执行事务：扣钱 + 加钱
    _, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
        pipe.DecrBy(ctx, "account:A", 100)   // A 扣100
        pipe.IncrBy(ctx, "account:B", 100)   // B 加100
        return nil
    })
    return err
}, "account:A")  // Watch account:A，如果事务期间被修改则自动重试

// WATCH 实现乐观锁的原理
// 1. WATCH key → 监视某个 key
// 2. MULTI → 开启事务
// 3. 命令入队
// 4. EXEC → 执行时检查 key 是否被修改过
//    - 如果没被修改 → 正常执行
//    - 如果被修改了 → 事务取消，返回 nil

// 伪代码说明 WATCH 的乐观锁效果：
// 时间线    事务A                      事务B
// T1      WATCH balance
// T2      MULTI
// T3      SET balance 200              SET balance 500
// T4      EXEC
//         → 检测到 balance 被事务B修改了
//         → 事务取消，返回 nil
//         → 应用层需要重试

// epoll 核心API（Linux 系统调用）

// 1. 创建 epoll 实例
int epfd = epoll_create(1);

// 2. 注册要监听的文件描述符（比如客户端连接）
struct epoll_event event;
event.events = EPOLLIN;       // 监听可读事件（有数据到达）
event.data.fd = client_fd;    // 记录是哪个客户端
epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event);

// 3. 等待事件发生（阻塞，直到有事件）
struct epoll_event events[100];
int n = epoll_wait(epfd, events, 100, -1);  // -1表示无限等待
// n = 有多少个事件发生

// 4. 遍历发生事件的fd，逐一处理
for (int i = 0; i < n; i++) {
    int fd = events[i].data.fd;
    // 读取该客户端的数据并处理
    read(fd, buffer, sizeof(buffer));
    // 执行Redis命令
    executeCommand(buffer);
}

// Redis 内部维护一个过期字典（expires dict）
// key → 过期时间戳（毫秒）

// 设置过期时间的命令
rdb.SetEX(ctx, "key1", "value", 60)           // 60秒后过期
rdb.Expire(ctx, "key2", 120)                   // 已存在的key设置120秒过期
rdb.ExpireAt(ctx, "key3", time.Now().Add(1*time.Hour))  // 在指定时间过期

// 查看剩余TTL
ttl, _ := rdb.TTL(ctx, "key1").Result()
// ttl = 剩余秒数（-1表示永不过期，-2表示key不存在）

// 取消过期时间（变成永久key）
rdb.Persist(ctx, "key1")

// Go 中使用 Redis Cluster

// 创建 Cluster 客户端
clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{
        ":7000",  // 节点A
        ":7001",  // 节点B
        ":7002",  // 节点C
        ":7003",  // 节点A的从节点
        ":7004",  // 节点B的从节点
        ":7005",  // 节点C的从节点
    },
    // 读写分离配置
    RouteRandomly: true,  // 读请求可以打到从节点
})

// 使用方式和单机完全一样！Cluster 客户端会自动路由
clusterClient.Set(ctx, "user:1", "张三", 0)
val, _ := clusterClient.Get(ctx, "user:1").Result()

// 如果 key 的槽位在 Node B 上，客户端会自动路由到 Node B
// 这个过程对应用透明（不需要手动指定节点）

// Hash Tag 示例：让相关 key 落到同一个槽

// 不加 tag：user:1 和 user:1:profile 可能在不同节点
// 加 tag {user1}：CRC16 只对大括号内的 user1 计算，保证同槽
clusterClient.Set(ctx, "user:{user1}", "基本信息", 0)
clusterClient.Set(ctx, "user:{user1}:profile", "详细资料", 0)
clusterClient.Set(ctx, "user:{user1}:orders", "订单列表", 0)

// 这三个 key 的 CRC16 都是 CRC16("user1") % 16384
// 所以一定在同一个节点上，可以做事务操作

// 方案1：缓存空值（最简单）

func GetUserByID(ctx context.Context, userID int64) (*User, error) {
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 查缓存
    cached, err := redisClient.Get(ctx, cacheKey).Result()
    if err == nil {
        if cached == "NULL" {
            // 缓存中存的是空值标记，说明数据库中不存在
            // 直接返回，不再查数据库
            return nil, ErrNotFound
        }
        var user User
        json.Unmarshal([]byte(cached), &user)
        return &user, nil
    }

    // 查数据库
    user, err := db.QueryUser(ctx, userID)
    if err == sql.ErrNoRows {
        // 数据库中不存在，缓存一个空值标记（短过期时间）
        // 过期时间设短一些（比如5分钟），防止之后新增了数据
        redisClient.Set(ctx, cacheKey, "NULL", 5*time.Minute)
        return nil, ErrNotFound
    }
    if err != nil {
        return nil, err
    }

    // 存在，写入缓存
    userJSON, _ := json.Marshal(user)
    redisClient.Set(ctx, cacheKey, userJSON, 30*time.Minute)
    return user, nil
}

// 方案2：布隆过滤器（更优雅，下一节详解）

func GetUserWithBloom(ctx context.Context, userID int64) (*User, error) {
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 第一步：先查布隆过滤器，判断 key 是否可能存在
    exists, err := bloomFilter.TestString(cacheKey)
    if !exists {
        // 布隆过滤器说不存在 → 一定不存在，直接返回
        // 这一层拦截了绝大多数无效请求
        return nil, ErrNotFound
    }

    // 布隆过滤器说可能存在 → 继续查缓存和数据库
    // ...（后续逻辑同上）
}

// 方案1：过期时间加随机值，避免同时过期

func CacheUser(ctx context.Context, user *User) error {
    cacheKey := fmt.Sprintf("user:%d", user.ID)
    userJSON, _ := json.Marshal(user)

    // 基础过期时间 30 分钟 + 随机 0~5 分钟
    // 这样不同 key 的过期时间分散开，不会同时失效
    baseTTL := 30 * time.Minute
    randomTTL := time.Duration(rand.Intn(300)) * time.Second  // 0~300秒随机
    redisClient.Set(ctx, cacheKey, userJSON, baseTTL+randomTTL)
    return nil
}

// 方案2：多级缓存（本地缓存 + Redis）

type MultiLevelCache struct {
    localCache *lru.Cache   // 本地缓存（进程内，极快）
    redis      *redis.Client // 分布式缓存
}

func (c *MultiLevelCache) Get(ctx context.Context, key string) (string, error) {
    // 第一级：查本地缓存（纳秒级）
    if val, ok := c.localCache.Get(key); ok {
        return val.(string), nil
    }

    // 第二级：查 Redis（毫秒级）
    val, err := c.redis.Get(ctx, key).Result()
    if err == nil {
        // 回填本地缓存（设置较短的本地TTL）
        c.localCache.Add(key, val)
        return val, nil
    }

    // 两级缓存都没有，查数据库
    // ...
    return "", ErrNotFound
}

// 方案3：熔断降级（Redis 宕机时保护数据库）

func GetWithCircuitBreaker(ctx context.Context, key string) (string, error) {
    // 如果 Redis 不可用，启用熔断器
    // 直接返回降级数据（默认值、空数据等），不查数据库
    if circuitBreaker.IsOpen() {
        return getDefaultValue(key), nil
    }

    val, err := redisClient.Get(ctx, key).Result()
    if err != nil {
        // Redis 异常，记录失败次数
        circuitBreaker.RecordFailure()
        return getDefaultValue(key), nil
    }

    circuitBreaker.RecordSuccess()
    return val, nil
}

// 方案1：互斥锁（只让一个请求查数据库，其他等待）

var (
    mu     sync.Mutex       // 互斥锁
    locks  sync.Map          // 每个key一把锁（避免全局锁影响其他key）
)

func GetHotKey(ctx context.Context, key string) (string, error) {
    // 先查缓存
    val, err := redisClient.Get(ctx, key).Result()
    if err == nil {
        return val, nil  // 缓存命中
    }

    // 缓存未命中，获取该key的互斥锁
    // 使用 sync.Map 为每个 key 创建独立的锁
    lockIface, _ := locks.LoadOrStore(key, &sync.Mutex{})
    lock := lockIface.(*sync.Mutex)
    lock.Lock()
    defer lock.Unlock()

    // 双重检查：拿到锁后再查一次缓存（可能其他请求已经重建了缓存）
    val, err = redisClient.Get(ctx, key).Result()
    if err == nil {
        return val, nil  // 别人已经重建好了
    }

    // 确实没有，查数据库
    val = queryFromDatabase(key)

    // 写回缓存（设置较长的过期时间）
    redisClient.Set(ctx, key, val, 1*time.Hour)

    // 清理锁
    locks.Delete(key)
    return val, nil
}

// 方案2：热点 key 永不过期（逻辑过期）

// 不给热点key设置TTL，而是在value中存一个逻辑过期时间
type CacheItem struct {
    Value      string    `json:"value"`
    ExpireTime time.Time `json:"expire_time"`  // 逻辑过期时间
}

func GetWithLogicalExpire(ctx context.Context, key string) (string, error) {
    cached, err := redisClient.Get(ctx, key).Result()
    if err != nil {
        return "", ErrNotFound
    }

    var item CacheItem
    json.Unmarshal([]byte(cached), &item)

    // 检查逻辑过期时间
    if time.Now().Before(item.ExpireTime) {
        // 未过期，直接返回
        return item.Value, nil
    }

    // 逻辑已过期，但缓存还在（物理未过期）
    // 异步重建缓存，当前请求返回旧数据（可接受的短暂不一致）
    go func() {
        // 加分布式锁，防止多个线程同时重建
        ok, _ := redisClient.SetNX(ctx, "lock:"+key, "1", 30*time.Second).Result()
        if ok {
            // 拿到锁，重建缓存
            newVal := queryFromDatabase(key)
            newItem := CacheItem{
                Value:      newVal,
                ExpireTime: time.Now().Add(1 * time.Hour),
            }
            newJSON, _ := json.Marshal(newItem)
            redisClient.Set(ctx, key, newJSON, 0)  // 0 = 永不物理过期
            redisClient.Del(ctx, "lock:"+key)
        }
    }()

    // 返回旧数据（虽然过期了，但还能用）
    return item.Value, nil
}

// 使用 Go 实现 Bloom Filter

import (
    "hash/fnv"
    "math"
)

// BloomFilter 布隆过滤器结构体
type BloomFilter struct {
    bitSet    []bool    // 位数组（实际应用中用 bitmap 更省内存）
    bitSize   uint      // 位数组大小
    hashCount uint      // Hash函数个数
}

// NewBloomFilter 根据预期元素数量和误判率创建布隆过滤器
// 参数：expectedItems 预期元素数量，falsePositiveRate 期望误判率（如0.01表示1%）
func NewBloomFilter(expectedItems uint, falsePositiveRate float64) *BloomFilter {
    // 计算需要的位数组大小
    // 公式：m = -n * ln(p) / (ln(2)²)
    // n=元素数量，p=误判率
    m := uint(math.Ceil(float64(expectedItems) * math.Abs(math.Log(falsePositiveRate)) / (math.Log(2) * math.Log(2))))

    // 计算需要的Hash函数个数
    // 公式：k = (m/n) * ln(2)
    k := uint(math.Ceil(float64(m) / float64(expectedItems) * math.Log(2)))

    return &BloomFilter{
        bitSet:    make([]bool, m),
        bitSize:   m,
        hashCount: k,
    }
}

// hashN 使用双Hash法生成第 i 个Hash值
// 原理：h_i(x) = (h1(x) + i * h2(x)) % m
// 这样只需要两个基础Hash函数，就能生成 k 个Hash值
func (bf *BloomFilter) hashN(data []byte, i uint) uint {
    // 第一个Hash函数：FNV-1a
    h1 := fnv.New32a()
    h1.Write(data)
    hash1 := h1.Sum32()

    // 第二个Hash函数：FNV-1
    h2 := fnv.New32()
    h2.Write(data)
    hash2 := h2.Sum32()

    // 双Hash法生成第 i 个Hash值
    return uint((uint(hash1) + uint(i)*uint(hash2)) % bf.bitSize)
}

// Add 向布隆过滤器中添加元素
func (bf *BloomFilter) Add(data []byte) {
    for i := uint(0); i < bf.hashCount; i++ {
        pos := bf.hashN(data, i)  // 计算第 i 个Hash位置
        bf.bitSet[pos] = true      // 将对应位置设为1
    }
}

// Test 检查元素是否可能存在
// 返回false → 一定不存在
// 返回true → 可能存在（有误判概率）
func (bf *BloomFilter) Test(data []byte) bool {
    for i := uint(0); i < bf.hashCount; i++ {
        pos := bf.hashN(data, i)
        if !bf.bitSet[pos] {
            // 只要有一个位置是0，就一定不存在
            return false
        }
    }
    // 所有位置都是1，可能存在（也可能是误判）
    return true
}

// 实际应用：防止缓存穿透

// 启动时预热：把数据库中所有用户ID加入布隆过滤器
func InitBloomFilter(ctx context.Context) {
    // 创建布隆过滤器：预期100万用户，误判率1%
    bf = NewBloomFilter(1000000, 0.01)

    // 从数据库加载所有用户ID
    userIDs, _ := db.GetAllUserIDs(ctx)
    for _, id := range userIDs {
        bf.Add([]byte(fmt.Sprintf("user:%d", id)))
    }
}

// 查询时先用布隆过滤器过滤
func GetUser(ctx context.Context, userID int64) (*User, error) {
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 布隆过滤器检查
    if !bf.Test([]byte(cacheKey)) {
        // 一定不存在，直接返回，不查缓存和数据库
        // 这一层拦截了99%以上的无效请求
        return nil, ErrNotFound
    }

    // 可能存在，继续查缓存
    cached, err := redisClient.Get(ctx, cacheKey).Result()
    if err == nil {
        // ... 返回缓存数据
    }

    // 查数据库
    // ...
}

// 延迟双删策略
func UpdateUserWithDoubleDelete(ctx context.Context, userID int64, newData *User) error {
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 第一次删除缓存
    redisClient.Del(ctx, cacheKey)

    // 更新数据库
    err := db.UpdateUser(ctx, userID, newData)
    if err != nil {
        return err
    }

    // 延迟一段时间后再次删除缓存
    // 目的：等读线程把旧值回填后，再删一次
    go func() {
        time.Sleep(1 * time.Second)  // 延迟时间需要大于"一次读请求耗时"
        redisClient.Del(ctx, cacheKey)
    }()

    return nil
}

// 删除缓存失败时，通过消息队列异步重试

func UpdateUserWithMQ(ctx context.Context, userID int64, newData *User) error {
    // 更新数据库
    err := db.UpdateUser(ctx, userID, newData)
    if err != nil {
        return err
    }

    // 删除缓存
    err = redisClient.Del(ctx, fmt.Sprintf("user:%d", userID)).Err()
    if err != nil {
        // 删除失败，发送消息到MQ，稍后重试
        // 消息内容包含：要删除的key
        mq.Publish("cache_delete_retry", map[string]interface{}{
            "key":     fmt.Sprintf("user:%d", userID),
            "retry":   0,  // 重试次数
        })
    }

    return nil
}

// MQ 消费者：处理删除重试
func ConsumeCacheDelete() {
    msgs := mq.Consume("cache_delete_retry")
    for msg := range msgs {
        key := msg["key"].(string)
        retry := msg["retry"].(int)

        err := redisClient.Del(ctx, key).Err()
        if err != nil {
            if retry < 3 {
                // 重试次数未超限，再次投递（带延迟）
                time.Sleep(time.Duration(retry+1) * time.Second)  // 指数退避
                mq.Publish("cache_delete_retry", map[string]interface{}{
                    "key":   key,
                    "retry": retry + 1,
                })
            }
            // 超过3次放弃（记录日志人工处理）
        }
    }
}

// 方案1：本地缓存（多级缓存）

type HotKeyCache struct {
    localCache sync.Map         // 进程内本地缓存
    redis      *redis.Client
    lock       sync.Mutex
}

func (c *HotKeyCache) Get(ctx context.Context, key string) (string, error) {
    // 第一级：查本地缓存（纳秒级，不走网络）
    if val, ok := c.localCache.Load(key); ok {
        return val.(string), nil
    }

    // 本地缓存没有，查 Redis
    val, err := c.redis.Get(ctx, key).Result()
    if err != nil {
        return "", err
    }

    // 回填本地缓存（短TTL，如5秒，保证数据新鲜度）
    c.localCache.Store(key, val)
    time.AfterFunc(5*time.Second, func() {
        c.localCache.Delete(key)  // 5秒后自动过期
    })

    return val, nil
}

// 方案2：拆分热点 key
// 把一个热点 key 拆成多个副本，分散到不同节点

func SetHotKey(ctx context.Context, key, value string) error {
    // 拆分成3个副本，分散到不同槽位
    for i := 0; i < 3; i++ {
        subKey := fmt.Sprintf("%s:%d", key, i)  // 如 product💯0
        redisClient.Set(ctx, subKey, value, 1*time.Hour)
    }
    return nil
}

func GetHotKey(ctx context.Context, key string) (string, error) {
    // 随机选择一个副本读取，分散压力
    i := rand.Intn(3)
    subKey := fmt.Sprintf("%s:%d", key, i)
    return redisClient.Get(ctx, subKey).Result()
}

# 方法1：redis-cli --bigkeys 命令（扫描采样）
redis-cli --bigkeys

# 方法2：redis-cli --memkeys 命令（Redis 7.0+，按内存排序）
redis-cli --memkeys

# 方法3：MEMORY USAGE 命令查看单个key的内存占用
redis-cli MEMORY USAGE mykey

// 方案1：拆分大 key
// 把一个大的 Hash 拆成多个小 Hash

// 拆分前：一个Hash存10万用户信息
// HSET users:user_info field1 value1 field2 value2 ...

// 拆分后：按用户ID取模拆成100个小Hash
func GetUserFromSplitHash(ctx context.Context, userID int64) (string, error) {
    // 根据 userID 取模决定存到哪个子 Hash
    shard := userID % 100
    hashKey := fmt.Sprintf("users:user_info:shard:%d", shard)
    field := fmt.Sprintf("user:%d", userID)
    return redisClient.HGet(ctx, hashKey, field).Result()
}

// 方案2：压缩大 value
// 把大的 JSON 用 gzip 压缩后再存
import "compress/gzip"

func SetCompressed(ctx context.Context, key, value string) error {
    // 压缩
    var buf bytes.Buffer
    gz := gzip.NewWriter(&buf)
    gz.Write([]byte(value))
    gz.Close()

    // 存压缩后的数据
    return redisClient.Set(ctx, key, buf.Bytes(), 0).Err()
}

// 方案3：异步删除大 key（Redis 4.0+）
// 不要用 DEL（阻塞），用 UNLINK（异步删除）

// 错误做法：DEL 大 key 会阻塞主线程
// redisClient.Del(ctx, "bigkey")  // 可能阻塞几秒！

// 正确做法：UNLINK 异步删除
redisClient.Do(ctx, "UNLINK", "bigkey")  // 后台异步删除，不阻塞

// 方案4：分批删除大 Hash 中的元素

func DeleteBigHash(ctx context.Context, hashKey string) error {
    // 每次删除100个field
    batchSize := 100

    for {
        // 使用 HSCAN 安全遍历（不会阻塞）
        fields, cursor, _ := redisClient.HScan(ctx, hashKey, 0, "", batchSize).Result()

        if len(fields) == 0 {
            break  // 没有元素了
        }

        // 批量删除这些 field
        if len(fields) > 0 {
            redisClient.HDel(ctx, hashKey, fields...)
        }

        if cursor == 0 {
            break  // 遍历完毕
        }
    }

    // 最后删除空 Hash 本身
    redisClient.Del(ctx, hashKey)
    return nil
}

// 基础版：使用 SETNX + 过期时间

func TryLockV1(ctx context.Context, key string, ttl time.Duration) (bool, error) {
    // SETNX：不存在才设置（原子操作）
    // 同时设置过期时间（防止死锁）
    ok, err := redisClient.SetNX(ctx, key, "locked", ttl).Result()
    if err != nil {
        return false, err
    }
    return ok, nil
}

func UnlockV1(ctx context.Context, key string) error {
    // 直接删除
    return redisClient.Del(ctx, key).Err()
}

// 问题：
// 1. 任何客户端都能解锁（不安全，A的锁可能被B解了）
// 2. 不支持可重入
// 3. 锁过期后其他客户端获取锁，原客户端还以为是自己持有

import "github.com/google/uuid"

// 安全版：用 UUID 标识锁的持有者

// TryLock 获取分布式锁
// 参数：key 锁名称，ttl 过期时间
// 返回：lockValue（解锁时用），是否获取成功
func TryLockV2(ctx context.Context, key string, ttl time.Duration) (string, bool, error) {
    // 生成唯一标识，证明"这把锁是我的"
    lockValue := uuid.New().String()

    // SET key value NX EX seconds
    // NX：不存在才设置（保证互斥）
    // EX：设置过期时间（防止死锁）
    ok, err := redisClient.SetNX(ctx, key, lockValue, ttl).Result()
    if err != nil {
        return "", false, err
    }

    if !ok {
        return "", false, nil  // 获取失败（锁被其他人持有）
    }

    return lockValue, true, nil
}

// Unlock 释放分布式锁
// 关键：必须用 Lua 脚本保证"判断+删除"是原子操作
func UnlockV2(ctx context.Context, key, lockValue string) error {
    // Lua 脚本：先判断锁是不是自己的，是才删除
    // 如果不用 Lua，GET+DEL 两步之间可能被其他人修改
    luaScript := `
        if redis.call("get", KEYS[1]) == ARGV[1] then
            -- 锁的值和传入的lockValue一致，说明是自己的锁
            return redis.call("del", KEYS[1])
        else
            -- 不是自己的锁，不能删
            return 0
        end
    `

    // 执行 Lua 脚本（Redis 保证 Lua 脚本的原子性）
    result, err := redisClient.Eval(ctx, luaScript, []string{key}, lockValue).Result()
    if err != nil {
        return err
    }

    if result.(int64) == 0 {
        return errors.New("解锁失败：锁已过期或不是你的锁")
    }
    return nil
}

// 锁续期：后台 goroutine 定期延长锁的过期时间

func TryLockWithWatchdog(ctx context.Context, key string, ttl time.Duration) (string, bool, error) {
    lockValue := uuid.New().String()

    ok, err := redisClient.SetNX(ctx, key, lockValue, ttl).Result()
    if err != nil || !ok {
        return "", false, err
    }

    // 启动看门狗 goroutine，定期续期
    go func() {
        ticker := time.NewTicker(ttl / 3)  // 每隔 TTL/3 续期一次
        defer ticker.Stop()

        for {
            select {
            case <-ctx.Done():
                // 上下文取消，停止续期
                return
            case <-ticker.C:
                // 续期：用 Lua 脚本保证原子性
                renewScript := `
                    if redis.call("get", KEYS[1]) == ARGV[1] then
                        return redis.call("pexpire", KEYS[1], ARGV[2])
                    else
                        return 0
                    end
                `
                result, err := redisClient.Eval(ctx, renewScript,
                    []string{key},
                    lockValue,
                    int64(ttl/time.Millisecond),
                ).Result()

                if err != nil || result.(int64) == 0 {
                    // 续期失败（锁已过期或被删除）
                    return
                }
                // 续期成功，继续等待下一次续期
            }
        }
    }()

    return lockValue, true, nil
}

// Redlock 简化实现

type RedLock struct {
    clients []*redis.Client  // 多个独立的 Redis 实例
    quorum  int               // 法定数量（半数以上）
}

func NewRedLock(addrs []string) *RedLock {
    clients := make([]*redis.Client, len(addrs))
    for i, addr := range addrs {
        clients[i] = redis.NewClient(&redis.Options{Addr: addr})
    }
    return &RedLock{
        clients: clients,
        quorum:   len(clients)/2 + 1,  // 半数以上
    }
}

func (rl *RedLock) TryLock(ctx context.Context, key string, ttl time.Duration) (string, bool, error) {
    lockValue := uuid.New().String()
    successCount := 0
    start := time.Now()

    // 向所有 Redis 实例发送加锁请求
    for _, client := range rl.clients {
        ok, err := client.SetNX(ctx, key, lockValue, ttl).Result()
        if err == nil && ok {
            successCount++
        }
    }

    // 计算实际加锁消耗的时间
    elapsed := time.Since(start)

    // 判断是否获取成功：
    // 1. 成功数超过半数
    // 2. 加锁耗时小于TTL（否则锁可能已过期）
    if successCount >= rl.quorum && elapsed < ttl {
        return lockValue, true, nil
    }

    // 获取失败，向所有实例发送解锁请求（清理）
    rl.Unlock(ctx, key, lockValue)
    return "", false, nil
}

func (rl *RedLock) Unlock(ctx context.Context, key, lockValue string) {
    // 向所有实例发送解锁请求（不管之前是否成功）
    luaScript := `
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
    `
    for _, client := range rl.clients {
        client.Eval(ctx, luaScript, []string{key}, lockValue)
    }
}