第 9 课：Redis 缓存与限流

学完能独立写出：

• 带连接池 + 启动自检的 Redis 客户端封装
• 正确处理 redis.Nil 的缓存读写层（go-redis 头号坑）
• 完整的 Cache-Aside 文章详情缓存，同时防住穿透、击穿、雪崩
• 基于 ZSet 的热榜 + 阅读量批量回写 MySQL
• 固定窗口 / 滑动窗口两种限流的 Gin 中间件
• 用 Lua 保证原子释放的分布式锁
• 登出即失效的 JWT 黑名单（呼应第 6 课的伏笔）

更重要的是能回答：「这里该不该加缓存？加了以后我承担了什么新风险？」

cd ~/go-blog
go run ./cmd/server &
curl -s http://localhost:8080/api/v1/posts/1 | head -c 120
kill %1

redis-cli ping
redis-cli INFO server | grep redis_version

PONG
redis_version:8.4.0

go get github.com/redis/go-redis/v9
go get golang.org/x/sync

go: added github.com/redis/go-redis/v9 v9.22.0
go: added golang.org/x/sync v0.22.0

GOTOOLCHAIN=local go get golang.org/x/sync    # ✅ 正常装上
GOTOOLCHAIN=local go get golang.org/x/crypto  # ❌ 立刻报错

go: golang.org/x/[email protected] requires go >= 1.26.0 (running go 1.25.0; GOTOOLCHAIN=local)

3.1 先问：这个数据该不该缓存

缓存不是免费的性能。你用一致性换延迟，同时给系统加了一个新的故障点。

一条止损规则：先看慢查询日志、先加索引。如果一条 SQL 加个索引就能从 800ms 降到 3ms，就别加缓存——你省下一次网络往返，换来一整套失效逻辑要维护。

本课选文章详情做主线：读多写少、有 JOIN（作者 + 标签）、能容忍旧数据（改了标题晚 10 分钟生效没人会死）。

而用户权限本课不缓存——你不会希望一个被封禁的用户因为缓存还能发帖 10 分钟。

3.2 go-redis 的 API 风格

每个命令返回一个 *XxxCmd 对象，而不是直接返回值：

cmd := rdb.Get(ctx, "key")   // *redis.StringCmd,还不是结果
val, err := cmd.Result()     // 取值 + 错误
val := cmd.Val()             // 只取值(错误被吞,慎用)
err := cmd.Err()             // 只取错误(适合 Set/Del)

日常写成一行：val, err := rdb.Get(ctx, "key").Result()。

为什么这么设计？因为 Pipeline：先把一串命令排队，Exec 之后再从各自的 *XxxCmd 里取结果。限流会用到。

3.3 redis.Nil 不是错误 —— 本课头号坑

GET 一个不存在的 key 时，go-redis 返回哨兵错误 redis.Nil。实测它的文本是：

redis: nil

如果按平时习惯写：

// ❌ 未命中被当成故障上抛
val, err := rdb.Get(ctx, key).Result()
if err != nil {
    return nil, fmt.Errorf("查缓存失败: %w", err)
}

后果：任何一次缓存未命中都变成 500。而且上层看到的文本就是 redis: nil，你在日志里根本分不清是「数据不在缓存里」（完全正常）还是「Redis 连不上了」（真事故）。

// ✅ 未命中必须单独判断
val, err := rdb.Get(ctx, key).Result()
if err != nil {
    if errors.Is(err, redis.Nil) {
        return nil, ErrCacheMiss    // 正常,走回源
    }
    return nil, fmt.Errorf("查缓存失败: %w", err)   // 这才是故障
}

用 errors.Is 而不是 err == redis.Nil。两者在当前版本实测都是 true，但一旦中间某层用 %w 包装过，== 就失效了。

注意：只有 GET 类命令会返回 redis.Nil。DEL 删不存在的 key 返回 0，EXISTS 返回 0，都不报错。

4.1 NewRedis —— 连接池 + 启动自检

// internal/cache/redis.go
package cache

import (
    "context"
    "fmt"
    "time"

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

// NewRedis 建立客户端并立刻验证连通性。
// 为什么必须 Ping:redis.NewClient 是"懒"的,它只构造对象不建连接,
// 地址写错也不报错,要等第一个用户来了才炸。启动时 Ping,配置错就在启动阶段崩。
func NewRedis(addr, password string, db int) (*redis.Client, error) {
    rdb := redis.NewClient(&redis.Options{
        Addr:     addr,     // "127.0.0.1:6379"
        Password: password, // 本机开发为空
        DB:       db,       // 0~15,开发和测试用不同 DB 互不干扰

        PoolSize:     10, // 常驻连接数。博客量级 10 足够
        MinIdleConns: 2,  // 保底空闲连接,避免流量突增时现建连接(要一次 TCP 握手)

        DialTimeout:  3 * time.Second,
        ReadTimeout:  2 * time.Second, // Redis 正常 <1ms,2s 还不回说明它已经不健康了
        WriteTimeout: 2 * time.Second,
        PoolTimeout:  4 * time.Second, // 池满时等空闲连接最多等多久

        ConnMaxIdleTime: 5 * time.Minute, // 防止被中间 LB 悄悄断开后拿到坏连接
        MaxRetries:      2,               // ⚠️ -1 才是禁用重试,0 是"用默认值 3"
    })

    // 自检用独立短超时:不该被全局超时影响,也不该无限挂着
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    if err := rdb.Ping(ctx).Err(); err != nil {
        rdb.Close() // 失败要把已建的连接还回去,别泄漏
        return nil, fmt.Errorf("redis ping %s: %w", addr, err)
    }
    return rdb, nil
}

易错点：MaxRetries: 0 不是「不重试」。 go-redis 约定 0 = 用默认值 3 次，-1 才是禁用。你以为关掉了，实际一个超时的 INCR 会被悄悄重发三遍——限流计数就莫名多加了。

易错点：*redis.Client 并发安全，全局共用一个。 它内部就是连接池。不要在 handler 里 NewClient——那等于每个请求建一个新池，几百个请求就把 Redis 的 maxclients 打爆，报 ERR max number of clients reached。

4.2 key 设计：blog:v1:post:{id}

func postKey(id int64) string { return fmt.Sprintf("blog:v1:post:%d", id) }

冒号是社区事实标准，很多可视化工具会按冒号折叠成树。

绝不要把不可控输入直接拼进 key：

key := "blog:v1:search:" + c.Query("q")   // ❌ 用户传 100KB 的 q,你就存一个 100KB 的 key
                                          //    传一百万个不同的 q,内存就被撑爆(缓存污染)
sum := sha256.Sum256([]byte(q))           // ✅ 定长可控
key := "blog:v1:search:" + hex.EncodeToString(sum[:8])

序列化用 encoding/json 就够了：可读（redis-cli GET 直接看得懂，排查问题时极有用）、无依赖、性能不是瓶颈。等真测出序列化是热点了再换 msgpack（体积小 20~30%）或 protobuf（最快但要写 .proto）。别为省 0.1ms 提前付出可读性和构建复杂度。

Marshal   → {"id":1,"status":1,"published_at":"2026-09-06T10:00:00Z"}
Unmarshal → Status=1 (类型 model.Status), p.Status == model.StatusPublished → true

4.3 PostCache.Get —— 正确处理 redis.Nil

// internal/cache/post_cache.go
package cache

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "math/rand/v2"
    "time"

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

    "blog/internal/model"
)

