Redis 专题：把它用成"数据加速层"，而不是又一个数据库

本模块所有命令、配置、代码围绕 Redis 8.10.2（2026-09-17 安全发布线）编写；Go 客户端锁定 go-redis v9.23.0。这是全系列版本纪律的执行结果（08 §7），不是让你再做选择。

1.1 许可现状（知道即可，不影响本教程）

Redis 自 8.0 起采用 RSALv2 / SSPLv1 / AGPLv3 三许可任选的商业源码模式（不再是 OSI 认证的开源协议）。社区已分叉出 Valkey 9.0.6（Linux 基金会主导，活跃）。对你的影响：

• 个人学习、内部自用、做缓存与队列——免费照用 Redis 官方版，无任何商业风险。
• 若将来要把"托管 Redis 服务"作为产品对外售卖，需要重新评估 SSPL 条款；本教程场景不涉及。
• 教程一律按 Redis 官方走；哪天想换 Valkey，命令与 go-redis 客户端完全兼容，配置文件路径一致，替换成本极低。

本机（Windows + Docker Desktop 4.94）用容器，干净可删；1GB Linux 服务器不装 Docker（内存不够，versions.md 明确），直接用发行版包跑 systemd。

2.1 路线 A：本机 Docker Compose（推荐开发用）

在你喜欢的项目目录建 redis-docker/docker-compose.yml，内容如下（密码、端口、卷全部写死，照抄即可）：

# docker-compose.yml —— 本机开发用 Redis 8.10，带密码 + 持久化卷
services:
  redis:
    image: redis:8.10
    container_name: redis810
    restart: unless-stopped
    ports:
      - "6379:6379"
    command: ["redis-server",
              "--requirepass", "dev_redis_pwd_2026",
              "--appendonly", "yes",
              "--appendfsync", "everysec",
              "--maxmemory", "512mb",
              "--maxmemory-policy", "allkeys-lru"]
    volumes:
      - redis_data:/data

volumes:
  redis_data:

启动与验证：

# 在 docker-compose.yml 所在目录
docker compose up -d

# 进入容器用 redis-cli 验证（密码见上方 requirepass）
docker exec -it redis810 redis-cli -a dev_redis_pwd_2026

# 容器内逐条敲，预期结果写在注释里
127.0.0.1:6379> PING
PONG
127.0.0.1:6379> SET hello world
OK
127.0.0.1:6379> GET hello
"world"
127.0.0.1:6379> INFO server
# 看 redis_version:8.10.x、os、process_id 等，确认版本与运行状态
127.0.0.1:6379> EXIT

本机验证通过标志：PING 返回 PONG、INFO 里 redis_version:8.10 开头、persistence 段 aof_enabled:1。三条都对上，本机环境就绪。

2.2 路线 B：1GB Linux 服务器（不装 Docker）

以 Debian 13 / Ubuntu 26.04 为例。优先用发行版自带包（省心、自动 systemd）；想要精确 8.10.2 再走官方源码构建。

B1 · 发行版包（首选）

sudo apt update
sudo apt install -y redis-server
sudo systemctl enable --now redis-server
redis-cli -a 本机密码 PING      # 返回 PONG 即活

Debian 13 / Ubuntu 26.04 仓库自带的是 Valkey 9 或 Redis 8.x 系列，命令与 go-redis 完全兼容；版本号若不是 8.10.2 属正常——服务器跑缓存，补丁级差异无关紧要。若必须锁 8.10.2，走 B2。

B2 · 官方源码构建（要精确版本时）

sudo apt install -y build-essential tcl
cd /tmp
curl -LO https://download.redis.io/releases/redis-8.10.2.tar.gz
tar xzf redis-8.10.2.tar.gz
cd redis-8.10.2
make -j$(nproc) BUILD_TLS=yes
sudo make install
# 把 utils/redis_init_script 改成 systemd unit（模板见模块 20），
# 配置文件放 /etc/redis/redis.conf，数据目录 /var/lib/redis

1GB 服务器铁律：不装 Docker（versions.md：Ubuntu Server 官方要求 1.5GB 内存起）。Redis 本体驻留内存控制在 256MB 内，剩下给 Go 单二进制 + SQLite + 系统。调优参数见 §9.2。

2.3 本机无 Docker 时的退路

