Archived

This repository has been archived on 2026-05-24. You can view files and clone it. You cannot open issues or pull requests or push a commit.

Files

T

hhs bdeb889ee4 vault backup: 2026-05-17 00:06:11

2026-05-17 00:06:11 +08:00

13 KiB

Raw Permalink Blame History

tags, create time

Redis 高级特性

概述

掌握事务、Lua 脚本和 Pipeline，是 Redis 从「基础缓存」迈向「生产级系统」的关键。本节逐一剖析其原理、场景和陷阱。

一、事务（MULTI / EXEC）

基本用法

MULTI                # 开启事务（不阻塞，之后命令只是排队）
INCR counter         # QUEUED
INCR other_counter   # QUEUED
EXEC                 # 一次性执行所有排队的命令

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

事务的错误处理

MULTI
SET k v NX
GET KEYY xxx    # ⚠️ 语法错误！
EXEC

错误类型	发生时	行为
编译时错误（语法、参数）	EXEC 之前	EXEC 拒绝执行整个事务
运行时错误（类型不匹配）	EXEC 期间	当前命令报错，其余继续执行

MULTI
SET k 123       # OK, queued
INCR k          # OK, queued (先 SET 再 INCR 是合法的)
EXEC
# → [OK, 124] — 两个都成功了

[!WARNING] Redis 事务 ≠ 数据库事务

没有 ROLLBACK / ACID 保证

不能捕获异常后回滚

如果需要强一致性，请用 Lua 脚本

二、Lua 脚本

为什么需要 Lua？

核心目的：原子性操作多个命令。MULTI 只能保证"不插队"，Lua 能保证"逻辑原子性"（读-判断-写一气呵成）。

EVAL "return redis.call('GET', KEYS[1])" 1 mykey
#  ↑ script     ↑ commands         ↑ key count  ↑ keys...

经典场景 1：分布式锁

-- 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 调用
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：限流器（令牌桶简化版）

-- 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：订单状态机

-- 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

EVAL "script..." numkeys args...     # 每次传完整脚本
EVALSHA <sha1> numkeys args...       # 只传 SHA1，省带宽

[!TIP] 使用 EVALSHA 的最佳实践
script := redis.NewScript(...)
// 首次调用失败可能是 NOSCRIPT（缓存未命中），会自动重试 EVAL
result := script.Run(ctx, ...).Int()
go-redis 内部已处理了 NOSCRIPT → EVAL fallback 的逻辑。

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 脚本在主线程中同步执行，如果引入随机性或时间依赖，会导致两个严重问题：

主从不一致：主节点执行返回 42，从节点重放却得到 87，数据分裂。

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

// 方式一：普通 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 实战——数据迁移

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 —— 发布订阅

基础用法

# 消费者 A
SUBSCRIBE channel:newposts

# 消费者 B（也可以 subscribe 同一频道）
SUBSCRIBE channel:newposts

# 发布者
PUBLISH channel:newposts '{"id":42,"title":"Redis Tips"}'
→ 消费者 A 和 B 同时收到消息

// 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（消费者组）：

# 生产者写入
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 消息

标识符	含义	使用场景
`$`	"最后一条已写入的消息"	新 consumer 加入时，只接收将来的消息
`0`	"历史第一条"	数据迁移/重放（谨慎用于大 stream）
`>`	"我只取未分配给我的新消息"	正常消费循环中始终用 `>`，不取到 pending 消息

[!QUESTION] 如果消费者挂了怎么办？好消息会被移到 pending 列表。用 XPENDING mystream group1 查看，用 XCLAIM 将 pending 消息转移给其他 consumer 重新处理。这就是 Stream 比 Pub/Sub 可靠的地方：有状态、可追责。

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

场景	推荐
WebSocket 实时推送	Pub/Sub
任务队列、订单事件	Stream
海量消息、高吞吐	Kafka/RabbitMQ

五、Keyspace Notifications —— 键事件监听

原理与配置

Redis 会在每个 key 发生变更时发送通知，订阅者通过 Pub/Sub 协议接收。开启方式：

# 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 淘汰)

# 订阅过期事件（Keyevent 模式：更精确）
PSUBSCRIBE __keyevent@0__:expired

# 另一个连接设置过期 key
EXPIRE test:key 2
→ 收到: __keyevent@0__:expired "test:key"

// 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 — 知识索引总览

13 KiB Raw Permalink Blame History Unescape Escape