// 缓存层自己的哨兵错误:让 service 用 errors.Is 分辨两种情况。
// 不要把 redis.Nil 漏给 service —— service 不该知道底下是 Redis。
var (
    ErrCacheMiss = errors.New("cache miss") // 没有,去回源
    ErrCachedNil = errors.New("cached nil") // 明确记着"这条不存在",别回源
)

const nilPlaceholder = "__NIL__" // 正常 JSON 绝不可能等于它

type PostCache struct {
    rdb     *redis.Client
    baseTTL time.Duration // 基础存活时间
    jitter  time.Duration // 随机抖动上限,防雪崩
    nilTTL  time.Duration // 空值占位的存活时间,必须很短
}

func NewPostCache(rdb *redis.Client) *PostCache {
    return &PostCache{rdb: rdb, baseTTL: 10 * time.Minute, jitter: 2 * time.Minute, nilTTL: 60 * time.Second}
}

func postKey(id int64) string { return fmt.Sprintf("blog:v1:post:%d", id) }

// Get 读缓存。四种返回:
//   (post, nil)          命中
//   (nil, ErrCachedNil)  命中"不存在"占位,直接 404,不回源
//   (nil, ErrCacheMiss)  没命中,去 DB 回源
//   (nil, 其他 error)    Redis 真出问题了 → 调用方应降级查 DB,而不是返回 500
func (c *PostCache) Get(ctx context.Context, id int64) (*model.Post, error) {
    raw, err := c.rdb.Get(ctx, postKey(id)).Result()
    if err != nil {
        // ★ 全课最关键的三行:未命中必须和真故障分开
        if errors.Is(err, redis.Nil) {
            return nil, ErrCacheMiss
        }
        return nil, fmt.Errorf("cache get post %d: %w", id, err)
    }
    if raw == nilPlaceholder {
        return nil, ErrCachedNil
    }

    var p model.Post
    if err := json.Unmarshal([]byte(raw), &p); err != nil {
        // 反序列化失败 = 缓存里是脏数据(通常是改了结构体没升 key 版本)。
        // 删掉再当作未命中,让下次读把正确值写回来。
        // ⚠️ 千万不要 return error —— 一条脏缓存会让这个 id 永久 500
        _ = c.rdb.Del(ctx, postKey(id)).Err()
        return nil, ErrCacheMiss
    }
    return &p, nil
}

func (c *PostCache) Set(ctx context.Context, p *model.Post) error {
    b, err := json.Marshal(p)
    if err != nil {
        return fmt.Errorf("marshal post %d: %w", p.ID, err)
    }
    return c.rdb.Set(ctx, postKey(p.ID), b, c.ttl()).Err()
}

// SetNil 写"这条不存在"的短命占位,防穿透。
func (c *PostCache) SetNil(ctx context.Context, id int64) error {
    return c.rdb.Set(ctx, postKey(id), nilPlaceholder, c.nilTTL).Err()
}

func (c *PostCache) Del(ctx context.Context, id int64) error {
    return c.rdb.Del(ctx, postKey(id)).Err()
}

// ttl 返回 base + [0, jitter) 的随机时长,防雪崩。
// math/rand/v2 是 Go 1.22 新包:rand.N 是泛型的,而且不再需要手动 Seed。
func (c *PostCache) ttl() time.Duration {
    return c.baseTTL + time.Duration(rand.N(int64(c.jitter)))
}

4.4 Cache-Aside 模式与写路径

Cache-Aside（旁路缓存）：缓存不参与写，应用代码自己在读时填、写时删。

读路径                                  写路径
┌──────────┐                          ┌──────────┐
│ 请求到达  │                          │ 更新请求  │
└────┬─────┘                          └────┬─────┘
     ▼                                     ▼
┌──────────┐  命中  ┌────────────┐   ┌──────────────┐
│ 查 Redis  ├──────►│ 反序列化返回 │   │ 1. 写 MySQL   │
└────┬─────┘        └────────────┘   └────┬─────────┘
     │ 未命中(redis.Nil)                   ▼
     ▼                                ┌──────────────┐
┌──────────┐  没有  ┌──────────────┐  │ 2. 删 Redis   │
│ 查 MySQL  ├──────►│写空值占位(60s)│  └────┬─────────┘
└────┬─────┘        │  → 返回 404   │       ▼ (可选)
     │ 查到          └──────────────┘  ┌──────────────────┐
     ▼                                │ 3. 延迟 500ms 补删 │
┌──────────────────┐                  └──────────────────┘
│ 写回 Redis(带TTL) │
└────┬─────────────┘
     ▼
   返回结果

删缓存还是更缓存？主流选删

1. 并发写会互相覆盖。 A、B 同时改同一篇文章：

   A: 写DB(标题=甲) ─────────────► A: 写缓存(标题=甲)
   B:      写DB(标题=乙) ──► B: 写缓存(标题=乙)
   结果:DB 里是乙,缓存里是甲,一直错到 TTL 到期
   

   「写 DB」和「写缓存」之间没有原子性。改成删就没这问题——删除是幂等的，删两次和删一次结果一样。

2. 写完不一定有人读。 每次写都更新缓存，等于为可能永远没人访问的数据付出序列化 + 网络往返。

3. 更缓存要求写路径知道完整的缓存结构。 缓存的是「文章 + 作者名 + 标签列表」的聚合视图时，改一个标签名就得知道怎么重建整个视图。删不需要——回源逻辑只有一份，在读路径里。

先更 DB 还是先删缓存？推荐先更 DB

先更 DB 也有一个很窄的漏洞：

t0: 读请求 R 未命中,查 DB 读到旧值 V1
t1: 写请求 W 把 DB 更新为 V2
t2: W 删缓存(此时缓存本来就空,删了个寂寞)
t3: R 才把它 t0 读到的 V1 写进缓存   ← 脏数据诞生

要求 R 的「读 DB → 写缓存」耗时长于 W 的整个流程，概率很低但不是零。延迟双删就是给这个窗口打的补丁：

// internal/service/post_service.go

func (s *PostService) Update(ctx context.Context, p *model.Post) error {
    // 1. 先更 DB。DB 是唯一真相源
    if err := s.repo.Update(ctx, p); err != nil {
        return err
    }

    // 2. 立刻删缓存。用 context.WithoutCancel(Go 1.21+):
    //    客户端此时断开会取消 ctx,那就变成"DB 改了缓存没删" → 脏数据。
    //    删缓存必须跑完,不能被请求生命周期拖累
    if err := s.cache.Del(context.WithoutCancel(ctx), p.ID); err != nil {
        // 删失败不阻塞主流程:DB 已经对了,最多脏到 TTL 到期。记日志让人查得到
        slog.Warn("删缓存失败,将由 TTL 兜底", "post", p.ID, "err", err)
    }

    // 3. 延迟补删:清掉可能被并发读请求写回去的旧值。
    //    延迟多久?略大于"一次读请求查 DB + 写缓存"的耗时,一般 300ms~1s
    time.AfterFunc(500*time.Millisecond, func() {
        ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
        defer cancel()
        _ = s.cache.Del(ctx, p.ID)
    })
    return nil
}

4.5 三大经典问题

三个词长得像，讲的是三件完全不同的事：

记忆法：穿透是「没有这个数据」，击穿是「一个点破了」，雪崩是「一片全塌了」。