Windows 本机也可以直接装 Redis：用 WSL2 里 apt install redis-server，或下载 Memurai/Redis Windows 移植版。但本教程默认 Docker 路线——删容器即清理，不污染系统。

只讲"写代码时必须知道"的部分。教科书里那些持久化引擎细节、集群 gossip 协议，小项目用不到，先不展开。

3.1 五种核心数据结构 + Streams + Pub/Sub

3.2 过期时间与键空间

• 所有缓存 key 都要设 TTL，这是纪律。不设 TTL 的 key 叫"永久 key"，会让 maxmemory 淘汰策略束手无策、内存只涨不跌。
• 命名规范：用冒号分层，全小写，业务前缀开头。例：ws:task:list:uid:1001（workbench:模块:实体:维度:值）、ws:sess:token:<jwtid>。一眼能看出这个 key 属于谁、能删不能删。
• 别用 KEYS * 线上扫键——它是 O(n) 全库阻塞。批量找键用 SCAN（增量、不阻塞，见 §4）。

3.3 持久化：RDB 与 AOF 怎么选（选型表）

一句话选型：只当缓存 → 关 AOF 也行（RDB 定时快照兜底）；当队列 → AOF everysec 必开。工作台两者都有，统一开 AOF everysec，别纠结。

3.4 内存淘汰策略（maxmemory-policy 六种，一句话记住）

你的 1GB 服务器统一用 allkeys-lru（§9.2）。它的含义是：内存到顶了，Redis 自己挑最冷的 key 踢掉，业务无感。

3.5 单线程模型与"为什么快"（面试+排障都用得上）

• 命令执行是单线程：所有命令在一个事件循环里串行执行——所以不需要锁、没有竞态，但也意味着"一个慢命令阻塞全部"（这就是为什么禁止 KEYS *、禁止大 key 操作）。
• 快的三个原因：纯内存操作；IO 多路复用（epoll）网络模型；数据结构经过高度优化（SDS、跳表、压缩列表）。
• 注意：8.0 版本在"异步删除/后台 IO"等旁路用了多线程，但你写业务命令时依然按单线程理解——别指望它并发执行你的 SET。

3.6 集群 / 哨兵（一句话，小项目别碰）

哨兵（Sentinel）= 主从自动故障转移；Cluster = 数据分片多主。你的工作台就一个单实例：挂了就挂了，缓存本来就能重建（从 SQLite/MySQL 回源），队列丢了就重放。什么时候上哨兵？等到你多实例、多可用区、缓存宕机直接丢营收那天再说——不是现在。

按数据结构组织。左边是命令，右边是你什么时候用。redis-cli 里照敲即可（带密码先 redis-cli -a 你的密码）。

4.1 String（最常用）

4.2 Hash

4.3 List / Set / ZSet

4.4 Streams（队列核心，§7 展开）

4.5 通用：过期 / 删除 / 扫描

从建 go.mod 到连接、五种结构操作、Pipeline/事务、发布订阅，全部给全可运行代码。客户端版本锁定 v9.23.0。

5.1 初始化项目与依赖

mkdir redisdemo && cd redisdemo
go mod init redisdemo
go get github.com/redis/go-redis/v9@v9.23.0
go get golang.org/x/sync/singleflight@latest   # §6 防击穿用，先装上

go.mod 关键行（锁好后提交入库）：

module redisdemo

go 1.27

require (
    github.com/redis/go-redis/v9 v9.23.0
    golang.org/x/sync v0.x.x   # singleflight，go get 后自动写入实际版本
)

5.2 连接（含密码、连接池参数）

// main.go —— 连接 + 最常用操作一览
package main

