Redis 全系列：数据类型、分布式锁与缓存

学习目标

学完本章你应该能够：

1. 用自己的话讲清 Redis 五种基本数据类型的底层结构，画得出跳表的多层链表示意图，并解释 ZSet 为什么用跳表而不用红黑树。
2. 从零实现一套完整的 Redis 分布式锁——从 SETNX 的缺陷到 SET NX PX + Lua 脚本释放，再到 Redlock 算法，能讲清每一步为什么要这么做。
3. 画图区分缓存穿透、缓存击穿、缓存雪崩三者的触发场景与解决方案，面试时能口述完整的防御方案。
4. 说清 Redis 单线程为什么快、6.0 多线程改了什么，以及 RDB/AOF/混合持久化的取舍。
5. 理解主从复制、哨兵、Cluster 三种高可用架构的原理与适用场景，能根据业务量级做出选型决策。

前置知识：Go 基本语法（goroutine、channel、sync 包）、基本数据结构（链表、哈希表、树）、Linux 进程与 IO 模型的概念、HTTP 缓存的基本认知。

本章你会动手做的事：

1. 用 Go + go-redis 写一个自旋分布式锁，故意构造"锁误删"场景（A 的锁被 B 删掉），再用 Lua 脚本修复。
2. 用 Redis ZSet 做一个排行榜 Demo，插入 10 万条数据后观察 ZREVRANGE 的响应时间。
3. 启动一个三节点的 Redis Cluster（docker-compose），用 redis-cli 观察 CRC16 槽位路由过程。

一、Redis 数据类型与底层结构

1.1 用生活类比先建立直觉

类比：把 Redis 想象成一个超级智能的"存钱罐仓库"。这个仓库有五个不同的房间，每个房间存东西的方式不一样：

• String 房间——像一个个保险箱，每个保险箱里放一份东西，可以是钱（数字）、也可以是纸条（文本）。最简单也最常用。
• List 房间——像一排传送带，东西从左边放进去、右边出来（或反过来），保持顺序。适合排队。
• Hash 房间——像一个大文件柜，一个柜子（key）里面有多个抽屉（field），每个抽屉存一个属性值。适合存"一个对象"。
• Set 房间——像一个俱乐部会员名单，每个人只出现一次，没有重复。适合做"标签"“共同好友”。
• ZSet 房间——像俱乐部会员名单 + 每人身上挂一个分数牌，按分数从高到低排好队。这就是排行榜。

其中最特别的是 ZSet——它既要保证元素不重复（像 Set），又要按分数排序。Redis 底层用了一个叫"跳表"的数据结构来实现它。

跳表的核心思想是"空间换时间"：想象一条很长的走廊，每隔几米有一扇门。你从走廊头走到走廊尾要很久。但如果在走廊上方搭了几层"天桥"，天桥上只有部分门（每隔 2 个、4 个、8 个），你先在天桥上大步走，快到目标时再下到走廊细找——这就是"跳跃式查找"。

flowchart TB
    subgraph L3["第3层 稀疏索引"]
        H3["Head"] --> N3A["10"] --> N3B["50"] --> T3["Tail"]
    end
    subgraph L2["第2层 中间索引"]
        H2["Head"] --> N2A["10"] --> N2B["30"] --> N2C["50"] --> T2["Tail"]
    end
    subgraph L1["第1层 完整链表"]
        H1["Head"] --> N1A["10"] --> N1B["20"] --> N1C["30"] --> N1D["40"] --> N1E["50"] --> T1["Tail"]
    end
    N3A -.-> N2A
    N3B -.-> N2C
    N2A -.-> N1A
    N2B -.-> N1C
    N2C -.-> N1E

上面这张图就是跳表的核心结构。第 1 层是完整的有序链表，包含所有节点。第 2 层每隔一个节点抽一个，第 3 层更稀疏。查找时从最高层开始，大步跳跃前进，到了目标附近就下降一层——每下降一层就更精确。最终在第 1 层找到目标。整个过程像在二分查找，但不需要数组的随机访问能力——纯链表就能做到 O(logN)。

对应到工程里就是：Redis 的 ZSet 底层用 skiplist（跳表）+ hashtable（哈希表）组合实现。跳表负责按分数范围查询，哈希表负责 O(1) 查找某个成员的分数。

1.2 工程要点

五种基本类型与底层结构

⚠️ 新手必踩的坑： 很多人以为 List 的底层就是普通双向链表。实际上 Redis 3.2 之后用的是 quicklist——每个节点是一个 ziplist（压缩列表），多个 ziplist 用双向指针串起来。这样既避免了普通链表指针占内存过多的问题，又保留了头尾快速操作的优势。

SDS 为什么不用 C 字符串

Redis 自己实现了 SDS（Simple Dynamic String）而不是用 C 原生的 char[]，原因有三个：

// SDS 结构示意（简化版）
struct sdshdr {
    int len;      // 已使用长度
    int free;     // 剩余可用空间
    char buf[];   // 实际存储字符的数组
};

• O(1) 获取长度：SDS 记录了 len，直接读；C 字符串要遍历到 \0 才知道长度，是 O(N)。
• 二进制安全：C 字符串用 \0 判断结尾，存图片/二进制数据时遇到 \0 会截断；SDS 用 len 判断长度，什么都能存。
• 预分配与惰性释放：修改字符串时，SDS 会预分配多余空间（小于 1MB 翻倍，大于 1MB 多分配 1MB），减少频繁 realloc。

跳表原理详解

跳表的核心操作是查找。以查找值为 40 的节点为例，步骤如下：

flowchart TD
    S1["步骤1：从第3层Head开始
向右看到10，比40小，跳到10"] --> S2["步骤2：第3层10的下一个是50
比40大，下降到第2层"]
    S2 --> S3["步骤3：第2层10的下一个是30
比40小，跳到30"] --> S4["步骤4：第2层30的下一个是50
比40大，下降到第1层"]
    S4 --> S5["步骤5：第1层30的下一个是40
等于目标，找到！"]

整个查找过程经过了 5 次比较，而如果直接在第 1 层遍历需要 4 次比较（10→20→30→40）。数据量小时优势不明显，但当链表有 100 万个节点时，跳表只需要约 20 次比较（log2(1000000) ≈ 20），而遍历需要 100 万次。

⚠️ 新手必踩的坑： 跳表的层数不是固定的。每个新节点插入时，会通过"抛硬币"式的随机算法决定它出现在哪些层——50% 概率出现在第 2 层，25% 出现在第 3 层，以此类推。这意味着跳表的结构是随机的，但期望高度是 O(logN)，所以查询效率稳定在 O(logN)。

为什么 ZSet 用跳表不用红黑树

面试高频问题。Redis 作者 antirez 自己解释过，原因有三点：

跳表的唯一"缺点"是每个节点有多个指针，内存占用比红黑树稍多。但 Redis 是内存数据库，这点开销在工程上完全可以接受。

Go 操作 Redis 基本类型示例

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
)

func main() {
    // 步骤1：创建 Redis 客户端连接
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
        PoolSize: 20,
    })
    ctx := context.Background()

    // 步骤2：String —— 缓存 + 计数器
    rdb.Set(ctx, "user:1:name", "张三", 10*time.Minute)
    rdb.Incr(ctx, "page:home:views") // 原子自增，适合做计数器

    // 步骤3：Hash —— 存储对象
    rdb.HSet(ctx, "user:1", "name", "张三", "age", 28, "city", "北京")

    // 步骤4：Set —— 标签 / 共同好友
    rdb.SAdd(ctx, "user:1:tags", "golang", "redis", "docker")
    rdb.SAdd(ctx, "user:2:tags", "golang", "mysql", "docker")
    // 求交集：共同标签
    common, _ := rdb.SInter(ctx, "user:1:tags", "user:2:tags").Result()
    fmt.Println("共同标签:", common) // [golang docker]

    // 步骤5：ZSet —— 排行榜
    rdb.ZAdd(ctx, "game:ranking", redis.Z{Score: 100, Member: "player1"})
    rdb.ZAdd(ctx, "game:ranking", redis.Z{Score: 200, Member: "player2"})
    rdb.ZAdd(ctx, "game:ranking", redis.Z{Score: 150, Member: "player3"})
    // 取前三名（分数从高到低）
    top3, _ := rdb.ZRevRangeWithScores(ctx, "game:ranking", 0, 2).Result()
    for i, v := range top3 {
        fmt.Printf("第%d名: %s 分数: %.0f\n", i+1, v.Member, v.Score)
    }
}

⚠️ 新手必踩的坑： ZRevRange 是从高到低排，ZRange 是从低到高排。做排行榜时千万别用反了——用 ZRange 取出来的第一名是最低分。

二、Redis 分布式锁

2.1 用生活类比先建立直觉

类比：想象一个公共卫生间，门上有一把锁。你想用卫生间，需要做以下几件事：

1. 尝试锁门（SET NX）：如果门没锁，你锁上，进去用。如果门已经锁了，你等一会儿再试（自旋）。
2. 防止永久占用（PX 过期）：万一你进去了突然晕倒，门一直锁着，别人永远用不了。所以锁会自动过期——比如 30 秒后自动开锁。
3. 防止开错门（唯一 value）：你用完了出来开门，但万一这时候锁已经过期了、另一个人已经重新锁上门了，你一开门把别人的锁打开了！所以你开门前要先确认"这把锁是不是我锁的"——这就是用唯一标识（value）来校验。

对应到工程里：SET key value NX PX 就是"尝试锁门 + 自动过期"，唯一 value 就是"我的身份证号"，Lua 脚本释放锁就是"先看身份证再开门"。