穿透：攻击者遍历 id=999999

场景很具体：有人跑 for i in $(seq 1 10000000); do curl /api/v1/posts/$i; done。你库里只有 1000 篇文章，于是 999 万次请求全部「未命中 → 查 MySQL → 没有 → 404」。缓存一次都没起作用，MySQL 独自承受全部流量。

• 方案零（最该先做）：参数合法性校验。 id <= 0 直接拒绝，连 Redis 都不查。零成本，别忘了。
• 方案一：缓存空值（本课采用）。查不到就写一个短命占位。优点是五行代码立刻见效；代价是占内存，所以 TTL 必须很短（本课 60s）且要和正常 TTL 分开配置。注意：新建文章后要主动删掉对应 id 的占位，否则新文章 60 秒内查不到。
• 方案二：布隆过滤器。 启动时把所有存在的 ID 灌进去，请求先问「可能存在吗」，答「绝对不存在」就直接 404。优点是内存极省（一百万 ID 约 1MB）、能挡无限量无效 ID；代价是有假阳性（会漏一点到 DB，可接受）、标准布隆不支持删除、要额外引依赖（RedisBloom 或 bits-and-blooms/bloom）。

什么时候升级到布隆：当空值占位的 key 数量超过真实数据数量时。博客这个体量，缓存空值足够。

击穿：singleflight

一篇被首页推荐的文章 QPS 5000，缓存 TTL 到了：

        TTL 到期
           │
5000 并发 ─┼─► 5000 次全部未命中 ─► 5000 条一模一样的 SQL 同时打到 MySQL
           │                          │
           ▼                          ▼
                             连接池耗尽 → 后续请求全超时 → 服务雪崩

这 5000 次查询结果完全一样，为什么要查 5000 次？让第一个去查，其余 4999 个等着共享结果。这正是 golang.org/x/sync/singleflight（Go 官方扩展库）干的事：

v, err, shared := g.Do("post:7", func() (any, error) {
    return queryDB(7)   // 同一 key 同一时刻只有一个 goroutine 跑到这里
})

实测（100 goroutine 同时请求同一篇文章）：

✅ singleflight: 100 并发 → repo 实际被调用 1 次

三个返回值：v any（需类型断言）、err error、shared bool（结果是否被共享，实测 50 并发时 50 个全是 true，包括真正执行的那个，可以拿来打点观测击穿有多严重）。

三个必须知道的坑（均已实测）：

1. 错误会被共享——一次失败，所有等待者拿到同一个错误：

   一次失败 → 10 个 caller 全部拿到同一个错误
   

   所以 fn 里不要做重试，重试放在 Do 外面。

2. 第一个调用者的 ctx 取消会连累所有人（fn 闭包捕获的是第一个调用者的 ctx）：

   第一个 caller 取消后: err1=context canceled err2=context canceled (第二个 caller 无辜受害)
   

   这可接受：受害者只是拿到一个错误，重试即可，不会有数据问题。彻底解决要用 DoChan + select，先别做，是过度设计。

3. 只在单进程内生效。 部署 4 个实例就是 4 次回源，不是 1 次——这通常完全够了（4 条 SQL vs 5000 条）。

其他两种方案（了解即可）：互斥锁（只放一个 goroutine 回源，其他人 sleep 后重试读缓存，比 singleflight 笨但零依赖）；逻辑过期（key 永不物理过期，value 里存 expire_at，发现逻辑过期就返回旧值 + 异步刷新，永远不会有请求打到 DB，代价是会返回旧数据）。

雪崩：TTL 抖动

服务重启后你预热了 1000 篇热门文章，全部 Set(..., 10*time.Minute)。10 分钟后这 1000 个 key 在同一秒集体过期，MySQL 瞬间收到 1000 条并发查询。

修法极简单，就是 4.3 里那个 ttl()：

return c.baseTTL + time.Duration(rand.N(int64(c.jitter)))   // 10min + [0,2min)

实测：

$ redis-cli TTL blog:v1:post:1
608

608 秒 = 10 分 8 秒，落在 600~720 区间。多跑几次会得到不同值——抖动生效了。抖动幅度取 base 的 10%~30%：太小起不到摊平作用，太大会让有效期不可预测。

另一种雪崩：Redis 整个挂了。 抖动救不了。这时需要的是降级——Get 返回非 redis.Nil 的错误时不要 500，记 warn 日志然后直接查 DB。这就是 4.3 里要把 ErrCacheMiss 和真故障分成两种错误的原因：缓存挂了应该是性能问题，不该是可用性问题。

4.6 PostService.GetByID —— 本课高光函数

三种防护全在这一个函数里：

// internal/service/post_service.go
type PostService struct {
    repo  repository.PostRepository // 第 4 课的接口,service 不知道底下是 GORM
    cache *cache.PostCache
    sf    singleflight.Group // 零值可用。⚠️ 它含锁,PostService 必须用指针传递
}

// GetByID 全站 QPS 最高的路径,三种防护都在这里:
//   防穿透 —— 查不到时写空值占位
//   防击穿 —— singleflight 让并发回源收敛成一次
//   防雪崩 —— TTL 带随机抖动(在 cache.Set 里)
func (s *PostService) GetByID(ctx context.Context, id int64) (*model.Post, error) {
    // ---- 1. 参数校验:零成本挡掉一批垃圾请求,连 Redis 都不查 ----
    if id <= 0 {
        return nil, model.ErrNotFound
    }

    // ---- 2. 查缓存 ----
    p, err := s.cache.Get(ctx, id)
    switch {
    case err == nil:
        return p, nil // 命中。99% 的请求走这条路

    case errors.Is(err, cache.ErrCachedNil):
        // 命中"不存在"占位:刚才有人查过,DB 里确实没有。
        // 直接 404 不回源 —— 这就是防穿透生效的地方
        return nil, model.ErrNotFound

    case errors.Is(err, cache.ErrCacheMiss):
        // 正常未命中,继续往下回源

    default:
        // ★ Redis 真出问题了(超时/内存满/被 kill)。
        // 关键决策:降级,不是报错。缓存挂了服务还得能用,只是慢一点
        slog.Warn("缓存降级", "post", id, "err", err)
    }

    // ---- 3. singleflight 收敛回源 ----
    // key 用业务语义即可,singleflight 是进程内的,不需要 blog:v1: 前缀
    v, err, _ := s.sf.Do(fmt.Sprintf("post:%d", id), func() (any, error) {
        // ⚠️ 这个闭包在高并发下只会被执行一次,其余调用者阻塞等结果

        post, err := s.repo.GetByID(ctx, id)
        if err != nil {
            if errors.Is(err, model.ErrNotFound) {
                // 防穿透:DB 说没有,把这个"没有"缓存 60 秒。
                // WithoutCancel:即使调用方 ctx 已取消,这个占位也值得写进去
                if e := s.cache.SetNil(context.WithoutCancel(ctx), id); e != nil {
                    slog.Warn("写空值占位失败", "post", id, "err", e)
                }
                return nil, model.ErrNotFound
            }
            return nil, fmt.Errorf("查文章 %d: %w", id, err)
        }

        // 回填缓存。写失败不影响本次返回 —— 数据已拿到,大不了下次再回源
        if e := s.cache.Set(context.WithoutCancel(ctx), post); e != nil {
            slog.Warn("回填缓存失败", "post", id, "err", e)
        }
        return post, nil
    })
    if err != nil {
        return nil, err // 注意:同一批等待者共享这同一个 error
    }

    // singleflight 返回 any 必须断言。用不带 ok 的形式:
    // 这里断言失败说明代码写错了,让它 panic 比返回静默的 nil 更容易在测试阶段暴露
    return v.(*model.Post), nil
}