import (
    "context"
    "fmt"
    "time"

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

var rdb *redis.Client

func initRedis() {
    rdb = redis.NewClient(&redis.Options{
        Addr:     "localhost:6379", // 服务器换成 服务器IP:6379
        Password: "dev_redis_pwd_2026", // 对应 docker-compose 里的 requirepass
        DB:       0,              // 默认 0 号库；小项目别分库，用 key 前缀隔离
        PoolSize: 10,             // 连接池大小，1GB 小机器 10 足够
        MinIdleConns: 2,
        DialTimeout:  2 * time.Second,  // 连不上快速失败，别傻等
        ReadTimeout:  500 * time.Millisecond,
        WriteTimeout: 500 * time.Millisecond,
    })
}

func main() {
    initRedis()
    ctx := context.Background()

    // 连通性验证
    if err := rdb.Ping(ctx).Err(); err != nil {
        panic(err)
    }
    fmt.Println("connected to redis")

    demoString(ctx)
    demoHash(ctx)
    demoList(ctx)
    demoExpire(ctx)
}

func demoString(ctx context.Context) {
    // SET + 60 秒过期
    err := rdb.Set(ctx, "ws:hello", "world", 60*time.Second).Err()
    must(err)

    val, err := rdb.Get(ctx, "ws:hello").Result()
    must(err)
    fmt.Println("GET ws:hello =", val)

    // 计数器（原子）
    n, _ := rdb.Incr(ctx, "ws:cnt:visit").Result()
    fmt.Println("visit count =", n)
}

func demoHash(ctx context.Context) {
    must(rdb.HSet(ctx, "ws:user:1001", map[string]any{
        "name": "Tom", "email": "tom@example.com",
    }).Err())

    name, _ := rdb.HGet(ctx, "ws:user:1001", "name").Result()
    fmt.Println("user name =", name)

    all, _ := rdb.HGetAll(ctx, "ws:user:1001").Result()
    fmt.Println("user all =", all)
}

func demoList(ctx context.Context) {
    must(rdb.LPush(ctx, "ws:msgs", "m1", "m2").Err())
    // 阻塞弹出，3 秒超时（简单队列）
    res, err := rdb.BRPop(ctx, 3*time.Second, "ws:msgs").Result()
    if err == redis.Nil {
        fmt.Println("queue empty, timeout")
    } else {
        must(err)
        fmt.Println("pop =", res[1])
    }
}

func demoExpire(ctx context.Context) {
    must(rdb.Set(ctx, "ws:temp", "data", 0).Err()) // 0 = 先不设过期
    must(rdb.Expire(ctx, "ws:temp", 10*time.Second).Err())
    ttl, _ := rdb.TTL(ctx, "ws:temp").Result()
    fmt.Println("ttl =", ttl) // 约 10s
    must(rdb.Del(ctx, "ws:temp").Err())
}

func must(err error) {
    if err != nil && err != redis.Nil {
        panic(err)
    }
}

运行：go run .。看到 connected to redis 和几行输出，连接与基础操作就通了。注意 redis.Nil 是"key 不存在"的正常返回，不是错误——代码里要单独判（见 demoList）。

5.3 Pipeline 与事务（TxPipeline）

多个命令打包一次网络往返 = Pipeline。要原子性（中间不能被插命令）= TxPipeline（MULTI/EXEC）。

func demoPipeline(ctx context.Context) {
    pipe := rdb.Pipeline()
    pipe.Set(ctx, "k1", "v1", 10*time.Second)
    pipe.Set(ctx, "k2", "v2", 10*time.Second)
    g := pipe.Get(ctx, "k1")
    cmds, err := pipe.Exec(ctx) // 一次往返发完
    must(err)
    fmt.Println("pipeline cmds =", len(cmds), "k1 =", g.Val())
}

func demoTxPipeline(ctx context.Context) {
    // 原子：DECR 余额 + 加流水，要么都成要么都不成
    tx := rdb.TxPipeline()
    tx.DecrBy(ctx, "acct:balance", 100)
    tx.LPush(ctx, "acct:log", "pay 100")
    _, err := tx.Exec(ctx)
    must(err) // 若期间 key 被别人改导致 WATCH 冲突，会返回 EXECABORT
    fmt.Println("tx done")
}

5.4 发布订阅（Pub/Sub）简介

func demoPubSub(ctx context.Context) {
    sub := rdb.Subscribe(ctx, "ws:events")
    defer sub.Close()

    // 另一个 goroutine 发消息
    go func() {
        time.Sleep(200 * time.Millisecond)
        rdb.Publish(ctx, "ws:events", "task:created:1001")
    }()

    msg, err := sub.ReceiveMessage(ctx)
    must(err)
    fmt.Println("recv =", msg.Payload)
}

Pub/Sub vs Streams：Pub/Sub 是"广播即焚"——订阅者掉线期间的消息直接丢。要可靠、要回溯、要消费组，用 §7 的 Streams。你的工作台异步队列用 Streams，Pub/Sub 只做"前端实时通知"这类可丢场景。

这是 Redis 最值钱的部分。Cache Aside（旁路缓存）是默认模式，三防（击穿/穿透/雪崩）是它的三个加固点，限流是计数结构的招牌用法。全部给完整 Go 代码，可直接当 snippets 沉淀（08 §5.3）。

6.1 Cache Aside 标准写法

// cache.go —— 标准旁路缓存：读先查缓存，miss 再回源写回；写先更库再删缓存
package main

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

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

type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
}

