407 lines
13 KiB
Markdown
407 lines
13 KiB
Markdown
|
---
|
|||
|
|
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 <sha1> 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 <entry-id>
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
> [!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<br/>consumer-a"]
|
|||
|
|
S -->|"读取并标记 pending"| G2["Consumer Group 2<br/>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@<db>__ prefix
|
|||
|
|
# E = Keyevent events → 发布到 __keyevent@<db>__ 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]] — 知识索引总览
|