易错点：PostService 必须用指针传递。 singleflight.Group 内含 sync.Mutex。handler 收值类型的话，每次复制结构体就复制了一份锁，go vet 直接报：

./post_service.go:12:6: assignment copies lock value: contains sync.Mutex

看到 copies lock value 就是这个。含锁的结构体一律用指针。

易错点：context.WithoutCancel 没有 deadline。 它保留 ctx 的 value（比如 request_id）但去掉取消信号，同时也丢掉了 deadline（实测 Deadline() 返回 ok=false）。写缓存有 WriteTimeout: 2s 兜底还好，长操作记得再套一层 WithTimeout。

4.7 热榜 ZSet 与阅读量回写

为什么不用 ORDER BY view_count

问题不在慢，而在它和阅读量的写入互相拖累：

1. 每次阅读都要 UPDATE posts SET view_count = view_count + 1 WHERE id = ? —— 行锁 + 写盘 + 刷 binlog。QPS 5000 就是 5000 次并发更新同一行，全部串行排队等行锁。
2. 你想给 view_count 加索引来加速排序，但这列每秒被更新几千次，索引维护开销比排序省下的还多。
3. 排序结果每秒都在变，缓存也没法缓。

Redis 的 ZSet 天生干这个：ZINCRBY 是 O(log N) 原子递增，ZREVRANGE 取 TopN 是 O(log N + M)。

// internal/cache/post_cache.go
const hotKey = "blog:v1:hot:posts"

func viewKey(id int64) string { return fmt.Sprintf("blog:v1:views:%d", id) }

// IncrView 记一次阅读。Pipeline 把三条命令打成一个网络往返。
// 完全不碰 MySQL —— 一次阅读的成本从"一条 UPDATE 加行锁"降到"一次 Redis 往返"。
func (c *PostCache) IncrView(ctx context.Context, id int64) error {
    pipe := c.rdb.Pipeline()
    pipe.Incr(ctx, viewKey(id))                             // 待回写 MySQL 的增量
    pipe.ZIncrBy(ctx, hotKey, 1, strconv.FormatInt(id, 10)) // 热榜累加
    pipe.Expire(ctx, hotKey, 7*24*time.Hour)                // 防止冷门文章永远赖在榜上
    _, err := pipe.Exec(ctx)
    return err
}

// TopN 取热榜前 N。返回 []redis.Z,Z 是 { Score float64; Member interface{} }。
// ⚠️ Member 的动态类型是 string(实测确认),不是 int64 ——
//    Redis 里一切都是字符串,写进去 int64 取出来也是 string。
// ⚠️ 区间是闭区间:要前 10 个写 0, 9。写 0, 10 会拿到 11 条。
func (c *PostCache) TopN(ctx context.Context, n int64) ([]redis.Z, error) {
    return c.rdb.ZRevRangeWithScores(ctx, hotKey, 0, n-1).Result()
}

ZIncrBy 的签名容易记错，实测确认是 增量在前、成员在后：

ZIncrBy(ctx context.Context, key string, increment float64, member string) *redis.FloatCmd

实测输出：

✅ 热榜 TopN 返回 []redis.Z: {member=2(string) score=12} {member=1(string) score=5} {member=3(string) score=1}

批量回写：SCAN + GETDEL + 事务

// internal/job/flush_views.go

// FlushViews 把 Redis 累积的阅读增量批量刷进 MySQL,由每分钟一次的定时任务调用。
func FlushViews(ctx context.Context, rdb *redis.Client, db *gorm.DB) (int, error) {
    // ---- 1. 扫出待回写的 key ----
    // ⚠️ 用 SCAN 不要用 KEYS。Redis 是单线程的,KEYS 会阻塞整个实例直到遍历完 ——
    //    在生产库上执行 KEYS * 是把线上打挂的标准姿势。SCAN 是游标式分批的
    var keys []string
    var cursor uint64
    for {
        batch, next, err := rdb.Scan(ctx, cursor, "blog:v1:views:*", 200).Result()
        if err != nil {
            return 0, fmt.Errorf("scan view keys: %w", err)
        }
        keys = append(keys, batch...)
        cursor = next
        if cursor == 0 { // 游标回到 0 表示遍历完一圈
            break
        }
    }

    // ---- 2. 取值并原子清零 ----
    type delta struct{ id, n int64 }
    var deltas []delta
    for _, k := range keys {
        // GETDEL 是"取出并删除"的原子命令(Redis 6.2+)。
        // ⚠️ 为什么不用 GET + DEL 两条?两条之间可能有新的 INCR 进来,
        //    你 DEL 掉的是"已读到的 + 刚新增的",那次新增就永久丢了
        raw, err := rdb.GetDel(ctx, k).Result()
        if errors.Is(err, redis.Nil) {
            continue // 已被上一轮取走,正常
        }
        if err != nil {
            return 0, fmt.Errorf("getdel %s: %w", k, err)
        }
        n, err := strconv.ParseInt(raw, 10, 64)
        if err != nil || n == 0 {
            continue
        }
        id, err := strconv.ParseInt(k[strings.LastIndex(k, ":")+1:], 10, 64)
        if err != nil {
            continue
        }
        deltas = append(deltas, delta{id, n})
    }
    if len(deltas) == 0 {
        return 0, nil
    }

    // ---- 3. 一个事务里批量更新 ----
    // 用 view_count + ? 而不是 = ?:即使有别的路径也在改,增量也不会被覆盖丢失
    err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        for _, d := range deltas {
            if err := tx.Model(&model.Post{}).Where("id = ?", d.id).
                UpdateColumn("view_count", gorm.Expr("view_count + ?", d.n)).Error; err != nil {
                return err
            }
        }
        return nil
    })
    if err != nil {
        return 0, fmt.Errorf("flush views: %w", err)
    }
    return len(deltas), nil
}

实测（本机 blog_dev 真实执行）：

回写 3 条增量 → MySQL 现状: [id=1 "Hello Go" view_count=13] [id=2 "database/sql 入门" view_count=34] [id=3 "这是一篇草稿" view_count=1]

诚实说明代价：进程在 GETDEL 之后、事务提交之前崩了，那一批阅读量就丢了。阅读量丢几个没人在意，所以这个取舍划算。换成「账户余额」，这套方案一秒都不能用。

Pipeline 还是 TxPipeline？ TxPipeline 把命令包在 MULTI/EXEC 里，保证中间不插入别人的命令，但 Redis 事务不支持回滚。实测 Pipeline 中一条命令报错时：

Exec err=ERR value is not an integer or out of range; 后续命令仍执行 after="written"

Exec 返回第一个错误，但其余命令依然执行了。所以要「要么全做要么全不做」，Redis 事务给不了你，得用 Lua（见 4.9）。日常批量发命令用 Pipeline 就行，它只是省网络往返。

4.8 限流中间件

// internal/middleware/ratelimit.go