// 模拟"数据库"（工作台里这里换成 GORM 查 SQLite/MySQL）
func queryUserFromDB(ctx context.Context, id int) (*User, error) {
    return &User{ID: id, Name: "Tom"}, nil
}

func GetUser(ctx context.Context, rdb *redis.Client, id int) (*User, error) {
    key := fmt.Sprintf("ws:user:%d", id)

    // 1) 先查缓存
    raw, err := rdb.Get(ctx, key).Result()
    if err == nil {
        var u User
        if json.Unmarshal([]byte(raw), &u) == nil {
            return &u, nil // 缓存命中
        }
    } else if err != redis.Nil {
        return nil, err // 真错误，别回源，直接返回（或降级）
    }

    // 2) miss → 回源查库
    u, err := queryUserFromDB(ctx, id)
    if err != nil {
        return nil, err
    }

    // 3) 写回缓存（带过期，§6.4 加随机防雪崩）
    data, _ := json.Marshal(u)
    ttl := 5*time.Minute + jitter()
    _ = rdb.Set(ctx, key, data, ttl).Err()
    return u, nil
}

// UpdateUser：写路径——先写库，再删缓存（不是改缓存！）
func UpdateUser(ctx context.Context, rdb *redis.Client, u *User) error {
    if err := saveUserToDB(ctx, u); err != nil {
        return err
    }
    // 删而不是改：避免并发下新旧值竞争；下次读自然重建
    return rdb.Del(ctx, fmt.Sprintf("ws:user:%d", u.ID)).Err()
}

func saveUserToDB(ctx context.Context, u *User) error { return nil }
func jitter() time.Duration { return time.Duration(rand.Intn(60)) * time.Second }

为什么写路径是"删缓存"而不是"改缓存"：并发写时，"先改缓存再更库"或"先更库再改缓存"都可能留下脏数据；而"更库 + 删缓存"把数据一致性收敛到下一次读重建，是业界默认。

6.2 防击穿（singleflight 合并并发回源）

击穿：某个热点 key 过期瞬间，成千上万请求同时 miss、同时回源打挂数据库。解法：同一时刻对同一个 key 只放一个请求去回源，其他等待结果。Go 标准做法 singleflight。

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

var sf singleflight.Group

func GetUserProtected(ctx context.Context, rdb *redis.Client, id int) (*User, error) {
    key := fmt.Sprintf("ws:user:%d", id)

    // 1) 先查缓存（同 6.1）
    if raw, err := rdb.Get(ctx, key).Result(); err == nil {
        var u User
        if json.Unmarshal([]byte(raw), &u) == nil {
            return &u, nil
        }
    }

    // 2) miss 时用 singleflight 合并：同一 key 同一时刻只跑一次 fn
    v, err, _ := sf.Do(key, func() (any, error) {
        // 进函数后再查一次缓存（双重检查）：可能已被同组先到的人填上
        if raw, err := rdb.Get(ctx, key).Result(); err == nil {
            var u User
            if json.Unmarshal([]byte(raw), &u) == nil {
                return &u, nil
            }
        }
        return queryUserFromDB(ctx, id) // 真正回源，全组共享这次结果
    })
    if err != nil {
        return nil, err
    }
    u := v.(*User)

    // 3) 写回缓存
    data, _ := json.Marshal(u)
    _ = rdb.Set(ctx, key, data, 5*time.Minute+jitter()).Err()
    return u, nil
}

没有 singleflight 的等价手工方案：用一个 per-key 的互斥锁（如 SETNX lock:key 1 EX 5），拿到锁的才回源，没拿到的自旋等一小会再读缓存。生产优先 singleflight，一行库搞定。

6.3 防穿透（空值缓存）

