678 lines
28 KiB
Markdown
678 lines
28 KiB
Markdown
---
|
||
tags: [Redis, 缓存, 高级特性]
|
||
create time: 2026-05-15 18:15
|
||
---
|
||
|
||
# Redis 高级特性
|
||
|
||
## 概述
|
||
|
||
当你用 Redis 做的不只是「存一个值、取一个值」时——比如转账时要同时扣 A 加 B、限流时要「先判断再计数」——单条命令就不够用了。你需要**把多条命令打包**，确保它们要么一起成功，要么逻辑上不出错。
|
||
|
||
本节围绕这个核心问题，逐一介绍三种「打包」工具：
|
||
- **事务（MULTI/EXEC）**：保证命令按顺序执行、不被插队，但不能根据前一条结果决定下一步
|
||
- **Lua 脚本**：真正的「逻辑原子性」——读、判断、写一气呵成
|
||
- **Pipeline**：纯性能优化——批量发送命令，减少网络往返
|
||
|
||
此外还会介绍 Pub/Sub 和 Stream 这两种消息传递机制，以及 Keyspace Notifications 键事件监听。
|
||
|
||
## 一、事务（MULTI / EXEC）
|
||
|
||
### 先想一个问题
|
||
|
||
假设你要给两个计数器各加 1。如果直接逐条发送 `INCR counter`、`INCR other_counter`，在高并发下可能出现这种情况：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant A as "Client A"
|
||
participant B as "Client B"
|
||
participant S as "Redis"
|
||
A->>S: "INCR counter 得到 1"
|
||
B->>S: "INCR counter 得到 2 (插队了!)"
|
||
B->>S: "INCR other_counter 得到 1"
|
||
A->>S: "INCR other_counter 得到 2"
|
||
```
|
||
|
||
两个命令之间被别的客户端**插队**了。虽然最终结果可能没问题，但在更复杂的场景（比如「读余额→判断→扣款」）中，插队会导致严重 bug。
|
||
|
||
**事务就是用来解决这个问题的：把一组命令打包，保证它们连续执行、中间不被打断。**
|
||
|
||
### 基本用法
|
||
|
||
```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 的 Pipeline vs TxPipeline —— 简单理解
|
||
> - `rdb.Pipeline()` = 批量发送，**不加锁**，谁都能插队 → 纯粹为了快
|
||
> - `rdb.TxPipeline()` = 批量发送，**前后包上 MULTI/EXEC** → 保证连续执行
|
||
>
|
||
> 在 Cluster 模式下，go-redis 会按 slot 分组，对每个节点单独发 MULTI/EXEC（所以同一 pipeline 内的 key 必须在相同 hash slot，否则事务会拆开）。
|
||
|
||
| 维度 | MULTI/EXEC（TxPipeline） | Pipeline |
|
||
|------|-----------|----------|
|
||
| 原子性 | ✅ 连续执行，中途不插队 | ❌ 命令独立执行，可能被插队 |
|
||
| 错误处理 | 语法错 → 整批拒绝；运行时错 → 只影响当前命令 | 每条命令各自报错 |
|
||
| Cluster | ❌ 不支持跨 slot 事务 | ✅ 自动按 slot 分发 |
|
||
| 适用场景 | 需要保证执行顺序 | 只求快，不关心顺序 |
|
||
|
||
### 事务的错误处理
|
||
|
||
```bash
|
||
MULTI
|
||
SET k v NX
|
||
GET KEYY xxx # ⚠️ 语法错误！
|
||
EXEC
|
||
```
|
||
|
||
| 错误类型 | 发生时 | 行为 |
|
||
|---------|--------|------|
|
||
| **编译时错误**（语法、参数） | EXEC 之前 | EXEC 拒绝执行整个事务 |
|
||
| **运行时错误**（类型不匹配） | EXEC 期间 | 当前命令报错，其余继续执行 |
|
||
|
||
```bash
|
||
MULTI
|
||
SET k "hello" # OK, queued
|
||
INCR k # OK, queued（语法合法，EXEC 前不会检查类型）
|
||
EXEC
|
||
# → [OK, ERR value is not an integer...] — SET 成功，INCR 因类型不匹配报错
|
||
```
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant C as "Client"
|
||
participant S as "Redis Server"
|
||
C->>S: MULTI
|
||
S-->>C: OK
|
||
C->>S: SET k v
|
||
S-->>C: QUEUED
|
||
C->>S: INCR counter
|
||
S-->>C: QUEUED
|
||
C->>S: EXEC
|
||
Note over S: 依次执行所有排队命令
|
||
S-->>C: [OK, 1]
|
||
```
|
||
|
||
> [!QUESTION] 既然 MULTI 能批量执行，为什么还需要 Pipeline？
|
||
> 关键区别在于**排队阶段是否立即返回结果**。MULTI 的 QUEUED 只是"记下来"，命令要等 EXEC 才真正执行——这意味着你**无法根据前一条的结果决定后续命令**。而 Lua 脚本能做"读-判断-写"，Pipeline 则纯粹为减少 RTT。三者各有分工，切勿混淆。
|
||
|
||
> [!WARNING] Redis 事务 ≠ 数据库事务
|
||
> - 没有 ROLLBACK / ACID 保证
|
||
> - 不能捕获异常后回滚
|
||
> - 如果需要强一致性，请用 **Lua 脚本**
|
||
|
||
## 二、Lua 脚本
|
||
|
||
### 为什么需要 Lua？——MULTI 的局限
|
||
|
||
MULTI 能保证命令「不插队」，但它有一个致命缺陷：**你无法根据前一条命令的结果决定下一步操作**。
|
||
|
||
比如你想实现「如果余额 ≥ 100 就扣款」：
|
||
1. `GET balance` → 拿到 50
|
||
2. （根据结果判断：50 < 100，不应该扣款）
|
||
3. 但在 MULTI 里，`DECRBY balance 100` 已经在排队了——**你无法在 EXEC 之前取消它**
|
||
|
||
> [!QUESTION] 怎么办？
|
||
> 答案是：把「读 + 判断 + 写」放到**一个脚本**里交给 Redis 执行。这就是 Lua 脚本的核心价值——**逻辑原子性**：整个脚本在 Redis 主线程上一口气跑完，中间不会被任何其他命令插入。
|
||
|
||
简单对比一下两者的原子性：
|
||
|
||
| | MULTI/EXEC | 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 调用——与上方 Lua 脚本等价的分布式锁实现
|
||
script := redis.NewScript(`
|
||
-- 尝试获取锁：SET NX EX（只有 key 不存在时才设置，原子操作）
|
||
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 -- 被其他客户端持有
|
||
`)
|
||
|
||
// 解锁脚本：只有持有者才能释放，防止误删别人的锁
|
||
unlockScript := redis.NewScript(`
|
||
if redis.call("get", KEYS[1]) == ARGV[1] then
|
||
return redis.call("del", KEYS[1])
|
||
end
|
||
return 0 -- 锁不属于自己，拒绝释放
|
||
`)
|
||
|
||
// KEYS[1] = lock key, ARGV[1] = TTL 秒数, ARGV[2] = 唯一令牌（如 UUID）
|
||
ok, err := script.Run(ctx, rdb, []string{"lock:order:" + orderId},
|
||
lockTTL, clientToken).Int()
|
||
if ok == 1 {
|
||
defer func() {
|
||
// 解锁：只有持有者才能释放（Lua 保证原子性）
|
||
unlockScript.Run(ctx, rdb, []string{"lock:order:" + orderId}, clientToken)
|
||
}()
|
||
doBiz() // 临界区代码
|
||
}
|
||
```
|
||
|
||
> [!tip] 分布式锁深入阅读
|
||
> 本节仅展示 Lua 实现分布式锁的核心代码。关于安全解锁、锁续期（Watchdog）、Redlock 算法、可重入锁等进阶内容，详见 [[hhs/Redis/09-高级特性/分布式锁]]。
|
||
|
||
### 经典场景 2：限流器（固定窗口计数器）
|
||
|
||
```lua
|
||
-- rate_limit.lua（固定窗口计数器 —— Lua 脚本本身就是原子的，无需 pipeline）
|
||
-- KEYS[1] = 计数器 key
|
||
-- ARGV[1] = 每窗口上限
|
||
-- ARGV[2] = 窗口秒数
|
||
|
||
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
|
||
|
||
-- INCR 返回自增后的新值，首次写入时才设置 TTL（避免每次请求重置窗口）
|
||
local new_count = redis.call('incr', KEYS[1])
|
||
if new_count == 1 then
|
||
redis.call('expire', KEYS[1], window)
|
||
end
|
||
|
||
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 沙箱限制
|
||
|
||
Redis 为了让 Lua 脚本在**主从同步**和 **AOF 重放**时结果一致，在沙箱中屏蔽了部分标准库函数。核心原则：**「给同样的输入，永远得到同样的输出」的函数才能用**。
|
||
|
||
```lua
|
||
-- ❌ 不可用：非确定性函数
|
||
math.random() -- 依赖内部 PRNG 状态，跨平台/版本结果不同
|
||
os.time() -- 依赖系统时间
|
||
os.getenv() -- 依赖环境变量
|
||
|
||
-- ✅ 推荐做法：在客户端生成随机值/时间戳，通过 ARGV 传入
|
||
-- local token = ARGV[2] -- 由应用层生成的 UUID
|
||
-- local now = ARGV[3] -- 由应用层传入的 unix 时间戳
|
||
-- redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[4])
|
||
```
|
||
|
||
> [!NOTE] 关于 Lua 沙箱中的数学函数
|
||
> - ✅ `math.max`、`math.min`、`math.abs`、`math.floor` 等——纯函数，输入确定输出就确定
|
||
> - ❌ `math.random`——虽然同一平台 + 同一 seed 理论上可复现，但 **Lua 5.1 PRNG 的实现在不同平台（Linux vs macOS、不同 libc 版本）之间不一致**，会导致主节点返回 42 而从节点重放得到 87，数据分裂
|
||
>
|
||
> 所以随机值必须在客户端生成，通过 ARGV 传进去。
|
||
|
||
> [!NOTE] Redis 7.0+ 两种沙箱模式
|
||
> Redis 7.0 引入了更严格的 Lua 沙箱。默认行为（等效于 `newdata-safe` 模式）：
|
||
>
|
||
> | 模式 | 行为 | 说明 |
|
||
> |------|------|------|
|
||
> | Legacy | 仅屏蔽非确定性函数 | 与 Redis < 7.0 兼容，标准库大部分可用 |
|
||
> | Newdata-safe（默认） | **仅允许 `redis.*` 和 `math`/`table`/`string`/`cjson`/`cmsgpack`** | 屏蔽 `loadfile`、`os.*`、`io.*`、`pcall` 等，安全性更高 |
|
||
>
|
||
> 同时 Redis 7.0 改变了 Lua 脚本的**复制方式**：默认将脚本产生的写命令（而非脚本本身）发送给从节点，从而避免从节点重复执行脚本，提升主从一致性。
|
||
> 生产环境建议保持默认——它能防止恶意脚本访问系统资源，并减少主从不一致的风险。
|
||
|
||
> [!QUESTION] 为什么 Lua 脚本不能有随机函数？
|
||
> Lua 脚本在主线程中同步执行，如果引入随机性或时间依赖，会导致两个严重问题：
|
||
> 1. **主从不一致**：主节点执行返回 42，从节点重放却得到 87，数据分裂。
|
||
> 2. **AOF 重放不可靠**：重启后从 appendonly.aof 重放脚本，结果与上次不同，状态机紊乱。
|
||
> 因此 Redis 在 Lua 环境中屏蔽了所有非确定性 API，保证「同样的输入一定得到同样的输出」。
|
||
|
||
### Lua 脚本最佳实践
|
||
|
||
Lua 脚本在 Redis **主线程**上同步执行——脚本运行期间，所有其他命令都会被阻塞。因此：
|
||
|
||
| 实践 | 说明 |
|
||
|------|------|
|
||
| 脚本尽量短小 | 只封装「必须原子执行」的逻辑，不要在里面做大循环 |
|
||
| 控制执行时间 | `lua-time-limit`（默认 5s）控制超时；超时后其他客户端可发送 `SCRIPT KILL`，**但前提是脚本尚未执行写操作**——若已写入，只能 `SHUTDOWN NOSAVE` |
|
||
| 脚本大小适度 | 超大脚本（几十 MB）会导致 EVAL 传输慢 + 编译耗时，应拆分到客户端或使用 EVALSHA 复用 |
|
||
| 复杂计算放客户端 | Lua 只负责「读-判断-写」，数据聚合/排序等重计算应在应用层完成 |
|
||
| 优先 EVALSHA | 避免每次都传输完整脚本，利用 SHA1 缓存减少带宽 |
|
||
|
||
> [!WARNING] 生产环境踩坑
|
||
> 曾有人在 Lua 里遍历几千个 key 做 `HGETALL`，导致 Redis 阻塞数秒——其他所有请求全部超时。**记住：Lua 脚本 = 临界区代码，越快越好。** 如果脚本已执行了写操作（`SET`/`DEL` 等），`SCRIPT KILL` 将失效，只能强制 `SHUTDOWN NOSAVE`——这在生产环境意味着数据丢失风险。
|
||
|
||
### Lua 脚本调试
|
||
|
||
开发阶段，可以用 `redis-cli --eval` 直接在命令行测试 Lua 脚本，无需启动应用：
|
||
|
||
```bash
|
||
# 语法：redis-cli --eval <脚本文件> <key1> <key2> , <arg1> <arg2>
|
||
# 注意逗号两边必须有空格，左边是 KEYS，右边是 ARGV
|
||
redis-cli --eval rate_limit.lua mykey , 100 60
|
||
```
|
||
|
||
> [!TIP] `SCRIPT DEBUG` 模式
|
||
> 在 redis-cli 中执行 `SCRIPT DEBUG SYNC` 后，接下来执行的 `EVAL` 会进入调试模式——你可以在脚本中用 `redis.log(redis.LOG_WARNING, variable)` 打印变量值，日志会输出到 Redis 日志文件。调试完毕后执行 `SCRIPT DEBUG NO` 关闭。
|
||
>
|
||
> 生产环境务必关闭调试模式，它会让脚本**同步执行**（正常是异步），显著影响性能。
|
||
|
||
## 三、Pipeline
|
||
|
||
### 为什么需要 Pipeline？
|
||
|
||
前面讲的事务和 Lua 解决的是「正确性」问题——保证命令不出错。但还有一个完全不同的问题：**性能**。
|
||
|
||
Redis 的通信协议是「你一句我一句」——客户端发一条命令，等服务器返回结果，再发下一条。每次这样的网络来回叫做 **RTT（Round-Trip Time，网络往返时间）**。假设 RTT = 1ms，执行 1000 条命令：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant C as "Client"
|
||
participant S as "Redis"
|
||
Note over C,S: "逐条发送 1000 次 RTT 约 1s"
|
||
C->>S: "SET key1 val1"
|
||
S-->>C: "OK"
|
||
C->>S: "SET key2 val2"
|
||
S-->>C: "OK"
|
||
C->>S: "... 还有 998 条"
|
||
```
|
||
|
||
**Pipeline 的思路很简单：把多条命令攒一批一起发，服务器也一批返回。** 网络只往返一次（或少数几次），而不是每条命令都来回跑。
|
||
|
||
| 方式 | RTT 次数 | 总耗时估算 | 比喻 |
|
||
|------|---------|-----------|------|
|
||
| 逐条发送 | 1000 | ~1s | 寄 1000 封信，每封等回信再寄下一封 |
|
||
| Pipeline (batch=200) | 5 次 | ~5ms | 攒 200 封信一起寄，回信也一批到 |
|
||
|
||
### 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 + TxPipeline（乐观锁 —— 读-改-写原子的核心模式）
|
||
// 核心思路：WATCH 监听 key → 在回调内读-判断-写 → 如果 key 被别人改了则整个回调失败重试
|
||
err := rdb.Watch(ctx, func(tx *redis.Tx) error {
|
||
// 1. 读：获取当前余额
|
||
acc, err := tx.Get(ctx, "account:alice").Result()
|
||
if err != nil { return err }
|
||
balance, _ := strconv.ParseInt(acc, 10, 64)
|
||
if balance < 100 {
|
||
return fmt.Errorf("余额不足")
|
||
}
|
||
// 2. 改-写：在 Watch 回调内用 TxPipeline 保证原子性
|
||
pipe := tx.TxPipeline()
|
||
pipe.DecrBy(ctx, "account:alice", 100)
|
||
pipe.IncrBy(ctx, "account:bob", 100)
|
||
_, err = pipe.Exec(ctx)
|
||
return err
|
||
}, "account:alice")
|
||
if err == redis.TxFailedErr {
|
||
// 被 Watch 的 key 被其他客户端修改了，需要重试整个操作
|
||
log.Println("并发冲突，重试...")
|
||
}
|
||
```
|
||
|
||
> [!QUESTION] WATCH 是什么？为什么需要它？
|
||
> `WATCH` 是 Redis 提供的**乐观锁**机制——在事务开始前「监视」一个或多个 key，如果这些 key 在 EXEC 之前被其他客户端修改了，整个事务会被拒绝（返回 `TxFailedErr`），你需要自行重试。
|
||
>
|
||
> 回顾上面的转账例子：`WATCH account:alice` → 读余额 → 判断 → `TxPipeline` 扣款。如果在你读余额之后、扣款之前，别人修改了 `account:alice`，`EXEC` 会失败，从而避免「余额不足却被扣款」的竞态条件。
|
||
>
|
||
> **一句话**：WATCH 让 MULTI/EXEC 具备了「检查-执行」的能力，弥补了事务不能读中间结果的短板。
|
||
|
||
> [!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
|
||
|
||
for i := 0; i < len(srcKeys); i += batchSize {
|
||
batch := srcKeys[i : min(i+batchSize, len(srcKeys))]
|
||
|
||
// 第一轮：用 Pipeline 批量 GET（而非逐条 Get，否则失去 Pipeline 意义）
|
||
getPipe := rdb.Pipeline()
|
||
getCmds := make([]*redis.StringCmd, len(batch))
|
||
for j, key := range batch {
|
||
getCmds[j] = getPipe.Get(ctx, key)
|
||
}
|
||
getPipe.Exec(ctx)
|
||
|
||
// 第二轮：用 Pipeline 批量 SET
|
||
setPipe := rdb.Pipeline()
|
||
for j, key := range batch {
|
||
val, err := getCmds[j].Result()
|
||
if err != nil {
|
||
continue // 跳过不存在的 key
|
||
}
|
||
newKey := strings.TrimPrefix(key, "old:")
|
||
setPipe.Set(ctx, newKey, val, 0)
|
||
}
|
||
if _, err := setPipe.Exec(ctx); err != nil {
|
||
return err
|
||
}
|
||
// Exec 后 pipeline 已清空，无需 Close() / 重建
|
||
}
|
||
return nil
|
||
}
|
||
```
|
||
|
||
## 四、Pub/Sub —— 发布订阅
|
||
|
||
前面讲的事务、Lua、Pipeline 都是「一个客户端跟 Redis 对话」。但有些场景需要**多个客户端之间通信**——比如 WebSocket 服务要广播消息给所有在线用户，或者配置中心更新后通知所有微服务。
|
||
|
||
Pub/Sub 就是 Redis 内置的「广播电台」：一个客户端发布消息，所有订阅了对应频道的客户端都能收到。
|
||
|
||
> [!WARNING] 先记住最关键的一点
|
||
> Pub/Sub 是**即发即忘**的——消息不会存储，订阅者如果下线了就收不到。如果你需要可靠投递（比如任务队列），请直接跳到后面的 [[#Stream]]。
|
||
|
||
### 基础用法
|
||
|
||
```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 广播、实时通知、配置变更推送
|
||
|
||
### Sharded Pub/Sub（Redis 7.0+）
|
||
|
||
普通 Pub/Sub 在集群模式下有个隐患：**所有消息都由接收 PUBLISH 命令的单个节点转发给所有订阅者**，该节点容易成为瓶颈。
|
||
|
||
Redis 7.0 引入了 Sharded Pub/Sub，核心改进是**消息按 slot 分发到各节点**，避免单点压力：
|
||
|
||
```bash
|
||
# 分片发布（消息只会发送到该 slot 所在节点的订阅者）
|
||
SSUBSCRIBE channel:order
|
||
SPUBLISH channel:order '{"id":42}'
|
||
|
||
# 普通发布（广播到所有节点的订阅者）
|
||
SUBSCRIBE channel:order
|
||
PUBLISH channel:order '{"id":42}'
|
||
```
|
||
|
||
| 维度 | 普通 Pub/Sub | Sharded Pub/Sub |
|
||
|------|-------------|----------------|
|
||
| 集群消息路由 | 单节点广播 → 所有节点 | 按 slot → 仅目标节点 |
|
||
| 适用场景 | 需要全局广播 | 高吞吐、按 key 分区消费 |
|
||
| 最低版本 | 所有版本 | Redis 7.0+ |
|
||
|
||
> [!QUESTION] 什么时候该用 Sharded Pub/Sub？
|
||
> 如果你的频道消息量大、且订阅者只需要消费属于自己 slot 的消息（比如按用户 ID 分片），Sharded Pub/Sub 能显著降低单节点压力。但如果需要**全局广播**（所有节点都收到），还是用普通 Pub/Sub。
|
||
|
||
### Stream —— 可靠的替代方案
|
||
|
||
如果 Pub/Sub 是「聊天群」（消息刷过去就没了），那 Stream 就是「消息队列」——**消息会持久化，消费者必须确认收到，挂了可以重新投递**。
|
||
|
||
Stream 是 Redis 5.0+ 引入的数据结构，核心设计围绕 **Consumer Group（消费者组）**：
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
P["Producer"] -->|"XADD: 写入消息"| S["Stream (持久化存储)"]
|
||
S -->|"XREADGROUP: 读取并标记 pending"| C1["Consumer A<br/>处理消息"]
|
||
S -->|"XREADGROUP: 读取并标记 pending"| C2["Consumer B<br/>处理消息"]
|
||
C1 -->|"XACK: 确认处理完成"| S
|
||
C2 -->|"XCLAIM: 接管超时消息"| S
|
||
```
|
||
|
||
核心流程三步走：
|
||
1. **生产者**用 `XADD` 写入消息
|
||
2. **消费者组**中的消费者用 `XREADGROUP` 读取——消息会被标记为 pending（待确认）
|
||
3. **消费者**处理完后用 `XACK` 确认——确认后消息才算真正消费完
|
||
|
||
如果消费者中途挂了，pending 消息不会丢失，其他消费者可以用 `XCLAIM` 接管。
|
||
|
||
```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 可靠的地方：**有状态、可追责**。
|
||
|
||
> [!TIP] Pub/Sub vs Stream 选型
|
||
> | 场景 | 推荐 |
|
||
> |------|------|
|
||
> | WebSocket 实时推送 | Pub/Sub |
|
||
> | 任务队列、订单事件 | Stream |
|
||
> | 海量消息、高吞吐 | Kafka/RabbitMQ |
|
||
|
||
> [!tip] Stream 深入阅读
|
||
> 本节仅覆盖 Stream 基础用法。Consumer Group 内部机制、消息确认与重试（XCLAIM/XAUTOCLAIM）、Dead Letter Queue 模式、容量管理等进阶内容，详见 [[hhs/Redis/15-Stream]]。
|
||
|
||
## 五、Keyspace Notifications —— 键事件监听
|
||
|
||
### 原理与配置
|
||
|
||
简单来说，Keyspace Notifications 就是 Redis 的「事件总线」——当某个 key 发生变化（被删除、过期、被修改等），Redis 会通过 Pub/Sub 发出一条通知。
|
||
|
||
> [!TIP] 最常见的用途
|
||
> 监听 key 过期事件。比如：用户 session 过期时自动清理关联数据、分布式锁到期时告警。
|
||
|
||
开启方式很简单，在 `redis.conf` 中设置（或运行时 `CONFIG SET`）：
|
||
|
||
```conf
|
||
notify-keyspace-events KEA
|
||
```
|
||
|
||
这一串字母是事件类型的组合，不用全部记住，按需选用即可：
|
||
|
||
| 字母 | 含义 | 举例 |
|
||
|------|------|------|
|
||
| `K` | Keyspace 事件（按 key 名通知） | 有人操作了 `user:123` |
|
||
| `E` | Keyevent 事件（按事件类型通知） | 有 key 过期了 |
|
||
| `A` | All events 简写，覆盖下方所有类型 | 一次性开启全部 |
|
||
| `x` | 过期事件 | `EXPIRE` 到期触发 |
|
||
| `e` | 淘汰事件 | 内存满时被驱逐 |
|
||
| `g` | 通用命令 | `DEL`、`EXPIRE`、`RENAME` 等 |
|
||
| `l/s/h/z` | 数据结构相关 | list/set/hash/sorted-set 操作 |
|
||
|
||
> [!NOTE] K 和 E 的区别——通知「谁」vs 通知「什么」
|
||
> - `__keyspace@0__:<keyname>` → "有人动了 `user:123`"（以 key 为中心）
|
||
> - `__keyevent@0__:expired` → "有 key 过期了"（以事件为中心）
|
||
>
|
||
> 两者是同一件事的两种视角，按需选用。**生产环境最常用 `Ex`（只监听过期事件）。**
|
||
|
||
```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` 通配一切
|
||
|
||
## 六、选型指南——如何选择？
|
||
|
||
面对一个需求，到底该用事务、Lua、Pipeline 还是 Stream？
|
||
|
||
### 决策心法——三个问题搞定
|
||
|
||
拿到需求后，先问自己三个问题：
|
||
|
||
1. **我需要「原子性」吗？** → 如果只是批量写入追求性能，Pipeline 就够了
|
||
2. **我需要「读-判断-写」吗？** → 如果答案是 yes，只有 Lua 能做到真正的逻辑原子性
|
||
3. **我在做「消息传递」吗？** → 实时广播用 Pub/Sub，需要可靠投递用 Stream
|
||
|
||
> [!TIP] 一句话记忆
|
||
> - **Pipeline**：只求快，不关心顺序 → 「快递员一次送 200 个包裹」
|
||
> - **MULTI/EXEC**：要求连续执行，不被插队 → 「排队结账，不允许别人插队」
|
||
> - **Lua 脚本**：需要根据中间结果做判断 → 「医生先检查再开药，不能分开」
|
||
> - **Pub/Sub**：实时广播，丢了就丢了 → 「微信群消息」
|
||
> - **Stream**：消息必须可靠送达 → 「挂号信，签收才算数」
|
||
|
||
下面是完整的决策流程：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
A["需求场景"] --> B{"批量命令?"}
|
||
B -->|"否"| C["单条命令即可"]
|
||
B -->|"是"| D{"需要原子性?"}
|
||
D -->|"否"| E["Pipeline"]
|
||
D -->|"是"| F{"读-判断-写?"}
|
||
F -->|"是"| G["Lua 脚本"]
|
||
F -->|"否"| H{"支持 Cluster?"}
|
||
H -->|"是"| G
|
||
H -->|"否"| I["MULTI / EXEC"]
|
||
A --> J{"消息传递?"}
|
||
J -->|"可靠投递/任务队列"| K["Stream"]
|
||
J -->|"实时广播/通知"| L["Pub/Sub"]
|
||
```
|
||
|
||
| 需求 | 推荐方案 | 理由 |
|
||
|------|---------|------|
|
||
| 批量写入、减少 RTT | Pipeline | 无原子性要求，纯性能优化 |
|
||
| 读-判断-写原子操作 | Lua 脚本 | 逻辑原子性，一气呵成 |
|
||
| 简单批量 + 单机 | MULTI / EXEC | 轻量，无需写 Lua |
|
||
| 任务队列、事件溯源 | Stream | 持久化 + Consumer Group |
|
||
| 实时通知、WebSocket 广播 | Pub/Sub | 无状态、低延迟 |
|
||
|
||
## 关联笔记
|
||
|
||
- [[hhs/Redis/09-高级特性/分布式锁]] — 分布式锁完整解析（Redlock、Watchdog、可重入锁）
|
||
- [[hhs/Redis/06-主从与哨兵]] — Sentinel 故障检测（也是基于 PING/PONG 心跳）
|
||
- [[hhs/Redis/07-集群方案]] — Cluster 下 Pipeline/Lua 的限制
|
||
- [[hhs/Redis/README]] — 知识索引总览
|