// RateLimit 固定窗口限流:每个 (路由, IP) 在 window 内最多 limit 次。
func RateLimit(rdb *redis.Client, limit int64, window time.Duration) gin.HandlerFunc {
    return func(c *gin.Context) {
        ctx := c.Request.Context()
        // ⚠️ 用 c.FullPath() 拿路由模板 "/api/v1/posts/:id" 而不是 c.Request.URL.Path,
        //    否则每个不同的 id 都会产生一个独立的限流 key,限流形同虚设
        key := fmt.Sprintf("blog:v1:rl:%s:%s", c.FullPath(), c.ClientIP())

        // ⚠️ INCR 和 EXPIRE 必须打包成 Pipeline。分两次发送时,
        //    进程在两次之间崩了,这个 key 就永远没有 TTL,该 IP 被永久封禁
        //    —— 这是真实发生过的线上事故
        pipe := rdb.Pipeline()
        incr := pipe.Incr(ctx, key)
        pipe.Expire(ctx, key, window)
        if _, err := pipe.Exec(ctx); err != nil {
            // ★ 限流器自己挂了该放行还是拒绝?
            // 本课选 fail-open:限流是保护措施,不该因为它坏了就让全站不可用。
            // 反过来,如果限流是安全边界(防暴力破解登录),就该 fail-closed 直接拒绝
            slog.Warn("限流器不可用,放行", "err", err, "key", key)
            c.Next()
            return
        }

        n := incr.Val() // Pipeline Exec 之后才能取值
        c.Header("X-RateLimit-Limit", strconv.FormatInt(limit, 10))
        c.Header("X-RateLimit-Remaining", strconv.FormatInt(max(limit-n, 0), 10))

        if n > limit {
            ttl, _ := rdb.TTL(ctx, key).Result()
            c.Header("Retry-After", strconv.Itoa(int(ttl.Seconds())+1))
            // ⚠️ 用 AbortWithStatusJSON。只写 c.JSON 不 Abort 的话,业务 handler 照样执行
            // ⚠️ 这里不能 import handler 去用 APIError —— 会 import cycle,见下方易错点。
            // 中间件自己拼,但字段名必须和 APIError{Code,Message,Field} 一致
            c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
                "code": "RATE_LIMITED", "message": "请求太频繁,请稍后再试",
            })
            return
        }
        c.Next()
    }
}

实测（limit=5，连打 20 次）：

✅ 限流中间件: 20 次请求 → map[200:5 429:15]
✅ 第 20 次响应头: X-RateLimit-Remaining=0 Retry-After=11

易错点：为什么中间件要自己拼 JSON，而不是复用 APIError

第 2 课定稿了对外错误契约 APIError{Code, Message, Field}，第 4 课把它和 httpStatusFor(err) 一起放进了 internal/handler/errors.go。那这里为什么不直接 handler.APIError{Code: "RATE_LIMITED"}，反而要手拼一个 gin.H？

因为那样整个项目编译不过。 真实报错：

package blog/internal/handler
    imports blog/internal/middleware from errors.go
    imports blog/internal/handler from ratelimit.go: import cycle not allowed

拆开看这个环是怎么成的：

第 5 课已经有的边:  handler ──────► middleware
                    (errors.go 里 writeErrorFrom 要调 middleware.RequestIDFrom
                     把 request_id 写进错误响应)

你现在想加的边:    middleware ──────► handler
                    (ratelimit.go 想用 handler.APIError)

合起来:            handler ⇄ middleware        ← 环

根因是 Go 的语言级硬约束：包依赖必须是有向无环图（DAG）。 Go 没有 C 的前向声明、没有 Java 那种「同包不同文件随便互相引用」的宽松度，也没有任何逃生舱——不存在什么编译选项能让循环 import 通过。这不是编译器保守，是 Go 刻意的设计：它保证了任何一个包都能被独立编译和理解，也保证了初始化顺序（init() 和包级变量）永远是确定的。

值得停下来想一下的是：这恰好是第 4 课「依赖方向」那一节在编译期被强制执行。 当时我们约定 handler → service → repository 单向依赖，那还只是一条纪律，靠人自觉。而这里 Go 直接告诉你：箭头画歪了，不让编译。架构约束从「文档里的规矩」变成了「编译器的红线」——这是 Go 包机制最被低估的价值。

三种解法及取舍：

选 ① 的代价要说清楚：{"code","message"} 这个形状现在有两份定义（handler.APIError 一份、中间件手拼一份）。哪天要给契约加字段，两处都得改，而且编译器不会提醒你漏了哪个。项目再大一点就该上 ②。

固定窗口的临界问题

限流规则: 每 60 秒 100 次

      窗口A结束 │ 窗口B开始
   [ 窗口A 0~60s ]│[ 窗口B 60~120s ]
                 │
   59.9s 打 100 次┤ 60.1s 又打 100 次
                 └──► 这 0.2 秒内实际通过了 200 次

边界处实际通过量可以达到限额的 2 倍。 防刷场景通常可接受，但如果下游只能扛 100 QPS，这个毛刺会打挂它。

滑动窗口：ZSet

把每次请求的时间戳作为 score 存进 ZSet，每次先删窗口外的旧记录，再数窗口内还剩多少条：

// internal/middleware/ratelimit.go
func allowSliding(ctx context.Context, rdb *redis.Client, key string, limit int64, window time.Duration) (bool, error) {
    now := time.Now()
    minScore := strconv.FormatInt(now.Add(-window).UnixMilli(), 10) // 窗口左边界

    pipe := rdb.Pipeline()
    pipe.ZRemRangeByScore(ctx, key, "0", minScore) // 1. 清窗口外的,不清 ZSet 会无限增长
    pipe.ZAdd(ctx, key, redis.Z{                   // 2. 记本次请求
        Score: float64(now.UnixMilli()),
        // ⚠️ member 必须唯一,否则同一毫秒的两个请求会被 ZAdd 去重成一条
        Member: fmt.Sprintf("%d-%d", now.UnixNano(), rand.Int64()),
    })
    card := pipe.ZCard(ctx, key)  // 3. 数窗口内有多少条
    pipe.Expire(ctx, key, window) // 4. 兜底:之后没人访问就自己消失

    if _, err := pipe.Exec(ctx); err != nil {
        return true, err // fail-open
    }
    return card.Val() <= limit, nil
}

实测同样 limit=5、20 次请求：滑动窗口限流 limit=5: 20 次拦掉 15 次。

先用固定窗口，确实观测到边界毛刺造成问题再换。

按 IP 还是按用户？ 未登录接口（注册/登录/搜索）按 IP —— 注意 c.ClientIP() 依赖 X-Forwarded-For，必须先用 router.SetTrustedProxies() 配好可信代理，否则客户端伪造这个头就能绕过。已登录接口按 user_id（从第 6 课的 JWT claims 拿），换 IP 也照样被限。最好两个都加：IP 层挡扫描，用户层挡滥用。

4.9 分布式锁：SetNX + Lua 原子释放

场景：4.7 的 FlushViews。你部署了 3 个实例，每分钟都跑一次——三个实例同时 GETDEL 同一批 key，会重复回写或互相把对方读到的数据删掉。

// internal/cache/lock.go
var ErrLockNotHeld = errors.New("锁不是自己持有的(可能已超时被别人抢走)")

type Lock struct {
    rdb   *redis.Client
    key   string
    value string // ★ 持有者唯一标识,是安全释放的关键
}