穿透：查根本不存在的数据（如 id=-1、恶意爬虫），缓存永远 miss，每次都打数据库。解法：把"查无此数据"也缓存成一个空标记，短 TTL。

var errNotFound = errors.New("not found")

func GetUserWithNullCache(ctx context.Context, rdb *redis.Client, id int) (*User, error) {
    key := fmt.Sprintf("ws:user:%d", id)

    if raw, err := rdb.Get(ctx, key).Result(); err == nil {
        if raw == "__NULL__" {
            return nil, errNotFound // 缓存里明确"不存在"，直接挡掉
        }
        var u User
        if json.Unmarshal([]byte(raw), &u) == nil {
            return &u, nil
        }
    }

    u, err := queryUserFromDB(ctx, id)
    if errors.Is(err, errNotFound) {
        // 空值也缓存，但 TTL 要短（30~60s），防真数据来了迟迟看不到
        _ = rdb.Set(ctx, key, "__NULL__", 30*time.Second).Err()
        return nil, errNotFound
    }
    if err != nil {
        return nil, err
    }

    data, _ := json.Marshal(u)
    _ = rdb.Set(ctx, key, data, 5*time.Minute+jitter()).Err()
    return u, nil
}

穿透 vs 击穿：击穿是"热点 key 过期"导致的并发回源（key 存在过）；穿透是"查不存在的 key"（缓存压根没有）。空值缓存 + 布隆过滤器（超大规模才上）是穿透的两道墙。

6.4 防雪崩（过期时间加随机）

雪崩：大批 key 同一时刻集体过期，瞬间全部回源。解法：在基础 TTL 上加一个随机抖动。6.1 里的 jitter() 就是干这个的：

import "math/rand"

// 每个 key 的过期 = 基础 5 分钟 + 0~60 秒随机，把"同时过期"摊平
func jitter() time.Duration {
    return time.Duration(rand.Intn(60)) * time.Second
}
// 写缓存统一用：rdb.Set(ctx, key, data, 5*time.Minute + jitter())

6.5 限流（滑动窗口，ZSet 实现）

需求：每个用户 60 秒内最多 100 次请求。用 ZSet 把每次请求的时间戳存进去，只留窗口内的。

import "time"

// AllowRate 滑动窗口限流：uid 在 window 内最多 max 次
func AllowRate(ctx context.Context, rdb *redis.Client, uid string, max int64, window time.Duration) (bool, error) {
    now := time.Now().UnixMilli()
    key := fmt.Sprintf("ws:rate:%s", uid)
    winStart := now - window.Milliseconds()

    // 用 Pipeline 原子化：① 清窗口外 ② 数窗口内 ③ 没超就加入本次
       pipe := rdb.TxPipeline()
    pipe.ZRemRangeByScore(ctx, key, "0", fmt.Sprintf("%d", winStart))
    cnt := pipe.ZCard(ctx, key)
    pipe.ZAdd(ctx, key, redis.Z{Score: float64(now), Member: now})
    pipe.Expire(ctx, key, window)
    if _, err := pipe.Exec(ctx); err != nil {
        return false, err
    }
    return cnt.Val() <= max, nil
}

更简单的固定窗口（够本教程用）：INCR 一个 key 并 EXPIRE 60，超过阈值就拒绝。滑动窗口更准，这里给的是后者；固定窗口代码三行，照名字就能写。

阶段三会用 Redis Streams 做消息队列（08 §2 阶段三路线）。这里给最小可用模型 + go-redis 代码：生产者 XADD，消费者组 XREADGROUP，处理完 XACK。

7.1 三个必须先懂的概念

• Stream：只追加的消息日志，每条消息有全局递增 ID（如 1700000000000-0）。
• 消费组（Consumer Group）：一组消费者分摊一个 Stream 的消息；同一条消息组内只投递给一个消费者。组内每个消费者有名字（c1/c2...）。
• PEL（待处理列表）：已投递但还没 XACK 的消息。消费者挂了，消息留在 PEL，重启后用 XACLAIM 抢回来重处理——这就是"不丢消息"的保证。

7.2 go-redis 生产者 + 消费者完整代码

// stream_demo.go —— 一个文件内同时演示生产与消费
package main