flowchart TD
    Start["客户端A请求加锁"] --> Set["SET lock_key uuid NX PX 30000"]
    Set -->|"返回OK"| Hold["加锁成功
持有锁"]
    Set -->|"返回nil"| Retry["锁已被占用
自旋等待"]
    Retry --> Sleep["sleep 100ms 后重试"]
    Sleep --> Set
    Hold --> Work["执行业务逻辑"]
    Work --> Release["执行Lua脚本释放锁"]
    Release --> Check{"校验value
是否为自己加的锁?"}
    Check -->|"是"| Del["DEL key
释放成功"]
    Check -->|"否"| Skip["不删除
防止误删别人的锁"]

这张图就是分布式锁的完整生命周期：加锁→自旋重试→执行业务→安全释放。每个环节都有对应的坑，下面逐一拆解。

2.2 工程要点

SETNX vs SET NX 的区别

⚠️ 新手必踩的坑： 用 SETNX 加锁后，如果紧接着用 EXPIRE 设过期时间，这两步不是原子的。如果 SETNX 成功但 EXPIRE 之前进程崩溃了，锁就永远不会过期——别人永远拿不到锁。所以永远用 SET key value NX PX 30000，一条命令原子完成加锁+设过期。

自旋锁实现（Go 代码）

package main

import (
    "context"
    "errors"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/google/uuid"
)

// SpinLock 自旋分布式锁
type SpinLock struct {
    client   *redis.Client
    key      string
    value    string // 唯一标识，防止误删
    ttl      time.Duration
    interval time.Duration // 自旋间隔
}

func NewSpinLock(client *redis.Client, key string, ttl time.Duration) *SpinLock {
    return &SpinLock{
        client:   client,
        key:      key,
        value:    uuid.New().String(), // 步骤1：生成唯一标识
        ttl:      ttl,
        interval: 100 * time.Millisecond,
    }
}

// Lock 自旋加锁，直到成功或超时
func (l *SpinLock) Lock(ctx context.Context, timeout time.Duration) error {
    deadline := time.Now().Add(timeout)
    for {
        // 步骤2：尝试加锁，SET key value NX PX ttl
        ok, err := l.client.SetNX(ctx, l.key, l.value, l.ttl).Result()
        if err != nil {
            return err
        }
        if ok {
            return nil // 加锁成功
        }
        // 步骤3：加锁失败，检查是否超时
        if time.Now().After(deadline) {
            return errors.New("lock timeout")
        }
        // 步骤4：等待一段时间后重试
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(l.interval):
            // 继续自旋
        }
    }
}

// Unlock 安全释放锁（用Lua脚本保证原子性）
func (l *SpinLock) Unlock(ctx context.Context) error {
    // 步骤5：Lua脚本——先校验value再删除，保证原子性
    script := `
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
    `
    result, err := l.client.Eval(ctx, script, []string{l.key}, l.value).Int()
    if err != nil {
        return err
    }
    if result == 0 {
        return errors.New("unlock failed: lock not owned by this client")
    }
    return nil
}

⚠️ 新手必踩的坑： 释放锁时绝对不能直接 DEL key！设想这个场景：A 加了锁，执行业务时间太长导致锁过期自动释放，B 此时加锁成功，A 执行完业务后直接 DEL——把 B 的锁删了，C 也能加锁，锁彻底失效。必须用 Lua 脚本"先 GET 校验 value，再 DEL"——这两步在 Lua 脚本中是原子执行的。

完整版分布式锁的核心问题清单

可重入分布式锁思路

可重入锁的核心是"同一个线程/客户端可以多次获取同一把锁"。实现方式是用 Redis Hash 结构：

-- 加锁Lua脚本（可重入版）
-- KEYS[1] = lock_key
-- ARGV[1] = uuid（客户端标识）
-- ARGV[2] = thread_id（线程标识）
-- ARGV[3] = ttl（过期时间毫秒）

if redis.call("EXISTS", KEYS[1]) == 0 then
    -- 步骤1：锁不存在，第一次加锁
    redis.call("HSET", KEYS[1], ARGV[1] .. ":" .. ARGV[2], 1)
    redis.call("PEXPIRE", KEYS[1], ARGV[3])
    return 1
end
-- 步骤2：锁存在，检查是否是自己加的
local field = ARGV[1] .. ":" .. ARGV[2]
if redis.call("HEXISTS", KEYS[1], field) == 1 then
    -- 步骤3：是自己加的锁，重入次数+1
    redis.call("HINCRBY", KEYS[1], field, 1)
    redis.call("PEXPIRE", KEYS[1], ARGV[3])
    return 1
end
-- 步骤4：锁被别人持有，加锁失败
return 0

Redlock 算法

单节点 Redis 分布式锁有一个致命问题：如果 Redis 主节点宕机，且锁还没同步到从节点，从节点被提升为新主后，别的客户端可以在新主上加同样的锁——锁失效了。

Redlock 由 Redis 作者提出，核心思路是多节点多数派：

flowchart TD
    C["客户端"] -->|"SET NX PX"| R1["Redis节点1
加锁成功"]
    C -->|"SET NX PX"| R2["Redis节点2
加锁成功"]
    C -->|"SET NX PX"| R3["Redis节点3
加锁失败"]
    C -->|"SET NX PX"| R4["Redis节点4
加锁成功"]
    C -->|"SET NX PX"| R5["Redis节点5
加锁成功"]
    R1 --> Count["统计：4/5成功"]
    R2 --> Count
    R4 --> Count
    R5 --> Count
    Count -->|"多数成功
加锁成功"| OK["锁有效"]

Redlock 的步骤：

1. 获取当前时间 T1。
2. 依次向 5 个 Redis 节点发送 SET key value NX PX 请求，设短超时（如 50ms）。
3. 统计成功数量。如果 >= 3 个节点成功（多数派），且总耗时 < 锁的过期时间，则加锁成功。
4. 加锁失败时，向所有节点发送 DEL 释放锁。

⚠️ 新手必踩的坑： Redlock 在工程实践中争议很大。Martin Kleppmann（DDIA 作者）曾写文章批评 Redlock 在时钟漂移场景下不安全。如果你的业务对锁的正确性要求极高（如金融转账），应该用 Zookeeper 或 etcd 而不是 Redis 分布式锁。Redis 分布式锁适合"对性能要求高、偶尔丢锁可以接受"的场景。

Redis 分布式锁 vs Zookeeper 分布式锁

分布式锁的优缺点

优点：

• 实现简单，性能高，适合高并发场景。
• Redis 部署成本低，大多数团队已有 Redis 基础设施。

缺点：

• 锁有过期时间，业务执行超过过期时间会导致锁失效。
• 主从切换时可能丢锁（AP 模型的固有问题）。
• 不是公平锁，可能导致饥饿。

三、缓存三大问题

3.1 用生活类比先建立直觉

类比：把缓存想象成一家餐厅的"传菜窗口"（缓存），后厨是数据库。正常情况下，服务员从传菜窗口取菜，很快。但有三种异常情况：

• 缓存穿透：有人点了一道菜单上根本没有的菜（查不存在的数据）。服务员去传菜窗口找——没有，跑去后厨问——后厨也说没有。这个人在反复点这道不存在的菜，服务员每次都跑后厨，后厨被烦死了。
• 缓存击穿：传菜窗口上有一道"招牌菜"（热点 key），突然被撤走了（过期了）。一瞬间涌进来 100 个人都要点这道菜，服务员一看窗口没有，100 个人全冲向后厨——后厨直接被挤爆。
• 缓存雪崩：传菜窗口上所有菜同时过期了（大量 key 同时失效），或者传菜窗口本身塌了（Redis 宕机）。所有服务员全冲向后厨——后厨直接瘫痪。

flowchart TB
    subgraph P["缓存穿透
查不存在的数据"]
        P1["请求 key=-1"] --> P2["缓存未命中"] --> P3["DB未命中"] --> P4["每次都穿透到DB"]
    end
    subgraph B["缓存击穿
热点key过期"]
        B1["热点key过期"] --> B2["瞬间大量请求"] --> B3["同时打到DB"] --> B4["DB压力暴增"]
    end
    subgraph A["缓存雪崩
大量key同时过期或宕机"]
        A1["大量key同时过期
或Redis宕机"] --> A2["海量请求打到DB"] --> A3["DB可能崩溃
系统雪崩"]
    end

这张对比图是本节的核心：穿透是"查不存在的"、击穿是"一个热key过期"、雪崩是"一大片key同时过期"。三者的区别和解决方案完全不同，面试时必须分清。

3.2 工程要点

缓存穿透

触发场景：恶意攻击者用大量不存在的 ID 查询（如 id = -1 或不存在的 UUID），每次都绕过缓存打到数据库。

解决方案一：空值缓存

// GetUser 查询用户，空值缓存防穿透
func GetUser(ctx context.Context, rdb *redis.Client, db *sql.DB, userID int) (*User, error) {
    key := fmt.Sprintf("user:%d", userID)

    // 步骤1：先查缓存
    val, err := rdb.Get(ctx, key).Result()
    if err == nil {
        if val == "NULL" {
            // 步骤2：命中空值缓存，说明数据不存在，直接返回
            return nil, errors.New("user not found")
        }
        // 步骤3：正常命中缓存
        var u User
        json.Unmarshal([]byte(val), &u)
        return &u, nil
    }

    // 步骤4：缓存未命中，查数据库
    user, err := queryUserFromDB(db, userID)
    if err != nil {
        // 步骤5：数据库也没有，缓存空值，设短过期时间（如60秒）
        rdb.Set(ctx, key, "NULL", 60*time.Second)
        return nil, err
    }

    // 步骤6：数据库有，缓存真实值
    data, _ := json.Marshal(user)
    rdb.Set(ctx, key, data, 10*time.Minute)
    return user, nil
}

⚠️ 新手必踩的坑： 空值缓存的过期时间一定要短（如 60 秒），否则如果数据后来被创建了，缓存里一直是空值，用户永远查不到。另外，如果攻击者用几百万个不同的不存在的 ID 攻击，空值缓存方案会导致 Redis 被大量空值占满内存——这时候需要用布隆过滤器。

解决方案二：布隆过滤器

布隆过滤器是一个位数组 + 多个哈希函数。它能在 O(1) 时间内判断"某个元素一定不存在"或"可能存在"。

// BloomFilter 简化版布隆过滤器示例
type BloomFilter struct {
    bits   []bool
    hashes int // 哈希函数个数
}

func NewBloomFilter(size, hashes int) *BloomFilter {
    return &BloomFilter{bits: make([]bool, size), hashes: hashes}
}

// Add 向布隆过滤器添加元素
func (bf *BloomFilter) Add(item string) {
    // 步骤1：用多个哈希函数计算多个位置
    for i := 0; i < bf.hashes; i++ {
        pos := bf.hash(item, i)
        bf.bits[pos] = true
    }
}

// MightContain 判断元素是否可能存在
// 返回false表示一定不存在，返回true表示可能存在（有误判率）
func (bf *BloomFilter) MightContain(item string) bool {
    // 步骤2：检查所有哈希位置是否都为true
    for i := 0; i < bf.hashes; i++ {
        pos := bf.hash(item, i)
        if !bf.bits[pos] {
            return false // 有一个位置为false，说明一定不存在
        }
    }
    return true // 所有位置都为true，可能存在（有误判率）
}

func (bf *BloomFilter) hash(item string, seed int) int {
    h := fnv.New32a()
    h.Write([]byte(item))
    h.Write([]byte{byte(seed)})
    return int(h.Sum32()) % len(bf.bits)
}

实际项目中不需要自己实现，Redis 4.0 之后可以用 RedisBloom 模块，直接 BF.ADD 和 BF.EXISTS 命令。

缓存击穿

触发场景：某个热点 key（如首页推荐、秒杀商品）过期的一瞬间，大量并发请求同时打到数据库。

解决方案一：互斥锁（Mutex）

// GetHotData 互斥锁防击穿
func GetHotData(ctx context.Context, rdb *redis.Client, db *sql.DB, key string) (string, error) {
    // 步骤1：先查缓存
    val, err := rdb.Get(ctx, key).Result()
    if err == nil {
        return val, nil // 命中缓存
    }

    // 步骤2：缓存未命中，尝试获取互斥锁
    lockKey := key + ":lock"
    ok, err := rdb.SetNX(ctx, lockKey, "1", 10*time.Second).Result()
    if err != nil {
        return "", err
    }

    if !ok {
        // 步骤3：获取锁失败，说明其他请求正在重建缓存，短暂等待后重试
        time.Sleep(50 * time.Millisecond)
        return GetHotData(ctx, rdb, db, key) // 递归重试
    }

    // 步骤4：获取锁成功，查数据库重建缓存
    defer rdb.Del(ctx, lockKey) // 步骤5：用完释放锁
    data, err := queryFromDB(db, key)
    if err != nil {
        return "", err
    }
    // 步骤6：写入缓存，设较长过期时间
    rdb.Set(ctx, key, data, 30*time.Minute)
    return data, nil
}

解决方案二：热点 key 永不过期（逻辑过期）

不设 TTL，在 value 中存一个逻辑过期时间字段。后台异步任务定期检查，发现逻辑过期就重建缓存。

type CacheData struct {
    Data      string `json:"data"`
    ExpireAt  int64  `json:"expire_at"` // 逻辑过期时间（时间戳）
}

// GetWithLogicalExpire 逻辑过期方案
func GetWithLogicalExpire(ctx context.Context, rdb *redis.Client, key string) (string, error) {
    val, err := rdb.Get(ctx, key).Result()
    if err != nil {
        return "", errors.New("cache miss")
    }

    var cd CacheData
    json.Unmarshal([]byte(val), &cd)

    // 步骤1：检查是否逻辑过期
    if time.Now().Unix() < cd.ExpireAt {
        return cd.Data, nil // 未过期，直接返回
    }

    // 步骤2：已逻辑过期，尝试获取锁异步重建
    lockKey := key + ":rebuild:lock"
    ok, _ := rdb.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
    if ok {
        // 步骤3：获取锁成功，异步重建缓存
        go rebuildCache(rdb, key)
    }

    // 步骤4：无论是否获取到锁，都先返回旧数据（不阻塞用户请求）
    return cd.Data, nil
}

⚠️ 新手必踩的坑： 互斥锁方案会阻塞后续请求（等锁的线程都在 sleep），高并发下可能造成请求堆积。逻辑过期方案不阻塞但返回的是旧数据——如果业务不能容忍短暂的数据不一致（如库存扣减），不能用这个方案。

缓存雪崩

触发场景：大量 key 在同一时间过期，或者 Redis 整体宕机。

解决方案：

// SetWithRandomTTL 设过期时间时加随机值，防雪崩
func SetWithRandomTTL(ctx context.Context, rdb *redis.Client, key, value string) {
    // 步骤1：基础过期时间
    baseTTL := 30 * time.Minute
    // 步骤2：加随机偏移（0~5分钟），避免大量key同时过期
    randomTTL := time.Duration(rand.Intn(300)) * time.Second
    ttl := baseTTL + randomTTL
    rdb.Set(ctx, key, value, ttl)
}

三大问题对比总结

四、Redis 线程安全与单线程模型

4.1 用生活类比先建立直觉

类比：Redis 的单线程模型就像一家只有一个窗口的银行。你可能会问：“一个窗口不慢吗？"——但这个窗口的柜员手速极快（纯内存操作），而且安装了一套"叫号系统”（IO 多路复用 epoll），可以同时听到 1000 个人喊"我要办业务"，谁准备好了就先处理谁，不需要一个一个排队等。

Redis 6.0 的多线程改革，就像是：柜员（命令执行）还是只有一个，但"读身份证"和"递材料"（IO 读写）这些杂活交给多个助手并行做。柜员只需要专注"算账"这个核心工作。

flowchart TB
    C1["客户端1"] --> IO
    C2["客户端2"] --> IO
    C3["客户端3"] --> IO
    subgraph IO["IO多线程 读写socket+协议解析"]
        T1["IO线程1"] --> Queue["命令队列"]
        T2["IO线程2"] --> Queue
        T3["IO线程3"] --> Queue
    end
    Queue --> Exec["主线程
单线程串行执行命令"]
    Exec --> Queue2["结果队列"]
    Queue2 --> IO2["IO多线程
写回socket"]
    IO2 --> C1
    IO2 --> C2
    IO2 --> C3

这张图展示了 Redis 6.0 的多线程架构：IO 读写是多线程的，但命令执行仍然是单线程串行的。这样既提升了 IO 吞吐量，又保证了命令执行的线程安全。

4.2 工程要点

Redis 单线程为什么快

⚠️ 新手必踩的坑： “单线程"指的是命令执行是单线程的，不是说 Redis 整个进程只有一个线程。Redis 后台还有 BIO 线程做异步任务（如关闭文件、AOF 刷盘、lazy free 释放大key）。6.0 之后 IO 读写也变成了多线程。

Redis 6.0 多线程

Redis 6.0 引入了多线程 IO，但仅限于网络读写和协议解析，命令执行仍然是单线程。开启方式：

# redis.conf 配置
io-threads 4        # IO线程数（建议CPU核数的一半）
io-threads-do-reads yes  # 开启多线程读（默认只多线程写）

为什么命令执行不改成多线程？因为 Redis 的所有数据结构操作都不是线程安全的，如果改成多线程执行命令，就要加锁——加锁的性能损耗可能比单线程还大，得不偿失。

Redis 线程安全问题

单线程命令执行模型下，每条命令天然是原子的。但多个客户端并发操作时，仍需注意命令间的原子性：

// 危险示例：非原子操作
// 步骤1：GET 获取当前值
val, _ := rdb.Get(ctx, "counter").Int()
// 步骤2：在Go中加1
val++
// 步骤3：SET 写回
// 步骤1和步骤3之间，其他客户端可能也修改了counter，导致覆盖
rdb.Set(ctx, "counter", val, 0)

// 正确示例1：用INCR原子自增
rdb.Incr(ctx, "counter") // 单条命令，天然原子

// 正确示例2：用Lua脚本保证多步原子性
script := `
    local current = redis.call("GET", KEYS[1])
    if not current then current = 0 end
    current = tonumber(current) + 1
    redis.call("SET", KEYS[1], current)
    return current
`
result, _ := rdb.Eval(ctx, script, []string{"counter"}).Int()

// 正确示例3：用MULTI/EXEC事务
// 步骤1：开启事务
pipe := rdb.TxPipeline()
// 步骤2：事务内多个命令会被打包，不会被其他命令插入
pipe.Incr(ctx, "counter")
pipe.Expire(ctx, "counter", 10*time.Minute)
// 步骤3：执行事务
pipe.Exec(ctx)

⚠️ 新手必踩的坑： Redis 的 MULTI/EXEC 事务不支持回滚！如果事务中某条命令出错（如类型错误），其他命令仍然会执行。这与 MySQL 事务的原子性完全不同。如果需要严格的原子性+回滚，应该用 Lua 脚本。

五、Redis 持久化

5.1 用生活类比先建立直觉

类比：Redis 在内存里存数据，就像你在白板上写东西——断电就没了。持久化就是"把白板上的内容抄到本子上"的两种方式：

• RDB（快照）：定期给整个白板拍一张照片。恢复时直接看照片，快。但如果拍照间隔内断电，最近一次拍照之后写的内容就丢了。
• AOF（日志）：你每写一笔，就在本子上记一笔（追加日志）。恢复时把本子上的操作重放一遍。数据安全，但本子越来越厚，恢复慢。
• 混合持久化（4.0+）：先拍一张照片存到本子上，之后只记新增的笔迹。恢复时先看照片再重放日志——两全其美。

flowchart LR
    subgraph RDB["RDB 快照持久化"]
        R1["fork子进程"] --> R2["遍历所有key
生成二进制快照"] --> R3["压缩写入
dump.rdb"]
    end
    subgraph AOF["AOF 日志持久化"]
        A1["执行写命令"] --> A2["追加到
AOF缓冲区"] --> A3["按策略fsync
刷入磁盘"] --> A4["appendonly.aof
文件越来越大"]
    end
    subgraph Mixed["混合持久化 4.0+"]
        M1["BGREWRITEAOF时
先做RDB快照"] --> M2["快照之后的
新命令以AOF追加"] --> M3["恢复时
先加载RDB再重放AOF"]
    end

5.2 工程要点

RDB 持久化

RDB（Redis Database）是内存数据的二进制快照。触发方式：

RDB 的核心是 fork() + COW（Copy-On-Write）：

// RDB的fork过程（伪代码描述）
// 步骤1：主进程调用fork()，创建子进程
// 步骤2：子进程共享主进程的内存页（COW）
// 步骤3：子进程遍历所有key，写入RDB文件
// 步骤4：主进程继续处理命令，修改数据时触发COW，操作系统复制对应内存页
// 步骤5：子进程写完RDB后，替换旧文件

⚠️ 新手必踩的坑： fork() 在大内存实例上可能很慢（如 10GB 内存 fork 可能阻塞数百毫秒）。而且 COW 机制下，如果 fork 期间写入量很大，操作系统需要复制大量内存页，可能导致内存占用翻倍。所以大内存 Redis 实例建议用 AOF 或混合持久化。

AOF 持久化

AOF（Append Only File）记录每条写命令。三种 fsync 策略：

AOF 文件会越来越大，Redis 提供 AOF 重写机制：遍历内存中所有 key，用最精简的命令重新生成 AOF 文件。例如对同一个 key SET 了 100 次，重写后只保留最后一次的 SET。

# AOF 重写触发条件
auto-aof-rewrite-percentage 100  # 文件大小比上次重写后增长100%时触发
auto-aof-rewrite-min-size 64mb   # 文件最小达到64MB才触发重写

混合持久化（Redis 4.0+）

# 开启混合持久化
aof-use-rdb-preamble yes

混合持久化在 AOF 重写时，将当前内存数据以 RDB 格式写入 AOF 文件开头，之后新增的命令以 AOF 格式追加在后面。恢复时先加载 RDB 部分（快），再重放 AOF 部分（补全增量）。

六、过期与淘汰策略

6.1 用生活类比先建立直觉

类比：Redis 的内存就像一个固定大小的储物柜。当柜子满了，你需要做两件事：

• 过期策略——有些东西是"租"的，到期了要主动清理。但 Redis 不会时时刻刻盯着每个东西是否到期（太耗 CPU），而是用"惰性删除”（取东西时检查是否过期）+ “定期删除”（每隔一段时间随机抽查一批）的组合拳。
• 淘汰策略——如果过期策略清理得不够快，柜子还是满了，就需要"主动扔东西"。扔哪些？可以扔最久没用的（LRU）、最少用的（LFU）、随机扔（Random）、或者扔快过期的（TTL）。

flowchart TD
    Start["内存使用率
超过maxmemory"] --> Q1{"是否允许
写入失败?"}
    Q1 -->|"否 必须腾空间"| Q2{"是否区分
热数据和冷数据?"}
    Q1 -->|"是 优先保护
已有数据"| NoEvict["noeviction
拒绝新写入
返回OOM错误"]
    Q2 -->|"否 随机淘汰即可"| AllRand["allkeys-random
所有key中随机淘汰"]
    Q2 -->|"是 需要智能淘汰"| Q3{"只淘汰有过期
时间的key?"}
    Q3 -->|"否 所有key参与"| Q4{"用最近使用
还是访问频率?"}
    Q3 -->|"是 只淘汰会过期的"| Q5{"用LRU/LFU
还是TTL?"}
    Q4 -->|"最近最少使用"| AllLRU["allkeys-lru"]
    Q4 -->|"最少访问频率"| AllLFU["allkeys-lfu"]
    Q5 -->|"最近最少使用"| VolLRU["volatile-lru"]
    Q5 -->|"最少访问频率"| VolLFU["volatile-lfu"]
    Q5 -->|"随机"| VolRand["volatile-random"]
    Q5 -->|"快过期的优先"| VolTTL["volatile-ttl"]

这张决策树覆盖了 Redis 全部 8 种淘汰策略。面试时能画出这张图，基本就过了。

6.2 工程要点

过期策略

Redis 采用惰性删除 + 定期删除组合策略：

⚠️ 新手必踩的坑： 过期策略不能保证所有过期 key 都被及时清理。如果大量 key 过期但一直没被访问，它们会一直占用内存，直到定期删除抽到它们或触发淘汰策略。如果你发现 Redis 内存使用率远高于预期，可以用 redis-cli --bigkeys 检查是否有大量"僵尸key"。

8 种淘汰策略详解

# redis.conf 配置淘汰策略
maxmemory 4gb
maxmemory-policy allkeys-lru

⚠️ 新手必踩的坑： Redis 的 LRU 不是精确的 LRU。真正的 LRU 需要维护一个双向链表，每次访问都移动节点到头部——这在 Redis 的数据量下太耗内存。Redis 采用近似 LRU：每个 key 记录最近访问时间戳，淘汰时随机抽样 5 个（可配置 maxmemory-samples），从中淘汰最久未访问的。同理 LFU 也是近似的。

LRU vs LFU

七、Redis 高可用

7.1 用生活类比先建立直觉

类比：Redis 的高可用就像一家公司从"一个人干活"到"团队协作"的演进：

• 主从复制：老板（Master）干活，雇了几个助手（Slave）抄他的工作记录。助手只读不写，帮忙分担查询压力。但如果老板病倒了（Master宕机），没人接替——需要手动指定一个助手当新老板。
• 哨兵 Sentinel：请了一个"监工"（Sentinel），专门盯着老板。老板病倒了，监工自动从助手里选一个当新老板，并通知所有人。这就是自动故障转移。
• 集群 Cluster：公司太大了，一个老板管不过来。把业务拆成 16384 份（槽位），分给多个老板各管一部分。老板之间互相通信（Gossip协议），客户来了按业务编号（CRC16）自动找对应老板。

flowchart TB
    Sentinel["Sentinel 哨兵集群
3个节点投票监控"]
    Master["Master 主节点
读写"] -->|"全量RDB同步
+增量偏移量"| Slave1["Slave 1
只读"]
    Master -->|"复制"| Slave2["Slave 2
只读"]
    Sentinel -.->|"监控心跳"| Master
    Sentinel -.->|"监控心跳"| Slave1
    Sentinel -.->|"监控心跳"| Slave2
    Sentinel ==>|"Master宕机
选举Slave1为新Master
通知客户端切换"| Slave1

这张图展示了主从复制 + 哨兵的完整架构。哨兵集群监控所有节点，Master 宕机时自动提升 Slave 为新 Master。注意哨兵本身也要做成集群（至少3个节点），避免单点故障。

7.2 工程要点

主从复制

主从复制是 Redis 高可用最基础的架构。数据从 Master 复制到 Slave，Slave 只读不写。

全量同步流程：

1. Slave 连接 Master，发送 PSYNC ? -1（第一次连接，没有偏移量）。
2. Master 执行 BGSAVE 生成 RDB 文件。
3. Master 将 RDB 文件发送给 Slave。
4. Slave 加载 RDB 文件，恢复数据。
5. Master 将 RDB 生成期间的写命令发送给 Slave，Slave 重放。

增量同步流程：

1. Slave 断线重连后发送 PSYNC <runid> <offset>。
2. Master 检查 offset 是否在复制积压缓冲区（repl_backlog）中。
3. 如果在，只发送 offset 之后的增量命令。
4. 如果不在（断线太久），触发全量同步。

# 主从复制配置（Slave端）
replicaof 192.168.1.100 6379
repl-backlog-size 1mb  # 复制积压缓冲区大小，越大越不容易全量同步

⚠️ 新手必踩的坑： 主从复制是异步的。Master 写入后立即返回客户端成功，不等 Slave 确认。这意味着 Master 宕机时，Slave 可能还没收到最新的写命令——会丢数据。如果需要强一致，可以用 WAIT numreplicas timeout 命令等待至少 N 个 Slave 确认。

哨兵 Sentinel

哨兵的三大职责：

故障转移流程：

1. 哨兵每秒向 Master 发 PING，超过 down-after-milliseconds 未响应标记为主观下线（SDOWN）。
2. 超过半数哨兵都标记为 SDOWN，升级为客观下线（ODOWN）。
3. 哨兵集群选举 Leader 执行故障转移。
4. Leader 从 Slave 中选一个（优先级最高、偏移量最大、runid 最小）提升为新 Master。
5. 通知其他 Slave 改为新 Master 的从节点，通知客户端切换。

# sentinel.conf 配置
sentinel monitor mymaster 192.168.1.100 6379 2  # 2个哨兵同意才算客观下线
sentinel down-after-milliseconds mymaster 5000  # 5秒无响应判定下线
sentinel failover-timeout mymaster 60000       # 故障转移超时时间

集群 Cluster

Cluster 是 Redis 的分布式方案，通过分片将数据分散到多个节点。

flowchart TB
    Client["客户端"] -->|"CRC16计算
取模16384"| Router{"槽位路由"}
    Router -->|"0-5460"| NodeA["节点A
Master+Slave"]
    Router -->|"5461-10922"| NodeB["节点B
Master+Slave"]
    Router -->|"10923-16383"| NodeC["节点C
Master+Slave"]
    NodeA <-.->|"Gossip协议
状态同步"| NodeB
    NodeB <-.->|"Gossip协议"| NodeC
    NodeA <-.->|"Gossip协议"| NodeC

Cluster 的核心机制：

// Go 连接 Redis Cluster 示例
func NewClusterClient() *redis.ClusterClient {
    // 步骤1：配置集群节点地址
    rdb := redis.NewClusterClient(&redis.ClusterOptions{
        Addrs: []string{
            ":7000",
            ":7001",
            ":7002",
            ":7003",
            ":7004",
            ":7005",
        },
        // 步骤2：配置路由重试
        MaxRetries:  3,
        RouteRandomly: true, // 读请求随机路由到Slave
        // 步骤3：配置连接池
        PoolSize: 20,
    })
    return rdb
}

⚠️ 新手必踩的坑： Cluster 模式下，跨槽位操作会被拒绝。例如 MGET key1 key2 如果 key1 和 key2 在不同槽位，会报 CROSSSLOT 错误。解决方法是用 Hash Tag：MGET {user}:1 {user}:2——花括号内的内容参与 CRC16 计算，保证路由到同一槽位。

三种高可用方案对比

八、发布订阅

8.1 用生活类比先建立直觉

类比：Redis 的发布订阅就像一个广播电台。发布者是电台主持人，订阅者是听众：

• Pub/Sub：听众打电话到电台说"我要听音乐频道"，主持人开播时所有正在听的听众同时收到。但如果某个听众那时候手机没信号（离线），主持人说的话他就错过了——Pub/Sub 不存历史消息。
• Stream：相当于电台还有"回放"功能。听众不仅能在直播时听，还能回看之前错过的内容。Stream 会把消息持久化存下来，消费者可以按 ID 读取任意位置的消息。

flowchart LR
    Publisher["发布者
PUBLISH channel msg"] --> Channel["Channel 频道
Redis内部维护"]
    Channel --> Sub1["订阅者1
SUBSCRIBE channel"]
    Channel --> Sub2["订阅者2
SUBSCRIBE channel"]
    Channel --> Sub3["订阅者3
SUBSCRIBE channel"]
    Note["注意：消息不持久化
离线订阅者收不到
消息发完即丢弃"] -.-> Channel

这张图展示了 Pub/Sub 的消息流转：发布者发消息到频道，所有在线订阅者同时收到。但消息不存储，发完就没了。

8.2 工程要点

原生 Pub/Sub

package main

import (
    "context"
    "fmt"

    "github.com/redis/go-redis/v9"
)

// 订阅者
func subscribe(ctx context.Context, rdb *redis.Client, channel string) {
    // 步骤1：订阅频道
    sub := rdb.Subscribe(ctx, channel)
    defer sub.Close()

    // 步骤2：循环接收消息
    for {
        msg, err := sub.ReceiveMessage(ctx)
        if err != nil {
            fmt.Println("接收错误:", err)
            return
        }
        fmt.Printf("收到消息 [%s]: %s\n", msg.Channel, msg.Payload)
    }
}

func main() {
    rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
    ctx := context.Background()

    // 步骤3：启动订阅者协程
    go subscribe(ctx, rdb, "chat:room1")

    // 步骤4：发布者发送消息
    rdb.Publish(ctx, "chat:room1", "Hello, World!")
    rdb.Publish(ctx, "chat:room1", "第二条消息")
}

Pub/Sub 的缺点：

Stream（Redis 5.0+）

Stream 是 Redis 自己的"Kafka"，支持消费者组和持久化：

// 生产者：写入消息到Stream
// 步骤1：XADD 写入消息，* 表示自动生成ID
rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "orders",
    MaxLen: 10000,    // 限制Stream长度，近似裁剪
    Values: map[string]interface{}{
        "order_id": "10086",
        "amount":   "99.9",
    },
})

// 消费者：创建消费者组
// 步骤2：XGROUP CREATE 创建消费者组
rdb.XGroupCreate(ctx, "orders", "order_processors", "$")

// 消费者：读取并处理消息
// 步骤3：XREADGROUP 从消费者组读取消息
msgs, _ := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "order_processors",
    Consumer: "consumer-1",
    Streams:  []string{"orders", ">"},
    Count:    10,
    Block:    5 * time.Second,
}).Result()

for _, msg := range msgs[0].Messages {
    // 步骤4：处理消息
    fmt.Println(msg.Values)
    // 步骤5：XACK 确认消息已处理
    rdb.XAck(ctx, "orders", "order_processors", msg.ID)
}

⚠️ 新手必踩的坑： Stream 的消费者如果读取了消息但没发 XACK（比如处理时崩溃了），这条消息会进入 PEL（Pending Entries List）。其他消费者可以用 XCLAIM 或 XAUTOCLAIM 接手这些消息。但如果你忘了处理 PEL，消息会永远卡在那里。生产环境一定要有定时扫描 PEL 的机制。

Pub/Sub vs Stream 对比

九、缓存架构与选型

9.1 用生活类比先建立直觉

类比：缓存架构就像零售业的物流体系：

• 本地缓存（进程内缓存）= 门店货架。拿货最快（纳秒级），但每个门店货架有限，且各门店不共享（不同实例缓存不一致）。
• Redis 缓存 = 区域仓库。所有门店共享，拿货快（亚毫秒级），但有网络开销。
• MySQL 数据库 = 总厂仓库。货最全、最持久，但远（毫秒到秒级），运费贵。

多级缓存就是"门店货架→区域仓库→总厂仓库"的三级物流——先从最近的取，取不到再往上一级。

flowchart TD
    Request["用户请求"] --> L1["L1 本地缓存
进程内 sync.Map / FreeCache
纳秒级"]
    L1 -->|"未命中"| L2["L2 Redis缓存
分布式共享
亚毫秒级"]
    L2 -->|"未命中"| DB["MySQL 数据库
磁盘持久化
毫秒~秒级"]
    DB -->|"回填"| L2
    L2 -->|"回填"| L1

9.2 工程要点

Redis vs 本地缓存

多级缓存实践：

package main

import (
    "context"
    "encoding/json"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
)

// MultiLevelCache 多级缓存：本地缓存 + Redis + DB
type MultiLevelCache struct {
    local  *sync.Map       // L1：本地缓存
    rdb    *redis.Client   // L2：Redis缓存
    ttl    time.Duration   // 本地缓存TTL
    rdbTTL time.Duration   // Redis缓存TTL
}

func NewMultiLevelCache(rdb *redis.Client) *MultiLevelCache {
    return &MultiLevelCache{
        local:  &sync.Map{},
        rdb:    rdb,
        ttl:    10 * time.Second,   // 本地缓存短TTL，降低不一致窗口
        rdbTTL: 30 * time.Minute,   // Redis长TTL
    }
}

type cacheEntry struct {
    Data      []byte
    ExpireAt  time.Time
}

// Get 多级缓存查询
func (c *MultiLevelCache) Get(ctx context.Context, key string, loader func() ([]byte, error)) ([]byte, error) {
    // 步骤1：查L1本地缓存
    if val, ok := c.local.Load(key); ok {
        entry := val.(*cacheEntry)
        if time.Now().Before(entry.ExpireAt) {
            return entry.Data, nil // L1命中
        }
        c.local.Delete(key) // 过期删除
    }

    // 步骤2：查L2 Redis缓存
    data, err := c.rdb.Get(ctx, key).Bytes()
    if err == nil {
        // 步骤3：L2命中，回填L1
        c.local.Store(key, &cacheEntry{
            Data:     data,
            ExpireAt: time.Now().Add(c.ttl),
        })
        return data, nil
    }

    // 步骤4：L2未命中，查DB（通过loader回调）
    data, err = loader()
    if err != nil {
        return nil, err
    }

    // 步骤5：回填L2和L1
    c.rdb.Set(ctx, key, data, c.rdbTTL)
    c.local.Store(key, &cacheEntry{
        Data:     data,
        ExpireAt: time.Now().Add(c.ttl),
    })
    return data, nil
}

Redis vs MySQL

缓存数据量大放不下

⚠️ 新手必踩的坑： BigKey 是 Redis 运维的头号杀手。一个包含 100 万元素的 Hash 会导致：阻塞主线程（DEL 时）、网络带宽暴涨、集群迁移卡顿、内存不均匀。生产环境务必监控 BigKey，用 redis-cli --bigkeys 定期扫描。

热点 Key 处理

当某个 key 的访问量极大（如明星微博、秒杀商品）：

100 万粉丝存储与推送

这是一个综合面试题：如何存储用户关注关系并实现消息推送？

关注关系存储：

// 存储关注关系
// 步骤1：用户A关注用户B
// 用ZSet存储，score为关注时间戳，方便按时间排序
rdb.ZAdd(ctx, "following:1001", redis.Z{
    Score:  float64(time.Now().Unix()),
    Member: "1002",
})

// 步骤2：反向存储——B的粉丝列表
rdb.ZAdd(ctx, "followers:1002", redis.Z{
    Score:  float64(time.Now().Unix()),
    Member: "1001",
})

// 步骤3：共同关注——求交集
rdb.ZInterStore(ctx, "common_following", &redis.ZStore{
    Keys: []string{"following:1001", "following:1003"},
})

消息推送方案对比：

// 推拉结合方案示例
// 步骤1：大V发微博，写入收件箱（Stream）
weiboID := rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "inbox:1002", // 1002是发布者
    Values: map[string]interface{}{
        "content": "今天天气真好",
        "time":    time.Now().Unix(),
    },
}).Val()

// 步骤2：对活跃粉丝，推送推送到他们的收件箱
activeFans := rdb.ZRangeByScore(ctx, "followers:1002", &redis.ZRangeBy{
    Min:    fmt.Sprintf("%d", time.Now().AddDate(0, 0, -7).Unix()),
    Max:    "+inf",
    Count:  1000, // 分批处理，每批1000个
    Offset: 0,
}).Val()

// 步骤3：批量推送（用pipeline减少网络往返）
pipe := rdb.Pipeline()
for _, fanID := range activeFans {
    pipe.XAdd(ctx, &redis.XAddArgs{
        Stream: "inbox:" + fanID,
        Values: map[string]interface{}{
            "weibo_id": weiboID,
            "from":     "1002",
        },
    })
}
pipe.Exec(ctx)

// 步骤4：非活跃粉丝登录时，主动拉取关注者的最新微博
func pullLatest(ctx context.Context, rdb *redis.Client, userID string, lastPullTime int64) {
    followings := rdb.ZRangeByScore(ctx, "following:"+userID, &redis.ZRangeBy{
        Min: fmt.Sprintf("%d", lastPullTime),
        Max: "+inf",
    }).Val()
    // 从每个关注者的收件箱拉取新消息
    for _, followID := range followings {
        msgs, _ := rdb.XRange(ctx, "inbox:"+followID, "-", "+").Result()
        // 处理消息...
        _ = msgs
    }
}

⚠️ 新手必踩的坑： 大V可能有几千万粉丝，全量推送会导致 Redis 瞬间写入暴增（“写扩散"问题）。实际工程中要对粉丝分层：活跃粉丝推（写扩散），非活跃粉丝拉（读扩散），超大体量用消息队列异步推送。

十、爬虫导致热数据被淘汰从而引发雪崩

10.1 用生活类比先建立直觉

类比：把 Redis 缓存想象成一家热门餐厅的"留座区”（缓存座位有限）。正常情况下，老顾客（热数据）长期占着好位置，新顾客（冷数据）来了坐一下就走。

现在来了一群不买东西只到处乱窜的蹭空调的人（爬虫）：他们疯狂地、反复地冲进来占座，每个人还换不同的名字（伪造不同 key / 不同 ID）。结果留座区被这些"幽灵顾客"挤满，真正该留下的老顾客（热数据）被赶了出去。等真正的老顾客（真实用户）回来想坐时——座位没了，全挤到大堂（数据库）去了，大堂瞬间瘫痪——这就是"雪崩"。

对应到工程里：爬虫的高频、广覆盖请求会快速消耗缓存容量，触发 Redis 的淘汰策略把真正的热 key 清掉；一旦热 key 被清，真实流量集中打到数据库，就可能从"热 key 失效"升级成"全面雪崩"。这条链路是 缓存雪崩 的一个特定触发源，需要针对性防御。

flowchart TD
    C["爬虫高频/伪造请求
反复爬取某批数据"] --> F["缓存被大量
幽灵key占满"]
    F --> E["触发淘汰策略
LRU/LFU 误删真实热key"]
    E --> M["热key被逐出缓存"]
    M --> H["真实用户请求
热key缓存未命中"]
    H --> D["请求集中打到MySQL"]
    D --> A["DB压力暴增
系统雪崩"]

这张图就是"爬虫 → 占用缓存 → 误淘汰热数据 → 真实流量打穿 DB → 雪崩"的完整链路。关键点是：它和普通雪崩的区别在于根因是爬虫在"挤占"缓存资源，所以防御要同时从"拦爬虫"和"保热 key"两头下手。

10.2 工程要点

根因拆解：为什么爬虫会"清掉"热数据

Redis 在内存达到 maxmemory 后会按淘汰策略清 key。爬虫带来的两类请求最容易引发误淘汰：

⚠️ 新手必踩的坑： 很多人以为"加了缓存雪崩的随机 TTL 就够了"。但爬虫引发的雪崩是淘汰驱动的，不是"同一时刻过期"驱动的——随机 TTL 对这类问题基本无效，必须从"限制爬虫流量 + 保护热 key 不被淘汰"入手。

防御方案一：在入口拦住爬虫

最根本的解法是别让爬虫把缓存填满。常用手段：

// AntiCrawler 简单爬虫限流：基于 IP 的滑动窗口计数
// 步骤1：用 Redis ZSet 记录某 IP 最近 N 秒的请求时间戳（score=时间戳）
func isCrawler(rdb *redis.Client, ip string, now int64, window, limit int64) (bool, error) {
    key := "req:cnt:" + ip
    // 步骤2：移除窗口外的旧请求（只保留最近 window 秒）
    rdb.ZRemRangeByScore(context.Background(), key, "0", fmt.Sprintf("%d", now-window))
    // 步骤3：统计窗口内剩余请求数
    cnt, err := rdb.ZCard(context.Background(), key).Result()
    if err != nil {
        return false, err
    }
    // 步骤4：窗口内请求数超过阈值，判定为爬虫
    if cnt > limit {
        return true, nil
    }
    // 步骤5：正常请求，记录本次时间戳，设短过期自动清理
    rdb.ZAdd(context.Background(), key, redis.Z{Score: float64(now), Member: fmt.Sprintf("%d", now)})
    rdb.Expire(context.Background(), key, time.Duration(window)*time.Second)
    return false, nil
}

配合 Nginx/网关层的 UA 黑名单、验证码、IP 限流，可以把绝大多数恶意爬虫挡在缓存之外。

防御方案二：保护热 key 不被淘汰

即便爬虫进来，也要保证真正的"热 key"不会被淘汰策略误删。核心思路是让热 key 远离淘汰算法：

// ProtectHotKey 用本地缓存 + Redis 双层保护热 key，避免被淘汰后打穿 DB
// 步骤1：先从本地缓存取（纳秒级，爬虫流量到这里就被消化大半）
func GetHotKey(localCache *sync.Map, rdb *redis.Client, key string) (string, error) {
    if v, ok := localCache.Load(key); ok {
        return v.(string), nil // 本地命中，完全不经过 Redis 淘汰
    }
    // 步骤2：本地未命中，回源 Redis；热 key 不设 TTL，避免被 volatile/allkeys 淘汰
    val, err := rdb.Get(context.Background(), key).Result()
    if err == nil {
        localCache.Store(key, val) // 回填本地，后续请求直接命中本地
        return val, nil
    }
    // 步骤3：Redis 也未命中（极端情况），回源 DB 并回填两层
    // ... 业务回源逻辑 ...
    return val, nil
}

防御方案三：淘汰策略与监控兜底

• 淘汰策略选择：如果 Redis 里混存"持久热数据 + 可丢缓存"，给热数据不设 TTL 并用 allkeys-lru——这样 LRU 只淘汰可丢的缓存，不会动无 TTL 的热数据；反之若用 volatile-lru 且热数据也没 TTL，则两组都不参与淘汰，内存压力会转成拒绝写入（OOM），所以要权衡。
• 热 key 发现：开启 maxmemory-policy 监控，用 redis-cli --hotkeys（基于 LFU）定期扫描，把发现的热 key 自动纳入本地缓存白名单。
• 多级缓存兜底：本地缓存 → Redis → DB 的三级结构（见第九章多级缓存），即使 Redis 热 key 被淘汰，本地层仍能扛住绝大部分流量，避免直接雪崩到 DB。

⚠️ 新手必踩的坑： 本地缓存 + 热 key 不设 TTL 的方案有个副作用——数据更新不及时。热 key 改了值，本地缓存还返回旧值。生产上要给热 key 设计"主动失效"机制（发布订阅通知各实例清本地缓存），否则会出现长时间脏数据。

十一、Redis 基础认知（什么是 Redis / 好处 / 为什么放内存 / 最大容量）

11.1 什么是 Redis

类比：把数据库想象成"地下冷库"——商品（数据）放在货架上（磁盘），每次取要下楼搬运，慢。Redis 像一个"开在内存里的超级便利店"——所有商品就摆在柜台（内存）上，店员（CPU）伸手即取，结算极快。

对应到工程里：Redis 是开源的、基于内存的键值型（Key-Value）数据结构存储系统，同时能当数据库、缓存、消息中间件用。它支持持久化、多种数据结构、单线程命令执行、高可用与分布式。

11.2 使用 Redis 的好处（面试要点 #3）

11.3 为什么把所有数据放内存（面试要点 #12）

内存读写是纳秒级，磁盘是毫秒级，差约 10 万倍。Redis 的设计目标是"快"，只有数据全在内存才能做到高吞吐、低延迟。持久化只是"保险"——宕机后靠 RDB/AOF 恢复，正常运行时服务完全依赖内存。如果数据放磁盘，就不是 Redis 了，而是另一个 KV 存储。

⚠️ 新手必踩的坑： “数据放内存"不等于"数据会丢”。Redis 通过持久化和主从复制保证了可用性，内存只是运行态，落盘是异步保底。

11.4 一个字符串值能存多大（面试要点 #7）

String 类型单个 value 的最大容量是 512MB。其他集合类型（List/Set/ZSet/Hash）的单个元素同样受 512MB 限制；集合本身的元素个数上限约为 2^32 - 1（见第十七章）。

flowchart LR
    subgraph Cold["磁盘数据库（冷库）"]
        D1["数据在磁盘"] --> D2["读写毫秒级
慢但容量大"]
    end
    subgraph Hot["Redis（便利店）"]
        M1["数据在内存"] --> M2["读写纳秒级
快但容量小"]
    end
    D2 -.->|"持久化兜底"| M2

考点总结：Redis 的"快"源于内存 + 单线程无锁 + IO 多路复用；String 最大 512MB，这是面试常考的硬指标，别答成"无限大"。

十二、Redis 与 Memcached 对比（面试要点 #4 #5）

12.1 类比

两者都是"内存 KV 缓存"，但 Memcached 像"只能存便签的简易储物柜"——只支持字符串、无持久化、无数据结构、无集群；Redis 像"带多种收纳盒的智能仓储"——多种数据结构、持久化、高可用、集群一应俱全。

12.2 核心区别

12.3 Redis 相比 Memcached 的优势（面试要点 #4）

1. 数据结构丰富：不仅仅是字符串，能做排行榜（ZSet）、计数器、集合运算。
2. 支持持久化：Memcached 纯内存，进程重启数据全丢；Redis 可恢复。
3. 原生高可用：Redis 有主从/哨兵/Cluster，Memcached 没有官方集群方案。
4. 功能全面：事务、Lua、发布订阅、Stream、键过期，Memcached 都没有。
5. 数据一致性更好：单线程无并发竞争问题（6.0 前）。

什么时候反而选 Memcached？ 纯缓存、value 都很小（<1MB）、追求极简、需要多线程高吞吐且不需要持久化/数据结构的场景，Memcached 更简单、内存小 value 利用率略好。

flowchart TB
    subgraph MC["Memcached"]
        A1["字符串"] --> A2["多线程"]
        A1 --> A3["无持久化"]
        A1 --> A4["无集群"]
    end
    subgraph RD["Redis"]
        B1["多种数据结构"] --> B2["单线程执行+IO多线程"]
        B1 --> B3["RDB/AOF持久化"]
        B1 --> B4["主从/哨兵/Cluster"]
        B1 --> B5["事务/Lua/PubSub/Stream"]
    end

考点总结：Redis 与 Memcached 的对比是高频题。记住核心差异三句话——Redis 数据结构丰富、支持持久化、有原生高可用；Memcached 只有字符串、不持久化、靠客户端分片。

十三、Pipeline 与 Redis 事务（面试要点 #14 #27 #28）

13.1 Pipeline 的好处（面试要点 #14）

类比：你去超市买 10 样东西，每拿一样都跑一趟收银台结账（10 次往返）很慢；Pipeline 像"把 10 样东西一次性放购物车，一次结账"——把多次网络往返（RTT）压缩成一次。

原理：客户端把多条命令打包发给服务端，服务端一次性顺序执行后批量返回结果，省去了 N 次 RTT（网络往返时间）。

sequenceDiagram
    participant C as 客户端
    participant S as Redis
    Note over C,S: 普通模式（3条命令=3次RTT）
    C->>S: SET a 1
    S-->>C: OK
    C->>S: SET b 2
    S-->>C: OK
    C->>S: SET c 3
    S-->>C: OK
    Note over C,S: Pipeline（1次RTT）
    C->>S: SET a 1 + SET b 2 + SET c 3
    S-->>C: OK + OK + OK

// 使用 Pipeline 批量写入，减少 RTT
// 步骤1：开启 pipeline
pipe := rdb.Pipeline()
// 步骤2：把多条命令放入管道（此时还没真正发送）
pipe.Set(ctx, "a", 1, 0)
pipe.Set(ctx, "b", 2, 0)
pipe.Set(ctx, "c", 3, 0)
// 步骤3：一次性发送给 Redis 并取回全部结果
_, err := pipe.Exec(ctx)

⚠️ 新手必踩的坑： Pipeline 不保证原子性！它只是把网络 IO 批量化，多条命令在服务端仍是逐条执行的，中间可能被其他客户端的命令插入。要原子性请用 MULTI/EXEC 或 Lua 脚本。

13.2 怎么理解 Redis 事务（面试要点 #27）

Redis 事务通过 MULTI 开启，把后续命令"打包入队"，执行 EXEC 时一次性顺序执行。关键特性：Redis 事务不支持回滚——某条命令执行失败（如类型错误），其他命令仍会继续执行。

13.3 事务相关命令（面试要点 #28）

// 用 WATCH 实现乐观锁：转账前监控余额，避免并发覆盖
// 步骤1：监控账户，闭包内执行事务
err := rdb.Watch(ctx, func(tx *redis.Tx) error {
    balance, _ := tx.Get(ctx, "account:A").Int()
    if balance < 100 {
        return errors.New("余额不足")
    }
    // 步骤2：在事务中完成扣减与增加
    _, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
        pipe.DecrBy(ctx, "account:A", 100)
        pipe.IncrBy(ctx, "account:B", 100)
        return nil
    })
    return err
}, "account:A")
// 若期间 account:A 被别的客户端改过，Watch 触发，事务不执行（err 非 nil）

flowchart TD
    S["MULTI 开启事务"] --> Q["命令入队
SET/INCR..."]
    Q --> E["EXEC 一次性执行"]
    E --> R["按顺序返回结果
某条失败不影响其他"]
    W["WATCH 监控 key"] -->|"被其他客户端修改"| F["EXEC 返回 nil
事务取消"]

考点总结：Pipeline 解决"网络往返多"的问题（非原子），事务解决"多条命令不被插入"的问题（但仍不回滚）；真正又要原子又要复杂逻辑用 Lua 脚本。

十四、运维基础：过期、密码、连通性与集群限制（面试要点 #29 #19 #26 #24 #25）

14.1 过期时间与永久有效（面试要点 #29）

不设 TTL 的 key 就是"永久有效"，直到被显式删除或淘汰策略触发。

14.2 设置密码与验证（面试要点 #19）

# redis.conf
requirepass yourStrongPassword

# 客户端连接后验证密码
redis-cli
127.0.0.1:6379> AUTH yourStrongPassword
OK

Redis 6.0 起推荐用 ACL（acl setuser）做更细粒度的权限控制，而非单一 requirepass。

14.3 测试连通性（面试要点 #26）

# 最常用：PING → 返回 PONG 说明正常
redis-cli -h 127.0.0.1 -p 6379 PING
# 输出 PONG

# 程序里可用 go-redis 的 Ping
pong, err := rdb.Ping(ctx).Result() // 返回 "PONG"

14.4 集群最大节点数（面试要点 #24）

Redis Cluster 的槽位总数是 16384，因此集群最大节点数理论上等于 16384 个；但官方建议不超过 1000 个节点（Gossip 通信开销随节点数上升明显增大）。

14.5 集群如何选择数据库（面试要点 #25）

• 单机 / 主从模式：SELECT 0~15 可切换 16 个逻辑库。
• Cluster 模式：只支持 db 0，SELECT 非 0 的库会直接报错（(error) SELECT is not allowed in cluster mode）。所以集群下所有数据都放在 db 0。

考点总结：TTL 返回 -1 是永久、-2 是不存在；集群只能用一个库（db 0）；节点上限受槽数 16384 限制、建议 <1000。

十五、Redis 集群深入：不可用、写丢失与主从复制模型（面试要点 #16 #21 #22 #23）

15.1 集群什么时候整体不可用（面试要点 #16）

Redis Cluster 的可用性取决于槽位（slot）是否全覆盖：

1. 某 master 及其所有 replica 同时宕机：该 master 负责的槽（约 16384/N）丢失，针对这些槽的读写直接报错——这部分"不可用"。
2. 超过半数的 master 宕机：无法形成选举多数派，集群无法完成故障转移，缺失槽整体不可用。

注意：只有"槽丢失"的部分不可用，存活 master 负责的槽仍可正常服务，不是"一台挂全挂"。

15.2 集群的主从复制模型（面试要点 #21 #23）

Cluster 里每个 master 可以挂多个 replica（从节点）。复制机制与普通主从复制完全一致：

• 新 replica 连上 master 后先全量同步（BGSAVE 发 RDB + 增量命令）。
• 之后通过 PSYNC 做增量同步（基于复制偏移量 + 复制积压缓冲区）。
• master 宕机时，其 replica 通过选举成为新 master，接管槽位。

flowchart TB
    M["Master 节点
负责 0-5460 槽"] -->|"PSYNC 全量+增量"| R1["Replica 1"]
    M -->|"PSYNC"| R2["Replica 2"]
    M -.->|"宕机"| Fail["FAIL 标记"]
    Fail -->|"Replica 选举
获多数派选票"| R1
    R1 -->|"接管槽位"| NewM["成为新 Master"]

15.3 集群会有写操作丢失吗（面试要点 #22）

会丢失，两个典型场景：

1. 异步复制丢写：master 写入成功后立即返回客户端，不等 replica 确认。若此时 master 宕机且 replica 还没收到最新写，故障转移后新 master 没有这条数据——写丢失。
2. 网络分区（脑裂）：原 master 因网络问题与集群隔离但仍接受客户端写，集群把其他节点选成新 master。网络恢复后原 master 被降级为 replica，它上面未同步的写被清空。

缓解：用 WAIT numreplicas timeout 让写命令等待至少 N 个 replica 确认（牺牲部分性能换取安全）；但无法 100% 杜绝（极端分区下仍可能丢）。

考点总结：Cluster 不是强一致，异步复制 + 脑裂都会丢写；集群"部分槽丢失"才不可用，不是"一台挂全挂"；可用 WAIT 降低丢写概率。

十六、Java 客户端：Jedis 与 Redisson（面试要点 #17 #18）

16.1 Redis 支持的 Java 客户端（面试要点 #17）

主流三个：

官方推荐：Lettuce（性能好、线程安全、与 Spring 生态契合）。

16.2 Jedis 与 Redisson 对比（面试要点 #18）

⚠️ 新手必踩的坑： Jedis 实例不是线程安全的，多线程共用一个 Jedis 会串命令。正确做法是用 JedisPool 每个线程借一个。Redisson 则天然线程安全，直接注入单例即可。

考点总结：Spring Boot 默认 Lettuce；要"裸命令 + 学原理"用 Jedis，要"直接拿到分布式锁/限流"用 Redisson。

十七、内存优化、容量上限与回收（面试要点 #30 #31 #32 #33 #34 #35）

17.1 内存用完会发生什么（面试要点 #33）

取决于 maxmemory-policy：

• 默认 noeviction：拒绝所有写命令（返回 OOM 错误），但读、删除仍正常。
• 设置了淘汰策略（如 allkeys-lru）：自动淘汰 key 腾出空间继续写。

17.2 一个实例最多放多少 key / 元素（面试要点 #34）

这是理论极限，实际受物理内存约束——2^32 个 key 根本放不进任何单机内存，所以线上瓶颈永远是内存而非这个上限。

17.3 如何做内存优化 / 降低内存（面试要点 #30 #32）

17.4 回收进程如何工作（面试要点 #31）

Redis 没有独立后台回收进程。内存回收发生在两条路径：

1. 命令执行时（惰性删除）：访问 key 时顺手检查是否过期，过期就删。
2. 定时事件 serverCron（定期删除）：每 100ms 随机抽取一批带 TTL 的 key 清理过期项；内存超 maxmemory 时按淘汰策略抽样淘汰。

17.5 2000W 数据只存 20W，如何保证都是热点（面试要点 #35）

思路：只把热数据放进有限内存，冷数据靠淘汰策略自动清掉。

maxmemory 4gb
maxmemory-policy allkeys-lru   # 或 allkeys-lfu

• 用 allkeys-lru/allkeys-lfu：Redis 自动保留最近/最频繁访问的 20W 数据，冷数据被淘汰。
• 也可 volatile-lru + 只给热数据设 TTL，冷数据不设 TTL 不参与淘汰。
• 配合访问频率统计（LFU），真正的热点自然留存，符合"20W 热点"的目标。

flowchart LR
    DB["MySQL 2000W 数据"] -->|"只有热点被访问"| Cache["Redis 20W 容量
allkeys-lru"]
    Cache -->|"冷数据被淘汰"| Evict["自动清理
只留热点"]

考点总结：内存满默认拒写（noeviction）；容量上限是理论值、实际瓶颈是内存；优化核心是"别放 BigKey + 冷热分离 + TTL"；热点筛选交给 LRU/LFU 淘汰策略。

十八、SCAN 安全遍历大库（面试要点 #37）

18.1 为什么不用 KEYS（面试要点 #37）

KEYS pattern 会遍历全库、阻塞主线程。1 亿个 key 的实例上跑 KEYS prefix:* 直接卡死数十秒，期间所有请求超时——生产事故。

18.2 用 SCAN 替代

SCAN 用游标分批返回，每次只扫一部分，不阻塞主线程。

# 从游标 0 开始，匹配 user: 前缀，每次Hint 1000 条
SCAN 0 MATCH user:* COUNT 1000
# 返回：1) 下一个游标  2) 本批 key 列表
# 游标回到 0 表示遍历结束

// go-redis 用 Iterator 遍历
// 步骤1：创建 SCAN 迭代器
iter := rdb.Scan(ctx, 0, "user:*", 1000).Iterator()
// 步骤2：逐批迭代，底层自动翻页
for iter.Next(ctx) {
    key := iter.Val()
    // 处理 key（如批量删除/统计）
    _ = key
}
// 步骤3：检查迭代错误
if err := iter.Err(); err != nil {
    // 处理错误
}

⚠️ 新手必踩的坑： SCAN 不保证一次返回全部匹配项，且可能重复返回同一个 key。COUNT 只是"提示"每次扫描的量，不是精确返回条数。客户端必须自己做去重，且不能用"返回空"判断遍历结束——只有游标回到 0 才结束。

flowchart TD
    C0["游标=0 开始"] --> B1["扫描第一批
返回 key 子集"]
    B1 --> C1["游标=12345"]
    C1 --> B2["扫描第二批"]
    B2 --> C2{"游标=0?"}
    C2 -->|"否 继续"| B2
    C2 -->|"是 遍历结束"| End["完成"]

考点总结：大库找前缀 key 必须用 SCAN，禁用 KEYS；SCAN 可能重复、不保证全量一次返回、COUNT 非精确，游标归零才算完。

十九、缓存一致性、可用性与预热（缓存章节补充）

19.1 用缓存可能出现的问题：双写一致性（缓存章节重叠）

更新数据库与更新缓存是两步操作，非原子，会出现不一致。最常用方案是 Cache Aside（旁路缓存）：

flowchart TD
    Read["读请求"] --> RC{"缓存有?"}
    RC -->|"有"| RHit["返回缓存"]
    RC -->|"无"| RDB["查 DB"] --> RFill["回填缓存
返回"]
    Write["写请求"] --> WDB["1. 更新 DB"] --> WDel["2. 删除缓存
（不是更新缓存）"]
    WDel --> WDone["完成"]

为什么"删缓存"而不是"更新缓存"？因为并发写时，先更新 A 后更新 B 可能被乱序成先 B 后 A，导致缓存是旧值。删缓存让下次读自动回填正确值，更简单安全。更稳妥可用"延迟双删"（更新 DB → 删缓存 → 延时再删一次）。极端一致可用 Canal 订阅 binlog 异步刷新。

19.2 查询缓存报错，怎么提高可用性（缓存章节重叠）

核心思想：缓存/DB 出问题时，宁可返回不那么新/不那么全的数据，也别让整个系统挂掉（可用性优先于强一致）。

19.3 什么是缓存预热（缓存章节重叠）

定义：系统启动、大促或版本发布前，提前把热点数据主动加载进缓存，避免"冷启动"瞬间海量请求直接打穿到数据库。

实现方式：

// 缓存预热：启动阶段批量加载热点数据
func warmUpCache(ctx context.Context, rdb *redis.Client, hotKeys []string, loader func(string) (string, error)) error {
    // 步骤1：并发加载热点 key
    for _, key := range hotKeys {
        val, err := loader(key) // 从 DB 取
        if err != nil {
            return err
        }
        // 步骤2：写入缓存，设较长 TTL
        rdb.Set(ctx, key, val, 30*time.Minute)
    }
    return nil
}

也可用定时任务定期预热、或监听数据变更事件触发预热。

考点总结：双写一致性用"先更新 DB 再删缓存"（Cache Aside）；可用性靠熔断+降级+多级缓存兜底；缓存预热是"提前把热点灌进缓存"，防冷启动打穿 DB。

二十、用 Redis 做异步队列（面试要点 #39）

20.1 用 List 做队列

最原始方案：LPUSH 生产，RPOP/BRPOP 消费。

• RPOP：非阻塞，队列空时返回 nil，需轮询。
• BRPOP：阻塞弹出，队列空时阻塞等待，不占用 CPU。

优点是简单；缺点：无 ACK 确认、无消费组、消息处理中崩溃会丢失（POP 后消息已从列表移除）。

20.2 可靠队列：BRPOPLPUSH

消费时用 BRPOPLPUSH 把消息从队列原子地移到一个"处理中"列表，处理完再 LREM 删除；若消费者崩溃，消息还留在处理中列表，可重放。

// 用 List + BRPOPLPUSH 实现可靠队列
// 步骤1：生产消息
rdb.LPush(ctx, "queue:tasks", taskJSON)

// 步骤2：消费者阻塞取，并同时转存到处理中列表（原子）
task, err := rdb.BRPopLPush(ctx, "queue:tasks", "queue:tasks:processing", 0).Result()
// 步骤3：处理业务...
process(task)
// 步骤4：处理成功，从处理中列表移除
rdb.LRem(ctx, "queue:tasks:processing", 1, task)

更现代的可靠方案是 Stream（见第八章）：自带消费组、ACK、PEL、消息回溯，几乎可替代专业消息队列。List 队列适合轻量、单消费者的简单场景。

flowchart LR
    P["生产者 LPUSH"] --> Q["queue:tasks"]
    Q -->|"BRPOPLPUSH 原子转移"| Proc["queue:tasks:processing"]
    Proc -->|"处理成功 LREM"| Done["完成"]
    Proc -.->|"消费者崩溃
消息仍在"| Retry["可重放重试"]

考点总结：List 做队列简单但无 ACK；用 BRPOPLPUSH 可做成可靠队列；需要消费组/ACK/回溯请直接上 Stream（第八章已讲）。

二十一、字典(dict)渐进式 rehash 详解

21.1 用生活类比先建立直觉

类比：把 Redis 的字典（hashtable）想象成图书馆的两套书架——一号书架 ht[0] 和 二号书架 ht[1]。平时读者只在一号书架借还书。有一天管理员决定"扩建"（扩容）：要把一号书架上的书按更大的格子重新摆放。如果让管理员停业、一次性把所有书搬到二号书架，读者就得干等很久（阻塞主线程）。

所以管理员采取**“边营业边搬”的策略：每次有读者来借/还书，管理员就顺手从一号书架搬几格书到二号书架；同时所有新买的书直接上二号书架**；等一号书架彻底搬空，就把二号书架改名为一号，腾出二号准备下次扩建。管理员手里那张写着"已搬到第几格"的便签，就是 rehashidx。这种"不一次性搬完、靠平时请求慢慢搬"的做法，就是渐进式 rehash——它把一次昂贵的 O(N) 搬迁摊薄到无数次请求里，主线程永不被长时间阻塞。

stateDiagram-v2
    [*] --> 正常服务: 只用 ht[0]
    正常服务 --> Rehashing: 扩容/缩容触发
分配 ht[1]，rehashidx=0
    Rehashing --> Rehashing: 每来一个请求
顺带迁移一个桶
rehashidx++
    Rehashing --> 完成: rehashidx == ht[0].size
ht[0] 已搬空
    完成 --> 正常服务: ht[0]=ht[1]
ht[1] 清空，rehashidx=-1
    完成 --> 暂停迁移: 正在执行 BGSAVE
COW 期间暂缓
    暂停迁移 --> Rehashing: BGSAVE 结束
继续渐进迁移

这张状态图是渐进式 rehash 的核心：平时只在 ht[0] 服务；触发后进入 Rehashing，靠请求"顺带"推进 rehashidx；搬完回到单表服务；而 BGSAVE 期间因 Copy-On-Write 会暂缓迁移，避免大量内存页被复制。

21.2 工程要点

两个 hashtable 与 rehashidx

dict 在源码里持有两个哈希表：

// dict 结构（简化）
typedef struct dict {
    dictht ht[2];      // ht[0] 当前使用，ht[1] rehash 时的目标表
    long rehashidx;    // -1 表示未在 rehash；>=0 表示已迁移到第 rehashidx 个桶
    int type;          // 哈希函数、key/value 复制/比较函数
} dict;

typedef struct dictht {
    dictEntry **table; // 桶数组（指针数组）
    unsigned long size;     // 桶数量（2 的幂）
    unsigned long sizemask; // size-1，用于取模
    unsigned long used;     // 已有元素数
} dictht;

• 平时只使用 ht[0]，ht[1] 是空壳。
• rehashidx = -1：没有在 rehash。
• rehashidx >= 0：正在 rehash，表示 ht[0].table[0 .. rehashidx-1] 已经迁移到 ht[1]，下一个要迁移的是第 rehashidx 个桶。
• rehash 完成时：把 ht[0] = ht[1]（指针交换），ht[1] 重置为空，rehashidx = -1。

扩容 / 缩容触发条件

负载因子 load_factor = ht[0].used / ht[0].size。比如 size=4、used=4 时 load_factor=1，触发扩容到 size=8。

⚠️ 新手必踩的坑： 为什么"正在 BGSAVE 时默认不扩容"？因为 BGSAVE 会 fork() 子进程，父子进程共享内存页（Copy-On-Write）。如果此时大肆 rehash，大量 key 在 ht[0] 和 ht[1] 间搬动，会触发 COW 把共享内存页逐页复制一份，内存占用瞬间翻倍，可能把机器拖垮。所以 Redis 在 BGSAVE 期间把 dict_can_resize 设为 0，只有 load_factor>=5 的极端情况才强制扩容。

rehash 期间增删改查如何双表操作

渐进式 rehash 进行中，字典同时挂着 ht[0]（旧，部分已空）和 ht[1]（新，正在填充）。所有操作都遵循一套统一规则：

redis-cli 观察 rehash

# 1. 真正触发字典 rehash 的是"大 Hash / 大量 key"场景（小数据用 listpack，不涉及 rehash）
#    用 INFO 观察后台任务与内存：
redis-cli INFO stats | grep -E "instantaneous_ops_per_sec|expired_keys"
# 2. 用 INFO persistence 看是否正在 BGSAVE（影响 rehash 门槛）
redis-cli INFO persistence | grep -E "rdb_bgsave_in_progress|aof_rewrite_in_progress"
# 3. OBJECT ENCODING 可确认当前编码（rehash 是内部行为，命令行看不到 rehashidx）
redis-cli OBJECT ENCODING mybigkey

实际工程中 rehash 是源码内部行为，命令行看不到 rehashidx；但可以用 Redis 源码的 dictRehash 函数逻辑理解——每次 _dictRehashStep 迁移一个桶。面试能口述"双表 + rehashidx + 顺带迁移 + 新增只写 ht[1]“就够了。

二十二、扩容 / 缩容过程中新请求如何处理

22.1 用生活类比先建立直觉

接着上一章的图书馆类比：rehash 进行中（一号、二号书架同时在使用），读者（客户端请求）不管知不知道"正在搬家”，都能正常借还书。规则很简单：

• 来找书的人：两个书架都找一遍，一号没有就去二号。
• 来还书 / 改书的人：两个书架都看看，书在哪就在哪改；但要新登记一本馆藏书（新增 key），只能写进二号书架——绝不能再往一号塞，否则一号永远搬不空。
• 每次有人来：管理员就顺手把一号书架的"下一格"搬去二号（推进 rehashidx），这样搬书进度跟着客流自然往前走，不需要专门停业搬。

这个"请求驱动 + 双表兜底"的设计，就是 Redis 能在不阻塞主线程的前提下完成扩容/缩容的关键。

sequenceDiagram
    participant C as 客户端请求
    participant M as 主线程(字典)
    participant H0 as ht[0] 旧表
    participant H1 as ht[1] 新表
    C->>M: 任意命令(查/增/删/改)
    M->>M: 先迁移 ht[0][rehashidx] 桶到 ht[1]
rehashidx++
    alt 查找类请求
        M->>H0: 先查旧表
        H0-->>M: 未命中(或已迁移为空)
        M->>H1: 再查新表
        H1-->>M: 命中返回
    else 新增 key 请求
        M->>H1: 只写入新表 ht[1]
        H1-->>M: 写入成功
    else 删除/修改请求
        M->>H0: 旧表有则改旧表
        M->>H1: 新表有则改新表
    end
    M-->>C: 返回结果（请求全程无感知 rehash）

这张时序图展示了"一次请求如何同时推进 rehash 又正确完成业务"：无论查/增/删/改，第一步永远是"顺带迁移一个桶"，然后按规则分流到双表。

22.2 工程要点

渐进迁移的推进者：不止请求，还有 serverCron

除了"每次请求顺带搬一格"，Redis 还有个后台定时任务 serverCron（见第二十五章），会在每个时间事件里调用 dictRehashMilliseconds，限时（默认 1ms）批量迁移若干桶——保证即使长时间没有写请求，rehash 也能持续推进直到完成。二者互补：

为什么"新增只写 ht[1]“如此关键

// 伪代码：dict 新增 key 的写入逻辑（理解用，非 Redis 源码）
func dictAdd(d *dict, key string, val string) {
    // 步骤1：如果正在 rehash，先把当前桶迁过去（推进 rehashidx）
    if d.rehashidx != -1 {
        dictRehashStep(d)
    }
    // 步骤2：新增 key 一律写到 ht[1]（目标表）
    //        ——绝不会写 ht[0]，否则 ht[0] 永远有数据、rehash 永不完
    if d.rehashidx != -1 {
        ht := &d.ht[1]
        ht.table[hash(key)&ht.sizemask] = newEntry(key, val)
        ht.used++
    } else {
        ht := &d.ht[0]
        ht.table[hash(key)&ht.sizemask] = newEntry(key, val)
        ht.used++
    }
}

缩容与扩容在处理上的异同

⚠️ 新手必踩的坑： rehash 期间如果某次请求触发了"顺带迁移一个桶”，而那个桶恰好很大（比如一个 bucket 链了成百上千个 key，哈希冲突严重），单次迁移也会稍慢。但相比"一次性搬完整个大表"，这已经是把尖峰削平成无数小坡——主线程最多卡这一下，不会长时间不可用。

二十三、其他底层数据结构与编码

23.1 用生活类比先建立直觉

第一章讲了 SDS 和跳表的"为什么"，但 Redis 的每种数据类型在不同数据量下会切换底层编码，像仓库会根据货物多少选不同货架：东西少时用"紧凑小货架"（省内存），东西多时换"标准大货架"（保性能）。面试常问的 encoding 就是"当前用哪种货架"。

flowchart TB
    S["String"] --> S1["int
整数直接存"]
    S --> S2["embstr
短字符串≤44字节
一次分配连续内存"]
    S --> S3["raw
长字符串>44字节
指针+独立buf"]
    L["List"] --> L1["quicklist
listpack 节点串成的双向链表"]
    H["Hash"] --> H1["listpack
字段少且短"]
    H --> H2["hashtable
字段多或长"]
    Set["Set"] --> Se1["intset
纯整数且量少"]
    Set --> Se2["hashtable
含非整数或量大"]
    Z["ZSet"] --> Z1["listpack
元素少"]
    Z --> Z2["skiplist + hashtable
元素多"]

这张图是 Redis “类型 → 编码"的总览。下面逐个讲清楚每种编码的适用场景。

23.2 工程要点

String 的三种编码：int / embstr / raw

embstr 的优势：一次内存分配、一次释放，且 cpu cache 友好（对象头和字符串连续）。超过阈值改成 raw，因为长字符串修改频繁，分开分配更灵活。注意：SDS 本身的原理（O(1) 长度、二进制安全、预分配）详见第一章 1.2 节，这里不再重复。

ziplist / listpack：紧凑顺序存储

ziplist（压缩列表）是 Redis 早期为省内存设计的连续内存结构，Hash / List / ZSet 在小数据量时都用它。但它有个致命缺陷：级联更新——一个节点长度变化可能导致后续所有节点的"前置长度字段"连锁扩容，最坏 O(N)。

Redis 5.0 后引入 listpack（紧凑列表）来替代 ziplist，解决了级联更新问题（用不同的长度编码方式，不向前依赖）。Redis 7.0 起 listpack 已全面取代 ziplist。

# 观察编码：用 OBJECT ENCODING 看一个 key 当前用哪种底层结构
redis-cli> HSET user:1 name tom age 18      # 小 Hash
redis-cli> OBJECT ENCODING user:1           # 返回 "listpack"（7.0+）或 "ziplist"
redis-cli> HSET user:1 f1 v1 f2 v2 ... f200 v200  # 超过阈值
redis-cli> OBJECT ENCODING user:1           # 返回 "hashtable"

intset：纯整数集合

Set 的元素全是整数且数量不多时，用 intset（整数集合）紧凑存储，按 int16/int32/int64 自适应升级。

redis-cli> SADD nums 1 2 3          # 纯整数、量少
redis-cli> OBJECT ENCODING nums     # 返回 "intset"
redis-cli> SADD nums "hello"        # 混入非整数
redis-cli> OBJECT ENCODING nums     # 升级为 "hashtable"

quicklist：List 的真实底层

第一章提到 List 的底层是 quicklist：它由多个 listpack 节点用双向指针串成的链表。这样既保留头尾 O(1) 操作，又避免普通链表每个元素都要存前后指针（太费内存）。

⚠️ 新手必踩的坑： 很多人背"List 底层是 ziplist”，这是 3.2 之前的旧答案。现在 List 统一是 quicklist，listpack 只是 quicklist 节点内部的存储格式（7.0 后是 listpack）。面试说"List 底层是 quicklist"才准确。

编码阈值配置（redis.conf）

hash-max-listpack-entries 128   # Hash 字段数 ≤128 用 listpack，超了转 hashtable
hash-max-listpack-value 64      # 字段值长度 ≤64 字节用 listpack
list-max-listpack-size -2       # quicklist 单节点大小（-2 = 8KB）
zset-max-listpack-entries 128   # ZSet 元素数 ≤128 用 listpack
set-max-intset-entries 512      # Set ≤512 个纯整数用 intset

调大这些阈值能"延迟"编码升级、更省内存，但单节点过大又会拖慢操作——是典型的内存/性能 tradeoff，和全文主旨一致。

二十四、跳表(skiplist)实现详解（腾讯 / 跟谁学高频）

24.1 用生活类比先建立直觉

第一章讲了跳表"怎么查"以及"为什么不用红黑树"。但面试（尤其腾讯、跟谁学）常追问一句：“跳表具体怎么实现的？插入和删除怎么写？” 核心就三步直觉：

1. 抛硬币定层数：新节点进来，抛硬币决定它"爬几层"——50% 概率在第 2 层、25% 第 3 层……直到没抛中。这样整张表天然维持"上层稀疏、下层稠密"。
2. 插入 = 先找落点再串指针：从最高层往下滑，每层找"最后一个小于目标"的节点记下来，到底层插入后，把各层记录的节点 forward 指针接到新节点上。
3. 删除 = 拔指针：找到节点后，把各层指向它的 forward 指针"跨过"它，再释放内存。

flowchart TD
    A["新节点 key=40 到来"] --> B["抛硬币决定层数
假设抛中第1、2层"]
    B --> C["从第2层Head出发
找最后一个 <40 的节点(30)"]
    C --> D["下降第1层
找最后一个 <40 的节点(30)"]
    D --> E["在30之后插入40
把30的forward改指向40"]
    E --> F["第2层:30.forward=40
第1层:30.forward=40
插入完成"]

这张图是插入的骨架：抛硬币定层 → 每层定位前驱 → 串指针。查找细节见第一章 1.2，本节聚焦"实现"。

24.2 工程要点

随机层数算法（Z 跳表核心）

Redis 的 zslRandomLevel 用"抛硬币"生成层数，期望层数约 1/(1-p)=2（p=0.25）：

// Redis 源码 zslRandomLevel（简化）
int zslRandomLevel(void) {
    int level = 1;
    // ZSKIPLIST_P = 0.25：每次有 25% 概率再升一层，上限 ZSKIPLIST_MAXLEVEL=32
    while ((random() & 0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level++;
    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

Go 实现一个最小跳表

package main

import (
    "fmt"
    "math/rand"
)

const skipListMaxLevel = 16
const skipListP = 0.25

// node 跳表节点
type node struct {
    score   float64
    member  string
    forward []*node // forward[i] 指向第 i 层的下一个节点
}

// skiplist 最小跳表（只演示结构，未含后退指针/span）
type skiplist struct {
    head  *node
    level int // 当前最大层数
}

func newSkiplist() *skiplist {
    return &skiplist{head: &node{forward: make([]*node, skipListMaxLevel)}}
}

// randomLevel 抛硬币决定新节点层数（与 Redis zslRandomLevel 同思路）
func randomLevel() int {
    level := 1
    for rand.Float64() < skipListP && level < skipListMaxLevel {
        level++
    }
    return level
}

// insert 插入 (score, member)
func (s *skiplist) insert(score float64, member string) {
    // 步骤1：update 记录每层"最后一个 < score"的前驱
    update := make([]*node, skipListMaxLevel)
    x := s.head
    for i := s.level - 1; i >= 0; i-- {
        for x.forward[i] != nil && x.forward[i].score < score {
            x = x.forward[i]
        }
        update[i] = x
    }
    // 步骤2：随机层数
    level := randomLevel()
    if level > s.level {
        for i := s.level; i < level; i++ {
            update[i] = s.head
        }
        s.level = level
    }
    // 步骤3：创建新节点，串接各层 forward 指针
    newNode := &node{score: score, member: member, forward: make([]*node, level)}
    for i := 0; i < level; i++ {
        newNode.forward[i] = update[i].forward[i]
        update[i].forward[i] = newNode
    }
    fmt.Printf("插入 %s(score=%.0f) 层数=%d\n", member, score, level)
}

// delete 删除指定 score 的节点
func (s *skiplist) delete(score float64) {
    // 步骤1：同样定位每层前驱
    update := make([]*node, skipListMaxLevel)
    x := s.head
    for i := s.level - 1; i >= 0; i-- {
        for x.forward[i] != nil && x.forward[i].score < score {
            x = x.forward[i]
        }
        update[i] = x
    }
    target := x.forward[0]
    if target == nil || target.score != score {
        return // 没找到
    }
    // 步骤2：把各层指向 target 的 forward "跨过" target
    for i := 0; i < s.level; i++ {
        if update[i].forward[i] != target {
            break
        }
        update[i].forward[i] = target.forward[i]
    }
    // 步骤3：若最高层空了，降级 level
    for s.level > 1 && s.head.forward[s.level-1] == nil {
        s.level--
    }
    fmt.Printf("删除 score=%.0f\n", score)
}

跳表 vs 平衡树：为什么 Redis 选跳表（实现视角补充）

第一章从"范围查询 / 实现复杂度 / 内存灵活"对比过，这里补一个实现者视角：

一句话总结给面试：跳表用"随机层数 + 多层链表"换来 O(logN) 查找/插入/删除，且插入删除只动局部指针、实现简单、范围遍历天然连续、并发易加锁，这就是 Redis 在 ZSet 里舍平衡树而取跳表的全部理由。

二十五、Redis 定时任务 serverCron

25.1 用生活类比先建立直觉

Redis 表面上"平时只处理命令"，但其实后台有个**“管家”**每隔一段时间巡检一遍：倒掉过期的垃圾（过期 key 清理）、记一下今天生意怎么样（统计）、看看搬家搬完没（推进 rehash）、到点了该不该做备份（触发 BGSAVE/AOF 重写）。这个管家就是 serverCron——一个挂在 Redis 时间事件上的周期性任务。

它的节奏由 hz 参数控制：hz 越大，巡检越勤（默认 10，即每 100ms 跑一次）；hz 越小，巡检越稀，但过期 key 清理可能不及时。

flowchart TB
    Tick["每 100ms（hz=10）
触发 serverCron"] --> T1["过期 key 清理
抽样删 + 活跃过期"]
    Tick --> T2["统计信息更新
ops/sec、内存、连接数"]
    Tick --> T3["推进渐进式 rehash
限时 1ms 迁移桶"]
    Tick --> T4["触发 BGSAVE / AOF 重写
按配置条件"]
    Tick --> T5["关闭超时客户端
清理空连接"]
    Tick --> T6["主从心跳 / Cluster 状态
发送 PING、检查 FAIL"]

这张图是 serverCron 每次"巡检"要做的事。注意它限时运行（不会一次干完所有活），保证不阻塞主线程。

25.2 工程要点

时间事件与 hz 参数

Redis 的事件循环（aeMain）里有两类事件：文件事件（处理客户端命令 IO）和时间事件（周期性任务，即 serverCron）。serverCron 通过返回"下次执行间隔"把自己重新挂回时间事件链表。

# redis.conf
hz 10          # 默认每秒 10 次（每 100ms 一次）
# 调大 hz（如 100）能让过期 key 清理更及时，但更占 CPU
dynamic-hz yes # 6.0+ 默认开启动态 hz：空闲时降低、繁忙时升高

serverCron 周期性执行的任务清单

过期 key 清理的两种路径

serverCron 里的 activeExpireCycle 是定期删除的主力（配合访问时的惰性删除，见第六章）：

// 伪代码：理解 serverCron 如何清理过期 key（非源码）
func activeExpireCycle() {
    // 步骤1：本轮最多花 25% 的 CPU 时间（由 hz 推算的时间预算）
    // 步骤2：循环抽取 20 个带 TTL 的 key
    for {
        samples := dictGetRandomKeys("expires", 20)
        expired := 0
        for _, k := range samples {
            if isExpired(k) {
                deleteKey(k) // 步骤3：过期就删
                expired++
            }
        }
        // 步骤4：若本轮过期比例 > 25%，再抽一批继续，直到时间预算用尽
        if expired < 20/4 {
            break
        }
    }
}

⚠️ 新手必踩的坑： serverCron 的过期清理是"抽样"的，不是全量扫描。如果瞬间大量 key 同时过期且一直没人访问，它们可能要等好几轮 serverCron 才被清掉——这段时间会占着内存。这就是为什么第六章强调"近似删除会漏、要配合淘汰策略"。

二十六、MongoDB vs Redis 区别

26.1 用生活类比先建立直觉

把两者放一起比，就像比较"便利店"和"档案室"：

• Redis（便利店）：所有商品摆在柜台（内存），随手拿、秒级结算；但柜台小，只放最热销的，结构单一（键值 + 几种数据类型）。适合"马上要用、量不大"的场景。
• MongoDB（档案室）：文件按抽屉归类存进柜子（磁盘），能按各种条件检索（文档查询），容量几乎无限；但存取要翻柜子，慢一些。适合"数据量大、结构灵活、要复杂查询"的场景。

flowchart LR
    subgraph R["Redis 内存 KV 库"]
        R1["数据在内存"] --> R2["键值 + 结构类型
极快、容量受内存限"]
    end
    subgraph M["MongoDB 文档数据库"]
        M1["数据在磁盘(BSON)"] --> M2["JSON 文档 + 丰富查询
容量大、结构灵活"]
    end
    R2 -->|"缓存 / 计数器 / 排行榜"| Use1["适用场景"]
    M2 -->|"业务主存 / 复杂查询"| Use2["适用场景"]

26.2 工程要点

核心区别

为什么面试喜欢拿它俩比

本质区别在于**“内存 vs 磁盘"和"KV vs 文档”**两条线：

• 要"快 + 简单结构 + 能丢"——选 Redis（缓存层）。
• 要"持久 + 复杂查询 + 大容量 + 事务"——选 MongoDB（存储层）。

实际架构里二者互补不替代：MongoDB 当主库存业务数据，Redis 当缓存扛热点读，缓解 MongoDB 的查询压力。

⚠️ 新手必踩的坑： 别把 Redis 当"唯一数据库"用——它本质是内存库，重启/故障有丢数据风险（即便开了持久化，也可能丢最多 1~2 秒）。MongoDB 也别当"缓存"用——它查得再快也比内存慢一个量级，扛不住超高并发热点。各司其职才是正解。

二十七、Redis 分布式锁在主节点宕机时的风险与 Redlock

27.1 用生活类比先建立直觉

类比：把单节点 Redis 分布式锁想象成小区门口唯一一间公共厕所——厕所门上挂一把锁，谁拿到钥匙谁用。现在出事了：管理员（主节点）拿着钥匙串去库房（宕机）了，还没来得及把钥匙复制一份交给副管理员（从节点）。副管理员被提拔成新管理员，但他手里没有那把钥匙的记录。这时候另一个人跑来要钥匙，新管理员说"我没记录你拿过，给你一把新的吧"——于是两个人同时拿到了厕所钥匙，锁彻底失效了。

这就是"单 master 锁丢失"的本质：锁的持有信息只存在主节点内存里，主从复制是异步的，主挂掉时从节点还没收到那条 SET 锁的命令，新主上任后根本不知道这把锁存在，于是另一客户端能再次加锁成功。

对应到工程里：Redis 主从复制是异步的（master 写入后立刻返回，不等 replica 确认）。如果此刻 master 宕机且锁还没同步到 replica，哨兵把 replica 提升为新 master，新 master 上没有这把锁，别的客户端就能再次加锁——同一把锁被两个客户端同时持有，互斥性被破坏。

sequenceDiagram
    participant A as 客户端A
    participant M as Master(主)
    participant R as Replica(从)
    participant B as 客户端B
    A->>M: SET lock uuid NX PX 30000
    M-->>A: OK（加锁成功）
    Note over M,R: 主从异步复制进行中...
    M--xR: 复制 SET 锁命令（还没到）
    Note over M: Master 宕机
    R->>R: 哨兵提升为 新Master
    B->>R: SET lock uuid2 NX PX 30000
    R-->>B: OK（新主上无锁记录）
    Note over A,B: 同一把锁被 A 和 B 同时持有！互斥失效

桥接：单 master 锁丢失 = “新官不认旧账”。只要锁的权威状态只在一台机器上、且同步有延迟，这台机器一挂，锁就可能有窗口期被重复获取。Redlock 的思路是用多台互相独立的机器投票，让"大多数都承认"才叫拿到锁，单台挂掉不至于让锁整体失效。

27.2 工程要点

Redlock 多节点多数派算法

Redlock 由 Redis 作者 antirez 提出，核心假设是：部署 N 个（通常 5 个）完全独立、互不复制的 Redis master 节点，客户端加锁时要向这 N 个节点都发起加锁请求，只有"在多数节点（≥ N/2+1）上成功，且总耗时小于锁有效期"才算加锁成功。

flowchart TD
    Start["客户端开始加锁"] --> T1["记录开始时间 T1"]
    T1 --> Try["依次向 5 个独立节点
发送 SET lock uuid NX PX"]
    Try --> Count["统计成功加锁的节点数"]
    Count -->|"成功数 >= 3 且
(T2-T1) < PX"| OK["加锁成功
有效时长 = PX - 耗时"]
    Count -->|"成功数 < 3 或 超时"| Fail["加锁失败
向所有节点发 DEL 释放"]
    OK --> Work["执行业务"]
    Work --> Release["向所有节点释放锁
（Lua 校验 uuid）"]
    Fail --> GiveUp["放弃，随机退避后重试"]

算法步骤（对应代码见下）：

1. 记录当前时间 T1（毫秒）。
2. 依次（建议并行）向 N 个节点发送 SET lock_key uuid NX PX ttl，每个请求带短超时（如 50ms）避免单点卡住。
3. 当且仅当：≥ N/2+1 个节点返回 OK，且从 T1 到收齐结果的总耗时 < ttl，锁才算获取成功。
4. 若失败（节点数不够或超时），立刻向所有节点发送释放脚本（Lua 校验 uuid 后 DEL），防止残留锁把别人挡住。
5. 锁的"有效时长"要扣除获取耗时：valid = ttl - (T2 - T1)。

⚠️ 新手必踩的坑： Redlock 在时钟漂移场景下被 DDIA 作者 Martin Kleppmann 公开质疑过：如果某个节点发生时钟跳跃（如 NTP 校正、GC 长时间停顿），可能导致锁过期后另一客户端仍能加锁成功。因此对强一致、丢锁不可接受（如金融转账）的场景，应直接用 ZooKeeper / etcd，而不是 Redis 锁。Redis 锁适合"性能优先、偶尔丢锁可接受"（如防重复提交、幂等保护）的场景。

Go 实现 Redlock 加锁与释放

下面是一份可运行的 Redlock 客户端骨架（用 go-redis 连接 5 个独立节点）：

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
)

// Redlock 持有多个独立 Redis 节点
type Redlock struct {
	nodes  []*redis.Client
	quorum int // 多数派门槛，通常 = len(nodes)/2 + 1
}

// 步骤1：构造 5 节点 Redlock，quorum = 3
func NewRedlock(addrs []string) *Redlock {
	nodes := make([]*redis.Client, 0, len(addrs))
	for _, a := range addrs {
		nodes = append(nodes, redis.NewClient(&redis.Options{Addr: a}))
	}
	return &Redlock{nodes: nodes, quorum: len(addrs)/2 + 1}
}

// tryLockOnNode 在单个节点上加锁（带短超时，避免单点阻塞）
func (r *Redlock) tryLockOnNode(ctx context.Context, node *redis.Client, key, val string, ttl time.Duration) bool {
	// 步骤2：每个节点最多等 50ms，整集群不能卡太久
	nodeCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
	defer cancel()
	ok, err := node.SetNX(nodeCtx, key, val, ttl).Result()
	return err == nil && ok
}

// Lock 向所有节点并行加锁，满足多数派才算成功
func (r *Redlock) Lock(ctx context.Context, key string, ttl time.Duration) (string, error) {
	val := uuid.New().String() // 步骤3：全局唯一 value，用于安全释放
	start := time.Now()

	// 步骤4：并行向所有节点请求加锁
	var wg sync.WaitGroup
	okCount := 0
	var mu sync.Mutex
	for _, node := range r.nodes {
		wg.Add(1)
		go func(n *redis.Client) {
			defer wg.Done()
			if r.tryLockOnNode(ctx, n, key, val, ttl) {
				mu.Lock()
				okCount++
				mu.Unlock()
			}
		}(node)
	}
	wg.Wait()

	// 步骤5：检查多数派，且总耗时 < 锁有效期
	elapsed := time.Since(start)
	if okCount < r.quorum || elapsed >= ttl {
		// 步骤6：失败立即释放所有节点上的锁，避免残留
		r.Unlock(ctx, key, val)
		return "", errors.New("redlock 获取失败：未达多数派或超时")
	}
	fmt.Printf("Redlock 获取成功: val=%s 有效时长=%v\n", val, ttl-elapsed)
	return val, nil
}

// Unlock 用 Lua 脚本在所有节点安全释放（先校验 value 再删除）
func (r *Redlock) Unlock(ctx context.Context, key, val string) {
	script := `
		if redis.call("GET", KEYS[1]) == ARGV[1] then
			return redis.call("DEL", KEYS[1])
		else
			return 0
		end`
	// 步骤7：遍历所有节点释放，忽略失败（已过期/不是自己的锁）
	for _, node := range r.nodes {
		node.Eval(ctx, script, []string{key}, val)
	}
}

func main() {
	addrs := []string{"127.0.0.1:7001", "127.0.0.1:7002", "127.0.0.1:7003", "127.0.0.1:7004", "127.0.0.1:7005"}
	rl := NewRedlock(addrs)
	ctx := context.Background()

	val, err := rl.Lock(ctx, "order:123:lock", 30*time.Second)
	if err != nil {
		fmt.Println("加锁失败:", err)
		return
	}
	// 步骤8：业务执行...（务必在 ttl 内完成，或配合看门狗续期）
	rl.Unlock(ctx, "order:123:lock", val)
}

看门狗续约（Watchdog）

Redlock 的 ttl 是固定的，如果业务执行时间超过 ttl，锁在多数节点上自动过期，互斥性又会失效。工业级方案用看门狗：加锁成功后启动一个后台协程，每隔 ttl/3 检查业务是否还在跑，是就对所有持有锁的节点 PEXPIRE 续期；业务结束就停掉看门狗并释放锁。

// Watchdog 看门狗：在锁有效期内周期性续约，防止业务跑太久锁过期
func (r *Redlock) LockWithWatchdog(ctx context.Context, key string, ttl time.Duration) (string, context.CancelFunc, error) {
	val, err := r.Lock(ctx, key, ttl)
	if err != nil {
		return "", nil, err
	}
	// 步骤1：用可取消的 context 控制看门狗生命周期
	watchCtx, cancel := context.WithCancel(ctx)

	// 步骤2：启动续约协程，每 ttl/3 续一次
	go func() {
		ticker := time.NewTicker(ttl / 3)
		defer ticker.Stop()
		for {
			select {
			case <-watchCtx.Done():
				return // 步骤3：业务结束，停止续约
			case <-ticker.C:
				// 步骤4：对所有节点续期（仅当仍是自己持有）
				script := `if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("PEXPIRE", KEYS[1], ARGV[2]) else return 0 end`
				for _, node := range r.nodes {
					node.Eval(watchCtx, script, []string{key}, val, int(ttl/time.Millisecond))
				}
			}
		}
	}()
	return val, cancel, nil
}

Redisson 的实现思路

Java 生态的 Redisson 把这些机制都封装好了，它的核心思路是：

• Hash 结构记录锁：lock_name -> {clientUUID:threadId : 重入次数}，天然支持可重入。
• 看门狗自动续期：加锁默认 30s 过期，Redisson 启动一个 scheduleExpirationRenewal 定时任务，每 10s（ttl/3）把锁续到 30s，直到主动 unlock 才停。
• Redlock 模式：RedissonRedLock 把多个独立 RedissonClient（连不同 master）组合，按多数派投票加锁，和上面 Go 骨架思路一致。
• 失败兜底：Lua 脚本保证加锁/释放/续期的原子性。

⚠️ 新手必踩的坑： 看门狗续约的前提是业务进程还活着。如果持有锁的进程被 kill -9 强杀，看门狗跟着死掉，锁会一直保留到 ttl 自然过期——这期间别的客户端拿不到锁。所以锁的 ttl 不要设太长，且业务要设计成"拿到锁后在 ttl 内必完成或能回滚"。

与 ZooKeeper / etcd 分布式锁对比

一句话总结：Redis 锁快但不绝对安全，ZooKeeper/etcd 慢一点但锁不会丢。选型看业务对"丢锁"的容忍度——秒杀防刷用 Redis 足矣，账户余额扣减这种"丢锁会出钱"的必须用 ZK/etcd。

二十八、Redis Cluster 扩槽（resharding）时防止 Kubernetes 探针误判 Pod 重启

28.1 用生活类比先建立直觉

类比：把 Redis Cluster 扩槽想象成"搬家公司在居民楼里逐户搬运家具"。搬运工（源节点）在搬运某一户（某个 slot 的 key）时，那户的房门会短暂锁上——这时候你去敲门（发 PING 探活），没人应，你就以为"这家人搬走了/出事了"，于是叫来铲车把房子推了（Kubernetes 因为探针失败重启 Pod）。但实际上人家只是正在搬一件大衣柜（迁移一个大 key），马上就好。

对应到工程里：Redis Cluster 做 resharding（扩槽/迁移 slot）时，源节点对正在迁移的 key 使用 MIGRATE 命令——这是个同步阻塞命令，迁移大 key 时源节点可能卡住几十毫秒甚至更久。如果此时 Kubernetes 的 liveness 探针用 redis-cli ping 探活，正好撞上迁移卡顿，连续几次超时就会被判定"不健康"，触发 Pod 重启——而重启又中断迁移，形成恶性循环。

sequenceDiagram
    participant K as K8s kubelet(探针)
    participant S as 源节点 Pod
    participant D as 目标节点
    K->>S: 每 periodSeconds 发 PING 探活
    S->>D: MIGRATE 大 key 中(同步阻塞)
    Note over S: 迁移期间无法及时响应 PING
    K--xS: 探针超时(连续 failureThreshold 次)
    K->>S: 判定不健康 → 重启 Pod
    Note over S,D: 正在进行的 resharding 被打断

这张时序图就是"探针误判"的链路：迁移卡顿 → 探活超时 → Pod 重启 → 迁移中断。下面给出根治方案。

28.2 工程要点

为什么扩槽会让节点"短暂无响应"

关键点：阻塞发生在 MIGRATE 执行期间，且是逐 key 串行的。一次 resharding 通常涉及成千上万个 key，但每个 key 的迁移都独立、短暂；真正危险的是单个超大 key（如一个存了几十万 field 的 Hash）的迁移。

方案一：调优探针阈值，给迁移留足余量

最稳妥的做法是让探针"更宽容"——即便节点偶尔卡几百毫秒，也不至于直接重启。

# redis-cluster-node-deployment.yaml（节选探针部分）
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: redis-cluster
spec:
  serviceName: redis-cluster
  replicas: 6
  template:
    spec:
      containers:
        - name: redis
          image: redis:7.2
          ports:
            - containerPort: 6379
            - containerPort: 16379
          # 步骤1：存活探针——判定"真死"才重启，给迁移留足余量
          livenessProbe:
            exec:
              command: ["redis-cli", "-h", "127.0.0.1", "-p", "6379", "ping"]
            initialDelaySeconds: 30      # 步骤2：启动后先等 30s 再探，避免冷启动误判
            periodSeconds: 10            # 步骤3：每 10s 探一次（别太频繁）
            timeoutSeconds: 5            # 步骤4：单次超时放宽到 5s（大于一次普通 MIGRATE 耗时）
            failureThreshold: 3          # 步骤5：连续 3 次失败才重启，给抖动留缓冲
          # 步骤6：就绪探针单独设置，只影响是否接流量，不重启 Pod
          readinessProbe:
            exec:
              command: ["redis-cli", "-h", "127.0.0.1", "-p", "6379", "cluster", "info"]
            initialDelaySeconds: 10
            periodSeconds: 5
            timeoutSeconds: 3
            failureThreshold: 1

⚠️ 新手必踩的坑： liveness 探针和 readiness 探针要分开配。很多人图省事共用一个，结果"短暂不可接流量"也变成了"重启 Pod"。正确做法是：readiness 失败只把 Pod 摘出服务端点（不再接新请求），liveness 失败才重启；resharding 期间应优先让 readiness 暂时失活，而不是让 liveness 杀进程。

方案二：分批迁移 + 迁移前先探活

不要用 redis-cli --cluster reshard 一次性迁移几千个 slot。改成小批量 + 每次迁移前确认节点健康：

#!/usr/bin/env bash
# safe-reshard.sh —— 把 slot 分批迁移，避免单次大迁移卡死探针
# 步骤1：定义参数
SRC_NODE="127.0.0.1:7000"
TOTAL_SLOTS=64          # 本次总共要迁移的 slot 数
BATCH=4                 # 每批只迁 4 个 slot（小步快跑）
COUNT=0

# 步骤2：循环按批迁移，每批之间检查源节点是否健康
while [ $COUNT -lt $TOTAL_SLOTS ]; do
  # 步骤3：迁移前先 PING，确认源节点能正常响应（留给探针的窗口）
  if ! redis-cli -h 127.0.0.1 -p 6379 ping | grep -q PONG; then
    echo "源节点无响应，跳过本轮，等下一个周期再试"
    sleep 10
    continue
  fi

  # 步骤4：计算本批槽范围并迁移（--cluster-yes 自动确认）
  redis-cli --cluster reshard 127.0.0.1:6379 \
    --cluster-from <src-node-id> \
    --cluster-to <dst-node-id> \
    --cluster-slots $BATCH \
    --cluster-yes

  COUNT=$((COUNT + BATCH))
  echo "已迁移 $COUNT / $TOTAL_SLOTS 个 slot"
  sleep 5   # 步骤5：批次之间歇一口气，让探针窗口恢复
done
echo "resharding 完成"

方案三：调大 cluster-node-timeout，降低迁移期误判

cluster-node-timeout 控制节点间心跳超时（默认 15s）。迁移期间若担心集群误判节点 FAIL，可适当调大（但不要过大，否则真故障发现慢）：

# redis.conf
cluster-node-timeout 20000   # 从 15s 提到 20s，给迁移卡顿留余量
cluster-migration-barrier 1  # 保证每个 master 至少 1 个副本可承载迁移流量

考点总结

Redis Cluster 扩槽（resharding）依靠 MIGRATE 同步迁移单 key，超大 key 会让源节点短暂阻塞；若 Kubernetes liveness 探针正好撞上阻塞并连续超时，会误杀 Pod、打断迁移。防御三板斧：① 调优探针（initialDelaySeconds/timeoutSeconds/failureThreshold 放宽，liveness 与 readiness 分离）；② 分批小步迁移并在每批前探活；③ 适度调大 cluster-node-timeout。核心思想：探针阈值要大于一次正常迁移抖动，且"摘流量"和"杀进程"必须用两套探针区分。

二十九、基于 Redis Operator 实现读写分离并限制只读节点最大内存

29.1 用生活类比先建立直觉

类比：把 Redis 集群想象成"图书馆 + 多个阅览室"。主节点（Master）是借还书柜台，负责写；只读副本（Replica）是若干个阅览室，只供读者（读请求）安静翻阅，不处理借还。如果阅览室里堆的书（内存）没有上限，读者越来越多、缓存越积越多，阅览室就被撑爆。所以我们既要让读请求分流到阅览室（读写分离），又要给每个阅览室挂一个"书架容量上限"（限制只读节点最大内存），满了就按规则清理旧书（maxmemory-policy）。

在 Kubernetes 上，我们用 Redis Operator（RedisLabs 开源的 redis-operator，即 OT-CONTAINER-KIT/redis-operator）以声明式 YAML 管理这套"柜台 + 阅览室"。

flowchart TB
    App[应用] -->|写请求| M[(Master 读写)]
    App -->|读请求| LB[读写分离路由
客户端 READONLY / 代理]
    LB --> R1[(Replica 只读
maxmemory 2gb)]
    LB --> R2[(Replica 只读
maxmemory 2gb)]
    M -->|异步复制| R1
    M -->|异步复制| R2

这张图是读写分离 + 内存上限的拓扑：写走 Master，读走 Replica；每个 Replica 用 maxmemory 封顶，超出按 allkeys-lru 淘汰，绝不把节点内存撑爆。

29.2 工程要点

RedisCluster CRD：开启副本 + 限制只读内存

下面这份 YAML 用 redis-operator 的 RedisCluster 资源，声明 3 主 3 从，并给所有节点（含只读副本）设置 maxmemory 与淘汰策略：

# redis-cluster-readwrite-split.yaml
apiVersion: redis.redis.opstreelabs.local/v1beta1
kind: RedisCluster
metadata:
  name: redis-read-cluster
  namespace: redis
spec:
  # 步骤1：3 个分片（主节点）
  size: 3
  # 步骤2：每个主节点挂 1 个只读副本，用于读写分离
  redisLeader:
    replicas: 3
    redisConfig:
      additionalConfig: |
        # 步骤3：主节点也设内存上限（兜底，避免写爆）
        maxmemory 2gb
        maxmemory-policy allkeys-lru        
    resources:
      requests:
        cpu: 250m
        memory: 2Gi
      limits:
        cpu: "1"
        memory: 2Gi
  redisFollower:
    replicas: 3
    redisConfig:
      additionalConfig: |
        # 步骤4：只读副本内存上限——核心诉求
        maxmemory 2gb
        # 步骤5：副本默认只读；超过 2gb 按 LRU 淘汰旧 key
        replica-read-only yes
        maxmemory-policy allkeys-lru        
    resources:
      requests:
        cpu: 250m
        memory: 2Gi
      limits:
        cpu: "1"
        memory: 2Gi
  # 步骤6：持久化（AOF）保证副本重建后可恢复
  persistence:
    size: 5Gi
    storageClassName: standard
  # 步骤7：探针同样放宽，避免复制/重启抖动被误杀
  kubernetesConfig:
    livenessProbe:
      initialDelaySeconds: 30
      periodSeconds: 10
      timeoutSeconds: 5
      failureThreshold: 3

⚠️ 新手必踩的坑： maxmemory 限制的是该节点 Redis 进程使用的内存上限，不是容器内存 limit。一定要让容器 memory limit 略大于 maxmemory（如上 2Gi limit vs 2gb maxmemory），否则 Redis 还没触发淘汰、容器就先被 OOMKill 了。经验值：容器 limit ≈ maxmemory × 1.2。

客户端如何"读走副本"（读写分离路由）

光有副本不够，客户端得真的把读请求发到副本才会分流。go-redis 的集群客户端用 RouteRandomly 或 READONLY 路由：

// 步骤1：构造集群客户端，开启"读请求随机打到副本"
rdb := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs:         []string{":7000", ":7001", ":7002"},
    RouteRandomly: true, // 步骤2：读命令随机路由到 master 或 replica，天然分流
    MaxRetries:   3,
    PoolSize:     20,
})

// 步骤3：对强一致要求的读，应连主节点单写客户端读取，避免读到旧值
// 步骤4：用只读命令让副本承接——客户端连接副本时先发 READONLY
func readFromReplica(ctx context.Context, replicaAddr string, key string) (string, error) {
    r := redis.NewClient(&redis.Options{Addr: replicaAddr})
    // 步骤5：告知副本"允许我读"，否则副本默认拒绝读命令
    r.ReadOnly(ctx)
    defer r.Close()
    return r.Get(ctx, key).Result()
}

注意：Redis Cluster 模式下，副本默认 replica-read-only yes，直接连副本读会报 (error) READONLY You can't write against a replica。客户端要先发 READONLY 命令"声明自己是只读客户端"，之后该连接上的读命令才会被副本处理——这就是读写分离在协议层的开关。

用 Redis Enterprise Operator（Redislabs 商业版）的等价写法

若使用 RedisLabs 的 Redis Enterprise Operator（RedisEnterpriseDatabase CRD），副本与内存上限通过以下字段声明：

# redis-enterprise-db.yaml（Redislabs 商业 Operator）
apiVersion: app.redislabs.com/v1
kind: RedisEnterpriseDatabase
metadata:
  name: rw-split-db
spec:
  memorySize: 2GB          # 步骤1：数据库总内存上限（含副本侧预算）
  replication: true        # 步骤2：开启副本——自动创建只读副本用于读扩展
  evictionPolicy: allkeys-lru  # 步骤3：内存满时按 LRU 淘汰
  shardCount: 3            # 步骤4：3 个分片，水平分散
  rackAware: true          # 步骤5：跨故障域分布，提升可用

考点总结

基于 Redis Operator 做读写分离，本质是"Master 负责写 + Replica 承接读 + 给每个节点（尤其是只读副本）设 maxmemory 上限并配 allkeys-lru 淘汰"。redis-operator 用 RedisCluster 的 redisFollower.redisConfig.additionalConfig 写 maxmemory 2gb + replica-read-only yes；Redis Enterprise 则用 RedisEnterpriseDatabase 的 memorySize + replication: true。客户端侧要靠 READONLY 命令或 RouteRandomly 把读流量真正引到副本，否则副本只是"热备"而非"分流"。容器 memory limit 必须略大于 maxmemory，防止 OOMKill 抢在淘汰之前。

三十、集群出现 hot key 时用 Redis + Kafka + Spark Streaming 实时打散

30.1 用生活类比先建立直觉

类比：想象一家网红奶茶店（hot key）门口排了 1000 人，全挤在一个收银台（单个 Redis key / 单个分区）前，收银员手速再快也扛不住。解决方法：在店门口装一个计数器 + 分流器（Kafka + Spark Streaming）——实时统计"哪个商品被点得最猛"（识别 hot key），一旦发现某个商品爆单，立刻在旁边多开几个临时收银台（把 hotkey 打散成 hotkey#0~hotkey#9 多个子 key），让顾客随机去不同台子结账（读请求随机打散到子 key）。

对应到工程里：单 key 访问过热（如秒杀商品、明星动态）会压垮单 Redis 节点 / 单 Kafka 分区。架构是 Redis（承载被打散的子 key）+ Kafka（收集访问事件流）+ Spark Streaming（窗口内实时统计并识别出 hot key，下发"打散策略"）。

flowchart LR
    Client[客户端访问 key] --> Redis[(Redis 热点 key)]
    Client -.->|埋点: 每次访问发事件| K[Kafka topic: key_access]
    K --> SP[Spark Streaming
窗口统计 key 频次]
    SP -->|识别 hot key
下发打散策略| Ctrl[Kafka topic: split_policy]
    Ctrl --> Writer[打散写入器
把 hotkey 拆成 N 个子key]
    Writer --> Redis2[(Redis: hotkey#0..#N)]
    Client --> Redis2

这张图是热 key 打散的完整链路：访问事件进 Kafka → Spark 窗口识别 → 下发打散策略 → 写入器把热 key 拆成多个子 key → 客户端随机读子 key，单点压力被均摊。

30.2 工程要点

步骤1：埋点——每次访问把事件发到 Kafka

客户端访问 key 时，异步把一条 {key, ts, op} 事件发到 Kafka（不影响主读路径）：

// 步骤1：访问埋点生产者——异步把 key 访问事件写入 Kafka
func reportAccess(producer *kafka.Writer, key string) {
    // 步骤2：事件体只含 key 名与时间戳，体积小、不阻塞业务
    event := []byte(fmt.Sprintf(`{"key":%q,"ts":%d}`, key, time.Now().UnixMilli()))
    // 步骤3：按 key 做分区，保证同一 key 的事件落到同一分区，便于 Spark 聚合
    producer.WriteMessages(context.Background(), kafka.Message{
        Key:   []byte(key),
        Value: event,
    })
}

// 步骤4：业务读路径——先上报再读，上报失败也不影响读
func GetWithHotProbe(rdb *redis.Client, producer *kafka.Writer, key string) (string, error) {
    go reportAccess(producer, key) // 异步埋点
    return rdb.Get(context.Background(), key).Result()
}

步骤2：Spark Streaming 窗口识别 hot key 并下发打散策略

用 PySpark Structured Streaming 读 Kafka，按 10 秒滚动窗口统计每个 key 的访问次数，超过阈值（如 5 万次/窗口）就判定为 hot key，把"打散为 N 份"的策略写入 split_policy topic：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window, count, expr

# 步骤1：建 SparkSession，启用 Kafka 数据源
spark = SparkSession.builder.appName("hotkey-detect").getOrCreate()

# 步骤2：从 Kafka 读访问事件，解析 JSON
raw = (spark.readStream.format("kafka")
       .option("kafka.bootstrap.servers", "kafka:9092")
       .option("subscribe", "key_access")
       .load())

events = (raw.selectExpr("CAST(key AS STRING) as key",
                          "CAST(value AS STRING) as payload")
          .select(from_json("payload",
                            "key STRING, ts LONG").alias("d"))
          .select("d.key", "d.ts"))

# 步骤3：10 秒滚动、5 秒滑动窗口内统计每个 key 的访问次数
counts = (events
          .withWatermark("ts", "10 seconds")
          .groupBy(window("ts", "10 seconds", "5 seconds"), "key")
          .agg(count("*").alias("cnt")))

# 步骤4：识别 hot key（阈值 50000 次/窗口），生成打散策略写回 Kafka
HOT_THRESHOLD = 50000
split = (counts.filter(col("cnt") > HOT_THRESHOLD)
         .select(expr("to_json(struct(key, cnt))").alias("value")))

# 步骤5：写入 split_policy topic，供打散写入器消费
query = (split.writeStream.format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("topic", "split_policy")
         .option("checkpointLocation", "/tmp/ckpt/hotkey")
         .outputMode("update")
         .start())
query.awaitTermination()