// Acquire 尝试获取锁,返回 (锁, 是否拿到, 错误)。
func Acquire(ctx context.Context, rdb *redis.Client, key string, ttl time.Duration) (*Lock, bool, error) {
    // value 必须全局唯一。生产上更常用 UUID
    value := fmt.Sprintf("%d-%d", time.Now().UnixNano(), rand.Int64())

    // SET key value NX PX ttl —— go-redis 里就是 SetNX,返回 (bool, error)。
    // NX = 只在 key 不存在时设置。这一条命令天然原子,不会有竞态。
    // ★ TTL 必须设!没 TTL 的锁,持有者崩溃后就永远锁死了 —— 死锁
    ok, err := rdb.SetNX(ctx, key, value, ttl).Result()
    if err != nil {
        return nil, false, fmt.Errorf("acquire lock %s: %w", key, err)
    }
    if !ok {
        return nil, false, nil // 没抢到不是错误,是正常的竞争结果
    }
    return &Lock{rdb: rdb, key: key, value: value}, true, nil
}

// unlockScript:校验 + 删除必须原子。
// Redis 单线程执行 Lua,期间不会插入别的命令,所以"比较后删除"是原子的。
var unlockScript = redis.NewScript(`
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end
`)

func (l *Lock) Unlock(ctx context.Context) error {
    // Script.Run 签名: Run(ctx, 客户端, keys []string, args ...any) *redis.Cmd
    // ⚠️ keys 是切片、args 是可变参数,顺序别搞反。
    // Run 内部先试 EVALSHA(只发哈希省带宽),失败自动回退到 EVAL 发全文
    res, err := unlockScript.Run(ctx, l.rdb, []string{l.key}, l.value).Int64()
    if err != nil {
        return fmt.Errorf("unlock %s: %w", l.key, err)
    }
    if res == 0 {
        // 锁已经不是自己的了 = 业务执行时间超过 TTL,锁被别人抢走。
        // 这是严重信号:刚才可能有两个实例在同时跑临界区。必须报出来
        return ErrLockNotHeld
    }
    return nil
}

为什么释放必须校验 value —— 「误删别人的锁」

实例 A: Acquire(ttl=30s) 成功,开始跑 FlushViews
        │  ← 因为一次 GC / 网络抖动 / 慢 SQL,任务跑了 35 秒
    30s ┤ 锁自动过期,Redis 里的 key 没了
        ├──► 实例 B: Acquire 成功(key 不存在了),开始跑 FlushViews
    35s ┤ 实例 A 终于跑完,执行 DEL lock_key
        │        ↓ 把 B 的锁删掉了!
        ├──► 实例 C: Acquire 成功,也开始跑
        └──► 现在 B 和 C 在同时跑临界区,锁完全失效

Unlock 里如果是裸 DEL，这个事故一定会发生。加上 value 校验后，A 的 GET 拿到的是 B 的 value，对不上就不会删。

为什么校验和删除必须原子：

// ❌ 看起来对,实际有竞态
val, _ := rdb.Get(ctx, key).Result()
if val == l.value {
    // ← 就在这一行,锁过期了,B 抢到锁并写了自己的 value
    rdb.Del(ctx, key)  // 删的是 B 的锁
}

「读」和「删」之间有时间窗口。Lua 在 Redis 里单线程、不可打断，中间插不进别的命令。这就是必须用 Lua 的原因。

实测：

✅ SetNX 锁: 第一次 ok=true 第二次 ok=false
✅ 别人的锁 Unlock → lock not held (ErrLockNotHeld=true)
✅ 自己的锁 Unlock 成功, key 剩余 0 个

用法：

lock, ok, err := cache.Acquire(ctx, rdb, "blog:v1:lock:flush-views", 60*time.Second)
if err != nil { slog.Error("获取锁失败", "err", err); return }
if !ok { slog.Debug("锁被别的实例持有,本轮跳过"); return } // 不是错误,是正常的
defer func() {
    if err := lock.Unlock(ctx); err != nil {
        slog.Error("释放锁异常,可能发生过并发执行", "err", err) // 这条值得告警
    }
}()
n, err := FlushViews(ctx, rdb, db)

诚实说明这把锁的边界

这是简化版，只在单点 Redis 上正确。 三个缺陷：

1. 没有续期（watchdog）。 TTL 到了业务还没跑完，锁就没了。缓解：TTL 设成「业务最长耗时的 3 倍」。成熟方案（redsync、Java 的 Redisson）会起后台 goroutine 定期 PEXPIRE 续期。
2. 主从切换会丢锁。 Redis 主从复制是异步的：A 在 master 拿到锁，master 还没同步就挂了，slave 升主后这把锁不存在，B 也能拿到。
3. Redlock 有争议。 Redis 作者提出的 Redlock（向 N 个独立节点申请，过半即成功）被分布式系统专家 Martin Kleppmann 公开质疑过，核心争论是它依赖各节点时钟不会大幅漂移。至今没有共识。

结论：这把锁适合「重复执行只是浪费资源、不会导致数据错误」的场景（定时任务去重、缓存预热）。如果重复执行会扣两次钱，别用 Redis 锁——用 etcd / ZooKeeper（有 lease 和真正的一致性保证），或者更好的办法是把业务做成幂等的，让重复执行本身无害。

4.10 JWT 黑名单（呼应第 6 课）

第 6 课留的问题：JWT 是无状态的，签发出去就没法撤回。用户点「退出登录」，那个 token 在过期前依然完全有效；token 被盗了你除了等它过期什么也做不了。

Redis 的标准答案：维护「已作废 token」名单，TTL 正好设成 token 的剩余有效期——token 自然过期的那一刻，黑名单记录也自动消失，不占额外内存。

前提：第 6 课签发时要在 claims 里带 jti（JWT ID）。当时没加现在补上。

// internal/cache/blacklist.go
func revokeKey(jti string) string { return "blog:v1:jwt:revoked:" + jti }

// Revoke 拉黑一个 token。exp 是它自己的过期时间。
func Revoke(ctx context.Context, rdb *redis.Client, jti string, exp time.Time) error {
    // ★ TTL = 剩余有效期。设固定值的话:设长了浪费内存,设短了有安全漏洞
    //   (记录没了但 token 还有效)。设成剩余有效期,不多不少
    ttl := time.Until(exp)
    if ttl <= 0 {
        return nil // 本来就过期了,不用记
    }
    return rdb.Set(ctx, revokeKey(jti), 1, ttl).Err() // value 存什么无所谓,存 1 最省
}

func IsRevoked(ctx context.Context, rdb *redis.Client, jti string) (bool, error) {
    // 用 EXISTS 不用 GET:只关心存不存在,不返回 value 省带宽;
    // 而且 EXISTS 对不存在的 key 返回 0 而不是 redis.Nil,少一层错误判断
    n, err := rdb.Exists(ctx, revokeKey(jti)).Result()
    if err != nil {
        return false, fmt.Errorf("check jwt blacklist: %w", err)
    }
    return n > 0, nil
}

改造第 6 课的鉴权中间件，验签通过后多查一次 Redis：

// internal/middleware/auth.go(在第 6 课基础上增补)
claims, err := parseToken(c, secret) // 第 6 课写的:取 header、验签、查过期
if err != nil { /* 401 */ }