import (
    "context"
    "fmt"
    "log"
    "time"

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

const streamName = "ws:queue:mail"

func ensureGroup(ctx context.Context, rdb *redis.Client) {
    // 创建消费组；MKSTREAM 表示流不存在就建。BUSYGROUP 错误忽略（已存在）
    if err := rdb.XGroupCreateMkStream(ctx, streamName, "mailers", "$").Err(); err != nil &&
        err.Error() != "BUSYGROUP Consumer Group name already exists" {
        log.Fatal(err)
    }
}

func producer(ctx context.Context, rdb *redis.Client) {
    for i := 0; i < 3; i++ {
        id, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamName,
            Values: map[string]any{
                "to":      fmt.Sprintf("user%d@example.com", i),
                "subject": "welcome",
            },
        }).Result()
        if err != nil {
            log.Fatal(err)
        }
        fmt.Println("produced:", id)
    }
}

func consumer(ctx context.Context, rdb *redis.Client, name string) {
    for {
        // 阻塞 5 秒读新消息（> 表示从未投递过的）
        res, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    "mailers",
            Consumer: name,
            Streams:  []string{streamName, ">"},
            Count:    10,
            Block:    5 * time.Second,
        }).Result()
        if err != nil {
            if err == redis.Nil {
                continue // 阻塞超时，没消息，继续等
            }
            log.Println("read err:", err)
            time.Sleep(time.Second)
            continue
        }
        for _, stream := range res {
            for _, msg := range stream.Messages {
                fmt.Printf("[%s] got id=%v to=%v\n", name, msg.ID, msg.Values["to"])
                // ---- 这里做真实工作：发邮件 ----
                if err := sendMail(msg.Values); err != nil {
                    log.Println("process err, keep in PEL:", err)
                    continue // 不 ACK，稍后重投
                }
                // 成功了才 ACK
                if err := rdb.XAck(ctx, streamName, "mailers", msg.ID).Err(); err != nil {
                    log.Println("ack err:", err)
                }
            }
        }
    }
}

func sendMail(vals map[string]any) error {
    fmt.Println("  >> sending mail to", vals["to"])
    return nil
}

func runStreamDemo() {
    ctx := context.Background()
    rdb := newDemoClient()
    ensureGroup(ctx, rdb)

    go consumer(ctx, rdb, "c1")
    go consumer(ctx, rdb, "c2") // 两个消费者分摊，演示负载均衡
    time.Sleep(300 * time.Millisecond)
    producer(ctx, rdb)
    time.Sleep(2 * time.Second)
}

跑起来你会看到：3 条消息被 c1/c2 分摊消费（组内广播变单播），每条处理完 XACK。故意在 sendMail 里 return error，那条消息就不 ACK——重启消费者后用 XReadGroup 从 PEL 重读即可重投。这就是工作台"异步发邮件/异步生成报表"的骨架。

取舍：你的量级（个人工作台）用 Redis Streams 足够。将来日消息量到百万级、要跨机房、要严格顺序，再评估专门 MQ（如 Kafka）——本阶段不需要。

按 workbench-contract.md：工作台是 Go(Gin+GORM) + Vue3，端点表固定（/healthz、/api/v1/auth/login、/api/v1/tasks CRUD）。Redis 是加速层，接在现有端点内部，不新增任何 API 路径。下面三处接入位置，给片段。

8.1 接入位置一：会话 / token 缓存

现状（契约）：JWT 无状态，每次请求都验签。可把"已签发的活跃 token ID"缓存进 Redis，支持主动登出（拉黑）与快速失效。接入点：JWT 认证 middleware 内部，路由表不变。

// internal/middleware/auth.go 内：验签通过后多查一次黑名单缓存
func AuthMiddleware(rdb *redis.Client) gin.HandlerFunc {
    return func(c *gin.Context) {
        token := extractBearer(c)
        claims, err := jwt.Parse(token) // 沿用契约 jwt/v5
        if err != nil { c.AbortWithStatusJSON(401, ...); return }

        jti := claims.ID
        // 若 jti 在黑名单缓存里 = 已登出
        if n, _ := rdb.Exists(c, "ws:jwt:bl:"+jti).Val(); n > 0 {
            c.AbortWithStatusJSON(401, gin.H{"code": 401, "msg": "logged out"})
            return
        }
        c.Set("userID", claims.Subject)
        c.Next()
    }
}
// 登出（复用现有 logout 逻辑）：rdb.Set(ctx, "ws:jwt:bl:"+jti, "1", 剩余有效期)
// 注意：这是在现有 handler 里加一行，不是新端点。

