--- tags: [Redis, 缓存, 高级特性] create time: 2026-05-15 18:15 --- # Redis 高级特性 ## 概述 掌握事务、Lua 脚本和 Pipeline,是 Redis 从「基础缓存」迈向「生产级系统」的关键。本节逐一剖析其原理、场景和陷阱。 ## 一、事务(MULTI / EXEC) ### 基本用法 ```bash MULTI # 开启事务(不阻塞,之后命令只是排队) INCR counter # QUEUED INCR other_counter # QUEUED EXEC # 一次性执行所有排队的命令 ``` ```go pipe := rdb.TxPipeline() // TxPipeline ≠ MULTI,见下方说明 pipe.Incr(ctx, "counter") pipe.Set(ctx, "key", "value", 0) results, _ := pipe.Exec(ctx) ``` > [!TIP] Go 的 TxPipeline vs MULTI > `rdb.TxPipeline()` 在 go-redis 中**并不发送 MULTI/EXEC**——它只是普通 pipeline。这是为了兼容 Cluster 环境(集群不支持事务)。两者区别: | 维度 | MULTI/EXEC | Pipeline | |------|-----------|----------| | 原子性 | ✅ 整体执行,中途不插队 | ❌ 命令独立发送执行 | | 错误处理 | 编译期错误全部拒绝;运行时错误逐个记录 | 每条返回各自结果 | | Cluster 支持 | ❌ | ✅(但需同 key tag) | | 性能 | 同 Pipeline | 批量发送减少 RTT | ### 事务的错误处理 ```bash MULTI SET k v NX GET KEYY xxx # ⚠️ 语法错误! EXEC ``` | 错误类型 | 发生时 | 行为 | |---------|--------|------| | **编译时错误**(语法、参数) | EXEC 之前 | EXEC 拒绝执行整个事务 | | **运行时错误**(类型不匹配) | EXEC 期间 | 当前命令报错,其余继续执行 | ```bash MULTI SET k 123 # OK, queued INCR k # OK, queued (先 SET 再 INCR 是合法的) EXEC # → [OK, 124] — 两个都成功了 ``` > [!WARNING] Redis 事务 ≠ 数据库事务 > - 没有 ROLLBACK / ACID 保证 > - 不能捕获异常后回滚 > - 如果需要强一致性,请用 **Lua 脚本** ## 二、Lua 脚本 ### 为什么需要 Lua? 核心目的:**原子性操作多个命令**。MULTI 只能保证"不插队",Lua 能保证"逻辑原子性"(读-判断-写一气呵成)。 ```bash EVAL "return redis.call('GET', KEYS[1])" 1 mykey # ↑ script ↑ commands ↑ key count ↑ keys... ``` ### 经典场景 1:分布式锁 ```lua -- acquire_lock.lua -- KEYS[1] = lock key -- ARGV[1] = expire seconds (TTL) -- ARGV[2] = unique token (client ID) if redis.call('set', KEYS[1], ARGV[2], 'NX', 'EX', ARGV[1]) == 'OK' then return 1 -- 成功获取 end if redis.call('get', KEYS[1]) == ARGV[2] then -- 已持有且未过期,原子性续期 redis.call('expire', KEYS[1], ARGV[1]) return 1 end return 0 -- 被其他客户端持有 ``` ```go // Go 调用 script := redis.NewScript(` if redis.call("get", KEYS[1]) == false then -- setex(key, seconds, value): ARGV[1]=过期秒数, ARGV[2]=唯一令牌 redis.call("setex", KEYS[1], ARGV[1], ARGV[2]) return 1 end return 0 `) ok, err := script.Run(ctx, c, []string{"lock:order:" + orderId}, lockTTL, clientToken).Int() if ok == 1 { defer unlock(clientToken) doBiz() // 临界区代码 } ``` ### 经典场景 2:限流器(令牌桶简化版) ```lua -- rate_limit.lua local limit = tonumber(ARGV[1]) -- 每分钟上限 local window = tonumber(ARGV[2]) -- 窗口秒数 local current = tonumber(redis.call('get', KEYS[1]) or '0') if current >= limit then return 0 -- 超限,拒绝 end local pipe = redis.pipeline() pipe.incr(KEYS[1]) pipe.expire(KEYS[1], window) pipe.exec() return 1 -- 通过 ``` ### 经典场景 3:订单状态机 ```lua -- order_status.lua -- KEYS[1] = order key -- ARGV[1] = from_status -- ARGV[2] = to_status local current = redis.call('hget', KEYS[1], 'status') if current ~= ARGV[1] then return 0 -- 状态不匹配(第一个返回值:失败) end redis.call('hset', KEYS[1], 'status', ARGV[2]) return 1 -- 成功(第一个返回值),后续通过 GET 读取新状态 ``` ### EVAL vs EVALSHA ```bash EVAL "script..." numkeys args... # 每次传完整脚本 EVALSHA numkeys args... # 只传 SHA1,省带宽 ``` > [!TIP] 使用 EVALSHA 的最佳实践 > ```go > script := redis.NewScript(...) > // 首次调用失败可能是 NOSCRIPT(缓存未命中),会自动重试 EVAL > result := script.Run(ctx, ...).Int() > ``` > go-redis 内部已处理了 NOSCRIPT → EVAL fallback 的逻辑。 ### Lua 沙箱限制 ```lua -- 不可用:非确定性函数(传统 Redis) math.random() -- ❌ 无法预测结果 os.date() -- ❌ 依赖系统时间 time.time() -- ❌ Lua 标准库不存在 -- ✅ 推荐做法:在客户端生成,通过 ARGV 传入 -- local token = client.generateUUID() -- redis.call('set', KEYS[1], ARGV[1]) -- 带 token 写入 > [!NOTE] Redis 7.2+ 更新 > Redis 7.2 起开放了部分确定性数学函数:`math.random`(需自行 set seed)、`math.max`、`math.min`、`math.log`。但不建议在状态机类脚本中使用随机性。 ``` > [!QUESTION] 为什么 Lua 脚本不能有随机函数? > Lua 脚本在主线程中同步执行,如果引入随机性或时间依赖,会导致两个严重问题: > 1. **主从不一致**:主节点执行返回 42,从节点重放却得到 87,数据分裂。 > 2. **AOF 重放不可靠**:重启后从 appendonly.aof 重放脚本,结果与上次不同,状态机紊乱。 > 因此 Redis 在 Lua 环境中屏蔽了所有非确定性 API,保证「同样的输入一定得到同样的输出」。 ## 三、Pipeline ### 为什么需要 Pipeline? Redis 协议本质是 request/response,每个命令一次网络往返(RTT)。假设客户端到服务器 RTT 为 1ms,执行 1000 条命令: | 方式 | RTT 次数 | 总耗时估算 | |------|---------|-----------| | 单次发送 | 1000 | ~1s | | Pipeline (batch=200) | 5 次 | ~5ms | ### Pipeline vs TxPipeline 在 go-redis 中,有两种管道模式可用: | 方式 | 适用场景 | 注意事项 | |------|---------|---------| | `Pipeline` | 独立命令批量 | 不保证原子性 | | `TxPipeline` (go-redis) | 需要原子的批量 | 对应 MULTI/EXEC | ```go // 方式一:普通 Pipeline(性能优化首选) pipe := rdb.Pipeline() pipe.Set(ctx, "user:1:name", "alice", 0) pipe.Set(ctx, "user:1:age", "25", 0) pipe.HSet(ctx, "user:1:profile", "email", "a@x.com") results, _ := pipe.Exec(ctx) // 返回 []redis.Result _ = results // 实际使用中检查 error // 方式二:Watch CAS(乐观锁 —— 读-改-写原子的核心模式) var balance int64 err := rdb.Watch(ctx, func(tx *redis.Tx) error { acc, err := tx.Get(ctx, "account:alice").Result() if err != nil { return err } balance, _ = strconv.ParseInt(acc, 10, 64) return nil // 释放 Watch 锁 }, "account:alice") if err != nil { /* NOSCRIPT / watch abort */ return } if balance < 100 { log.Fatal("余额不足") } _, err = rdb.ExecFunc(ctx, func(tx *redis.Tx) error { tx.DecrBy(ctx, "account:alice", 100) tx.IncrBy(ctx, "account:bob", 100) return nil }).Result() ``` > [!WARNING] Pipeline 的注意事项 > - 单个 pipeline 内的命令可能跨多个节点(Cluster 模式下会分发执行) > - 避免在 pipeline 中混用 WATCH/MULTI(Cluster 不支持) > - 批量越大越好?**不是**——单 batch 建议 50~200 条,过大反而降低吞吐且增加超时风险 ### Pipeline 实战——数据迁移 ```go func BatchMigrate(ctx context.Context, rdb *redis.Client, srcKeys []string) error { const batchSize = 100 pipe := rdb.Pipeline() for i := 0; i < len(srcKeys); i += batchSize { batch := srcKeys[i : min(i+batchSize, len(srcKeys))] for _, key := range batch { val, _ := rdb.Get(ctx, key).Result() newKey := strings.TrimPrefix(key, "old:") pipe.Set(ctx, newKey, val, 0) } _, err := pipe.Exec(ctx) if err != nil { return err } pipe.Close() pipe = rdb.Pipeline() // 重建 pipeline 避免内存泄漏 } return nil } ``` ## 四、Pub/Sub —— 发布订阅 ### 基础用法 ```bash # 消费者 A SUBSCRIBE channel:newposts # 消费者 B(也可以 subscribe 同一频道) SUBSCRIBE channel:newposts # 发布者 PUBLISH channel:newposts '{"id":42,"title":"Redis Tips"}' → 消费者 A 和 B 同时收到消息 ``` ```go // go-redis 订阅者 pubsub := rdb.Subscribe(ctx, "channel:newposts") defer pubsub.Close() ch := pubsub.Channel() // chan redis.Message,阻塞接收 for msg := range ch { fmt.Printf("收到: %s\n", msg.Payload) } // go-redis 发布者 — 无返回值,发出去就不管了 rdb.Publish(ctx, "channel:newposts", `{"id":42,"title":"Redis Tips"}`) ``` > [!WARNING] Pub/Sub 是无状态的 > - **连接级绑定**:每个 SUBSCRIBE 会独占一个 Redis 连接——线上建议限流并发 subscriber 数量 > - **消息不持久化**:订阅者下线就丢失,重连不会收到离线期间的消息 > - **适合场景**:WebSocket 广播、实时通知、配置变更推送 ### Stream —— 可靠的替代方案 Stream 是 Redis 5.0+ 引入的消息队列数据结构,弥补了 Pub/Sub 的可靠性缺陷。核心设计围绕 **Consumer Group(消费者组)**: ```bash # 生产者写入 XADD mystream * user_id 123 action login ip 192.168.1.1 # 创建消费者组 XGROUP CREATE mystream group1 $ # $ = 从最新消息开始消费 # XGROUP CREATE mystream group1 0 # 0 = 从头开始(生产环境慎用) # 消费者读取(阻塞模式,无消息时等待指定毫秒数) XREADGROUP GROUP group1 consumer1 COUNT 1 STREAMS mystream > # 确认消费完成(必须调用,否则该条消息会一直留在 pending 列表) XACK mystream group1 ``` > [!TIP] `$` vs `>` —— 初学者最容易踩的坑 > | 标识符 | 含义 | 使用场景 | > |--------|------|---------| > | `$` | "最后一条已写入的消息" | 新 consumer 加入时,只接收**将来**的消息 | > | `0` | "历史第一条" | 数据迁移/重放(谨慎用于大 stream) | > | `>` | "我只取未分配给我的新消息" | **正常消费循环中始终用 `>`**,不取到 pending 消息 | > [!QUESTION] 如果消费者挂了怎么办? > 好消息会被移到 **pending 列表**。用 `XPENDING mystream group1` 查看,用 `XCLAIM` 将 pending 消息转移给其他 consumer 重新处理。这就是 Stream 比 Pub/Sub 可靠的地方:**有状态、可追责**。 ```mermaid flowchart LR P["Producer"] -->|"XADD"| S["Stream (持久化)"] S -->|"读取并标记 pending"| G1["Consumer Group 1
consumer-a"] S -->|"读取并标记 pending"| G2["Consumer Group 2
consumer-b"] G1 -->|"XACK"| S style S fill:#dfd,stroke:#090 style G1 fill:#ddf,stroke:#669 style G2 fill:#ddf,stroke:#669 ``` > [!TIP] Pub/Sub vs Stream 选型 > | 场景 | 推荐 | > |------|------| > | WebSocket 实时推送 | Pub/Sub | > | 任务队列、订单事件 | Stream | > | 海量消息、高吞吐 | Kafka/RabbitMQ | ## 五、Keyspace Notifications —— 键事件监听 ### 原理与配置 Redis 会在每个 key 发生变更时发送通知,订阅者通过 Pub/Sub 协议接收。开启方式: ```conf # redis.conf(或运行时 CONFIG SET) notify-keyspace-events KEA # K = Keyspace events → 发布到 __keyspace@__ prefix # E = Keyevent events → 发布到 __keyevent@__ prefix # A = All events 简写 → 覆盖 g/l/s/h/z/x/e(不包含 K/E/d/t/n) # g = DEL/EXPIRE/RENAMExxx 等通用命令 # l = list, s = set, h = hash, z = sorted-set # x = expired, e = evicted (maxmemory 淘汰) ``` ```bash # 订阅过期事件(Keyevent 模式:更精确) PSUBSCRIBE __keyevent@0__:expired # 另一个连接设置过期 key EXPIRE test:key 2 → 收到: __keyevent@0__:expired "test:key" ``` ```go // go-redis 监听器 psub := rdb.PSubscribe(ctx, "__keyevent@0__:expired") defer psub.Close() ch := psub.Channel() for msg := range ch { fmt.Printf("Key %s expired\n", msg.Payload) // 可以触发清理逻辑、计数器归零等 } ``` > [!WARNING] 生产环境注意事项 > - **性能开销**:每条写命令都会生成通知,高 QPS + 全量监听 (`KAE`) 会显著增加 CPU/内存 > - **异步不保证送达**:本质是 Pub/Sub,网络抖动时通知可能丢失 > - **从节点不会推送**:默认只在写入节点生效,哨兵/集群切换后需重新订阅 > - **建议**:只订阅你需要的事件类型(如 `KElx`),而非 `KExa` 通配一切 ## 关联笔记 - [[hhs/Redis/06-主从与哨兵]] — Sentinel 故障检测(也是基于 PING/PONG 心跳) - [[hhs/Redis/07-集群方案]] — Cluster 下 Pipeline/Lua 的限制 - [[hhs/Redis/README]] — 知识索引总览