// ★ 新增:签名有效不代表没被撤销
revoked, err := cache.IsRevoked(c.Request.Context(), rdb, claims.ID)
if err != nil {
    // ★ 这里的降级决策和限流相反:必须 fail-closed。
    // 查不到黑名单就放行,等于给攻击者一条"打挂 Redis 即可绕过登出"的路。
    // 认证相关的降级永远选拒绝
    slog.Error("黑名单查询失败,拒绝请求", "err", err)
    c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{"code": "UNAVAILABLE", "message": "服务暂时不可用"})
    return
}
if revoked {
    c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"code": "TOKEN_REVOKED", "message": "登录已失效,请重新登录"})
    return
}

实测：JWT 黑名单: revoked=true other=false ttl=2h0m0s

一个诚实的取舍：加了黑名单之后，JWT 就不再是无状态的了——每次鉴权都要查一次 Redis，抵消掉了 JWT 相对 session 的主要优势。

那还用 JWT 吗？用。查 Redis 是一次 EXISTS（亚毫秒，且大概率不命中，走的是最快路径），而 session 每次都要读出完整会话数据；而且 JWT 的 claims 已经带了 user_id、角色，不用额外查库。

替代方案：把 access token 缩短到 15 分钟、用 refresh token 换新，登出时只作废 refresh token（数量少得多），access token 最多再活 15 分钟。这是 OAuth2 的标准做法，平衡更好但实现更复杂。

先打开 GORM 的 SQL 日志（第 8 课学过），这样能直接看到有没有查数据库：

gorm.Open(mysql.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Info)})

5.1 缓存真的生效了吗

redis-cli DEL blog:v1:post:1
curl -s http://localhost:8080/api/v1/posts/1 > /dev/null   # 第一次
curl -s http://localhost:8080/api/v1/posts/1 > /dev/null   # 第二次

服务端日志只应出现一次 SELECT：

[0.812ms] [rows:1] SELECT * FROM `posts` WHERE `posts`.`id` = 1 AND `posts`.`deleted_at` IS NULL
                    ← 第二次请求这里什么都没有,说明命中了缓存

第二次也打 SQL 的话，检查三处：cache.Set 的错误是不是被忽略了、key 两次算出来是不是一样、errors.Is(err, redis.Nil) 那个分支是不是漏写了。

5.2 直接看 Redis 里存了什么

redis-cli KEYS 'blog:v1:*'      # ⚠️ 仅限本地开发,线上用 SCAN
redis-cli TTL  blog:v1:post:1   # 应在 600~720 之间,多跑几次会看到不同的值
redis-cli GET  blog:v1:post:1
redis-cli ZREVRANGE blog:v1:hot:posts 0 9 WITHSCORES

实测输出：

blog:v1:post:1
blog:v1:hot:posts
blog:v1:views:1
blog:v1:jwt:revoked:abc-123
blog:v1:rl:/api/v1/posts/:id:127.0.0.1

608
{"id":1,"title":"Hello Go","content":"正文","view_count":3,"created_at":"2026-09-06T09:33:16-07:00"}

2
12
1
5
3
1

用 JSON 序列化的好处就在这里：GET 出来肉眼可读。

5.3 防穿透

redis-cli DEL blog:v1:post:999999
curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8080/api/v1/posts/999999   # → 404
redis-cli GET blog:v1:post:999999   # → "__NIL__"
redis-cli TTL blog:v1:post:999999   # → 60  (不是 600!)

再打几次这个 404，SQL 日志里不应再出现新的 SELECT。

5.4 防击穿

redis-cli DEL blog:v1:post:1
seq 1 100 | xargs -P 100 -I{} curl -s -o /dev/null http://localhost:8080/api/v1/posts/1

数一下 SQL 日志里 WHERE id = 1 出现几次：没有 singleflight 时是几十次，有的话应该是 1~2 次（可能 2 次，因为 100 个请求不是严格同时到达）。单元测试里更好观测，实测 100 并发 → repo 实际被调用 1 次。

5.5 限流

for i in $(seq 1 20); do curl -s -o /dev/null -w "%{http_code} " http://localhost:8080/api/v1/posts/1; done; echo
curl -s -D - -o /dev/null http://localhost:8080/api/v1/posts/1 | grep -iE 'ratelimit|retry-after'

200 200 200 200 200 429 429 429 429 429 429 429 429 429 429 429 429 429 429 429
X-RateLimit-Limit: 5
X-RateLimit-Remaining: 0
Retry-After: 11

等窗口过去 sleep 11 再试应该恢复 200。

5.6 分布式锁（两个终端）

# 终端 1
redis-cli SET blog:v1:lock:test mytoken NX PX 30000        # → OK
# 终端 2:抢不到
redis-cli SET blog:v1:lock:test othertoken NX PX 30000     # → (nil)
# 终端 2:用错误 value 释放,被拒绝
redis-cli EVAL 'if redis.call("GET",KEYS[1])==ARGV[1] then return redis.call("DEL",KEYS[1]) else return 0 end' 1 blog:v1:lock:test othertoken   # → 0
# 终端 1:用正确 value 释放
redis-cli EVAL 'if redis.call("GET",KEYS[1])==ARGV[1] then return redis.call("DEL",KEYS[1]) else return 0 end' 1 blog:v1:lock:test mytoken      # → 1

EVAL 语法是 EVAL 脚本 numkeys key1... arg1...，中间那个 1 是「接下来有几个参数属于 KEYS」。写错这个数字是新手用 EVAL 最常见的错误。

5.7 JWT 黑名单

TOKEN=$(curl -s -X POST http://localhost:8080/api/v1/auth/login -H 'Content-Type: application/json' \
  -d '{"username":"alice","password":"secret123"}' | jq -r .token)

curl -s -o /dev/null -w "%{http_code}\n" -H "Authorization: Bearer $TOKEN" http://localhost:8080/api/v1/me   # → 200
curl -s -X POST -H "Authorization: Bearer $TOKEN" http://localhost:8080/api/v1/auth/logout
curl -s -o /dev/null -w "%{http_code}\n" -H "Authorization: Bearer $TOKEN" http://localhost:8080/api/v1/me   # → 401

练习 1：给标签列表加缓存（★☆☆）

// internal/cache/tag_cache.go
// TODO(练习1): 实现标签列表缓存
//   1. key 用 "blog:v1:tags:all"(整个列表存一个 key,不要每标签一个)
//   2. TTL 30 分钟 + 抖动
//   3. 新建/删除标签时 Del 这个 key
//   4. redis.Nil 必须正确处理
func (c *TagCache) GetAll(ctx context.Context) ([]model.Tag, error) { panic("TODO") }

验收：curl /api/v1/tags 打两次，SQL 日志只出现一次 SELECT * FROM tags；redis-cli TTL blog:v1:tags:all 返回 1800~1900；新建标签后立刻再查，新标签能出现。

练习 2：缓存命中率统计（★★☆）

你现在无法回答「我的缓存到底有没有用」。

// internal/cache/metrics.go
// TODO(练习2): 用 HINCRBY 记录命中/未命中
//   1. key 用 "blog:v1:stats:cache:{YYYY-MM-DD}",field 用 "hit"/"miss"/"error"
//   2. 给 key 设 7 天 TTL,别让统计数据无限堆积
//   3. 加 GET /admin/cache/stats 返回今日命中率
//   4. 打点本身失败了只记日志,绝不影响主流程返回
func (c *PostCache) recordHit(ctx context.Context)  { /* TODO */ }
func (c *PostCache) recordMiss(ctx context.Context) { /* TODO */ }