8.2 接入位置二：任务列表缓存

现状（契约）：GET /api/v1/tasks 每次查 GORM。任务列表读多写少，最适合 Cache Aside（§6.1）。接入点：task handler 内部，响应结构不变。

// internal/handler/task.go 内：ListTasks 加缓存旁路
func (h *TaskHandler) ListTasks(c *gin.Context) {
    uid := c.GetUint("userID")
    key := fmt.Sprintf("ws:task:list:uid:%d", uid)

    // 1) 先读缓存
    if raw, err := h.rdb.Get(c, key).Result(); err == nil {
        var tasks []Task
        json.Unmarshal([]byte(raw), &tasks)
        c.JSON(200, gin.H{"code": 0, "data": tasks})
        return
    }

    // 2) miss 回源 GORM（契约里的 db）
    var tasks []Task
    h.db.Where("uid = ?", uid).Find(&tasks)

    // 3) 写回缓存，TTL 5 分钟 + 随机抖动
    if data, err := json.Marshal(tasks); err == nil {
        h.rdb.Set(c, key, data, 5*time.Minute+jitter())
    }
    c.JSON(200, gin.H{"code": 0, "data": tasks})
}

// 写路径（Create/Update/Delete task）末尾统一加一行：删缓存
// h.rdb.Del(c, fmt.Sprintf("ws:task:list:uid:%d", uid))

契约遵守：对外仍是 GET /api/v1/tasks，请求/响应字段（code/data/msg、Task 的 title/note/done）一个没变。Redis 只在 handler 内部多了一层旁路。

8.3 接入位置三：Stream 做异步队列

把"新建任务后发通知邮件"这类副作用从请求链路里摘出去：handler 只往 Stream 投一条消息就返回，后台消费者（独立 goroutine 或独立二进制）慢慢发邮件。接入点：POST /api/v1/tasks 内部投递，不新增端点。

// CreateTask 成功后：不直接发邮件，改投队列
func (h *TaskHandler) CreateTask(c *gin.Context) {
    var in struct{ Title, Note string; Done bool }
    c.BindJSON(&in)
    task := saveTask(in) // GORM，契约原有逻辑

    // 异步通知：投 Stream，立即返回，不等邮件发完
    h.rdb.XAdd(c, &redis.XAddArgs{
        Stream: "ws:queue:notify",
        Values: map[string]any{"task_id": task.ID, "title": task.Title},
    })

    c.JSON(200, gin.H{"code": 0, "data": task})
}
// 消费者进程（§7.2 的 consumer）在服务器后台常驻，消费 ws:queue:notify 发邮件。
// 主请求延迟从"发邮件 500ms"降到"XAdd 0.5ms"。

8.4 配置项（走契约的环境变量）

沿用契约：配置走环境变量（godotenv 加载）。在 internal/config/config.go 加两个字段，不引入新配置体系：

REDIS_ADDR=localhost:6379
REDIS_PASSWORD=dev_redis_pwd_2026
# 服务器上：REDIS_ADDR=127.0.0.1:6379（Redis 只绑本机，不暴露公网）

Redis 是加速层，备份的优先级低于 SQLite/MySQL 主库。但 Stream 队列里可能有未消费任务，所以给一份最小可用方案：RDB 定时快照 + 必要时 AOF。

9.1 RDB 快照备份与恢复

BGSAVE vs SAVE

• SAVE：主线程同步 dump，阻塞整个 Redis，生产禁止。
• BGSAVE：fork 子进程后台 dump，主线程继续服务，备份用它。

RDB 文件位置（默认）：dir 配置指向的目录下的 dump.rdb（Docker 卷里即 /data/dump.rdb）。

cron 定时备份脚本（照抄，放到服务器 /usr/local/bin/redis-backup.sh）

#!/bin/bash
# /usr/local/bin/redis-backup.sh —— 每天凌晨备份 RDB，留 7 天
set -euo pipefail

REDIS_CLI="redis-cli -a dev_redis_pwd_2026"
BACKUP_DIR=/var/backups/redis
mkdir -p "$BACKUP_DIR"
STAMP=$(date +%Y%m%d-%H%M)