验收：打 100 次同一篇文章后 redis-cli HGETALL blog:v1:stats:cache:2026-09-06 显示 hit≈99 miss=1；/admin/cache/stats 返回 {"hit_rate":0.99,...}；把 Redis 停掉（本机才能这么干：redis-cli SHUTDOWN NOSAVE）后服务仍返回 200 走 DB 降级，而不是 500。

练习 3：按路由配置的限流（★★☆）

type LimitRule struct {
    Limit  int64
    Window time.Duration
    By     string // "ip" | "user"
}

// TODO(练习3): 实现按路由配置的限流
//   1. POST /api/v1/auth/login: 5 次/分钟,按 IP,fail-closed(防暴力破解)
//   2. POST/PUT /api/v1/posts:  30 次/分钟,按 user
//   3. 读接口:                   300 次/分钟,按 IP,fail-open
//   4. 没配规则的路由不限流
func RateLimitByRule(rdb *redis.Client, rules map[string]LimitRule) gin.HandlerFunc { panic("TODO") }

验收：连续用错密码登录 6 次，第 6 次返回 429 而不是 401；同一用户在两个 IP 发文章共享同一份 30 次/分钟额度；停掉 Redis 后登录接口返回 503、文章读接口仍返回 200。

练习 4：热榜时间衰减（★★★）

现在 score 只增不减，三年前的爆款会永远霸占第一。

// internal/job/decay_hot.go
// TODO(练习4): 实现热榜时间衰减
// 方案 A(简单): 每小时把所有 score 乘 0.95(ZRANGE 取全部 + Pipeline 批量 ZADD)
// 方案 B(推荐): 分时间桶 —— key 用 "blog:v1:hot:posts:{YYYYMMDDHH}",
//              查 TopN 时 ZUNIONSTORE 合并最近 24 个桶,越近权重越高
//   1. 至少实现一个方案
//   2. 衰减任务必须用 4.9 的分布式锁保护,多实例部署时只有一个在跑
//   3. score 低于 0.5 的成员用 ZREMRANGEBYSCORE 清掉,别让 ZSet 无限增长
func DecayHotRank(ctx context.Context, rdb *redis.Client) error { panic("TODO") }

验收：手动刷某篇到 score=100，跑一次后 ZSCORE 返回 95；造一个 score=0.3 的成员，跑完后它消失；同时起两个进程跑，ZSCORE 只被衰减一次（是 95 不是 90.25）。

练习 5：缓存降级开关（★★★）

线上出事故时你需要一个能立刻关掉缓存的开关，而不是改代码重新发布。

// internal/cache/switch.go
// TODO(练习5): 运行时可切换的缓存开关
//   1. 状态存在 Redis 的 "blog:v1:switch:cache"(值 "on"/"off")
//   2. 用 sync/atomic 在内存里缓存状态,后台每 5 秒刷新一次
//      —— 绝不能每个请求都去 Redis 查开关,那本身就是一次额外往返
//   3. off 时 GetByID 直接走 DB,完全跳过缓存读写
//   4. 提供 POST /admin/cache/switch 切换(需管理员权限)
//   5. Redis 查不到开关时默认 "on"(缺省应该是正常状态)
type Switch struct{ /* TODO */ }

验收：redis-cli SET blog:v1:switch:cache off 后 5 秒内所有请求都打 SQL 日志；切回 on 后 5 秒恢复命中；go test -race 并发读开关无竞态报告；停掉 Redis 后开关按内存里最后的值继续工作，不 panic。

判断与基础

• ☐ 我能说出三个「不该加缓存」的具体场景，并解释为什么
• ☐ 我会先看慢查询日志、先加索引，再考虑缓存
• ☐ import 路径是 github.com/redis/go-redis/v9，不是 go-redis/redis
• ☐ *redis.Client 全局共用一个，不在 handler 里 NewClient
• ☐ 启动时 Ping 自检；连接池四个超时（Dial/Read/Write/Pool）都设了
• ☐ 我知道 MaxRetries: 0 是「用默认值 3」，-1 才是禁用

redis.Nil

• ☐ 我知道 Get 未命中返回的 redis.Nil 不是错误，用 errors.Is 判断
• ☐ 缓存层把「未命中」和「Redis 故障」包成两种不同的哨兵错误
• ☐ 我能分清 redis.Nil（缓存没有，回源）/ model.ErrNotFound（数据库没有，写空值占位）/ cache.ErrCachedNil（缓存记着没有，直接 404）
• ☐ 我知道 model.ErrNotFound 在 model 包不在 repository 包
• ☐ Redis 故障时服务降级查 DB，而不是返回 500

Cache-Aside

• ☐ key 有业务前缀 + 版本号；改结构体或改 model.Status 枚举值都要升版本号
• ☐ 不可控输入不直接拼进 key（先 hash 再截断）
• ☐ 每个 key 都设了 TTL，没有 Set(..., 0)
• ☐ 更新时是删缓存不是更缓存，我能说出至少两个理由
• ☐ 顺序是先更 DB 再删缓存，我能说出反过来的具体后果
• ☐ 延迟双删的 AfterFunc 里用的是新 context，不是请求的 ctx

三大问题

• ☐ 我能一句话说清穿透 / 击穿 / 雪崩的区别
• ☐ 穿透：缓存空值 TTL 很短（60s），新建数据时要删占位
• ☐ 击穿：singleflight，我知道错误共享和 ctx 传染两个坑
• ☐ 雪崩：TTL 加 10%~30% 抖动；Redis 整挂要靠降级不是抖动
• ☐ singleflight.Group 含锁，service 是指针传递（go vet 不报 copies lock value）

ZSet、限流、锁

• ☐ 我知道为什么不用 ORDER BY view_count（行锁 + 索引维护成本）
• ☐ ZRevRangeWithScores 是闭区间，前 10 个写 0, 9；redis.Z.Member 是 string
• ☐ 遍历用 SCAN 不用 KEYS；取增量用 GETDEL 不用 GET+DEL
• ☐ INCR 和 EXPIRE 打包成 Pipeline（分开发送可能永久封禁某个 IP）
• ☐ 限流用 AbortWithStatusJSON，key 里用 c.FullPath() 而不是实际路径
• ☐ 中间件的错误响应自己拼且字段名对齐 APIError{Code,Message,Field}；我知道 middleware 不能 import handler（import cycle）
• ☐ 我知道限流该 fail-open、认证该 fail-closed，并能说出原因
• ☐ SetNX 一定带 TTL；锁 value 唯一；释放用 Lua 原子完成「比较 + 删除」
• ☐ 我能讲清「误删别人的锁」的完整时间线
• ☐ 我知道这把锁没有续期、主从切换会丢锁，只适合「重复执行无害」的场景

JWT 黑名单

• ☐ TTL 设为 token 的剩余有效期，不是固定值；用 EXISTS 不用 GET
• ☐ 鉴权 fail-closed（Redis 挂了返回 503，不放行）
• ☐ 我知道加了黑名单后 JWT 就不再无状态了，并理解这个取舍

下一课：第 10 课是收官课——配置管理、log/slog 结构化日志、优雅关闭、go test + httptest、并发基础、构建部署。我们会把这 10 课的所有组件在一个 main() 里装配起来，然后打成一个能扔到服务器上跑的二进制。