# 1) 触发后台快照
$REDIS_CLI BGSAVE

# 2) 等快照完成（通过 LASTSAVE 时间戳变化判断）
LAST_BEFORE=$($REDIS_CLI LASTSAVE)
while true; do
    sleep 2
    LAST_NOW=$($REDIS_CLI LASTSAVE)
    [ "$LAST_NOW" != "$LAST_BEFORE" ] && break
done

# 3) 拷贝快照出来（RDB 在 /var/lib/redis/dump.rdb，按你的 dir 改）
cp /var/lib/redis/dump.rdb "$BACKUP_DIR/dump-$STAMP.rdb"

# 4) 只留最近 7 份
ls -1t "$BACKUP_DIR"/dump-*.rdb | tail -n +8 | xargs -r rm -f

echo "backup done: dump-$STAMP.rdb"

# 加执行权限 + 挂 cron（每天 3:17 跑，避开整点）
sudo chmod +x /usr/local/bin/redis-backup.sh
sudo crontab -e
# 写入这一行：
17 3 * * * /usr/local/bin/redis-backup.sh >> /var/log/redis-backup.log 2>&1

恢复流程

1. 停 Redis：sudo systemctl stop redis-server。
2. 把要恢复的快照拷回数据目录：cp /var/backups/redis/dump-YYYYMMDD-HHMM.rdb /var/lib/redis/dump.rdb，chown redis:redis。
3. 启动：sudo systemctl start redis-server，redis-cli INFO keyspace 确认 key 回来了。

9.2 AOF 开启、rewrite 与恢复

# redis.conf 关键行（队列场景开 everysec）
appendonly yes
appendfilename "appendonly.aof"
appendfsync everysec        # 每秒刷盘，宕机最多丢 1 秒
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb  # AOF 涨到 64MB 且翻倍就自动 rewrite 瘦身

手动 rewrite：redis-cli BGREWRITEAOF（后台重写，压缩历史命令）。AOF 损坏时：用 redis-check-aof --fix appendonly.aof 修复截断尾部再启动。恢复顺序：有 AOF 先加载 AOF，没有才加载 RDB。

9.3 1GB 低配服务器调优（照抄 redis.conf 片段）

# ===== 1GB 服务器 Redis 调优基线 =====
bind 127.0.0.1            # 只绑本机！公网暴露 = 被扫到挖矿钱包
protected-mode yes
port 6379
requirepass dev_redis_pwd_2026   # 换强密码

maxmemory 256mb           # Redis 最多吃 256MB，剩下留给 Go+SQLite+系统
maxmemory-policy allkeys-lru   # 满了自动踢最冷的 key（§3.4）

# 持久化按角色选：纯缓存可 no；有队列必须 everysec（下面是队列场景）
appendonly yes
appendfsync everysec

# 危险命令改名/禁用（一句话：别让 KEYS/CONFIG/FLUSH 在公网能跑）
rename-command FLUSHDB ""
rename-command FLUSHALL ""
rename-command CONFIG "RedisCfg_2026"

• swap 兜底：1GB 机器给系统开 1~2GB swap（swapfile），Redis 不是被 OOM killer 一刀杀的首选，但兜底能避免整机僵死。详见模块 20。
• AOF 策略按场景：纯缓存节点 appendonly no（重启从库重建即可）；队列节点 everysec。你的工作台单实例两者都有，统一 everysec。

9.4 什么时候用 Redis（引用 08 §8.1）

照报错原文查"原因 → 解决"。这是闭关期最常翻的一节。

11.1 本模块验收（打勾才算过）

11.2 与其他模块的关系

• 环境前置（Go 1.27.2、Docker Desktop 4.94）见 09 环境与版本清单；本模块不重复装环境。
• 工作台结构/端点/模型以 11 工作台实战与 workbench-contract 为准，本模块不新增端点。
• systemd、swap、Nginx/Caddy、监控的生产化细节见 20 部署运维专题。
• 业务主库仍是 SQLite（19）/MySQL（16）/PG（17），Redis 只做加速层。

来源表

（编者整理：Cache Aside、三防、限流、Stream 队列的 Go 代码为本教程基于 go-redis v9 API 的演示实现，无外部独立来源；错误表为常见运维场景归纳。）