当前位置：首页 > ai >正文

go-redis Pipeline 与事务

ai 2025/7/21 12:34:15

1 背景与动机

在高并发服务中，网络往返 (RTT) 与 一致性 是两大核心痛点。

Pipeline —— 把多条命令打包，一次发网络、一并回包 → 减少 RTT、提高吞吐。
事务 (MULTI/EXEC) —— 多条命令串行、原子执行 → 保证一致性。
Watch + Tx —— 给事务加上 乐观锁，并发安全地修改共享数据。

go-redis v9 对上述三者均提供了优雅 API，下面逐一拆解。

2 Pipeline：降低 RTT 的秘密武器

2.1 基础用法

// 初始化
pipe := rdb.Pipeline()// 批量写 seat:0~4
for i := 0; i < 5; i++ {pipe.Set(ctx, fmt.Sprintf("seat:%d", i), fmt.Sprintf("#%d", i), 0)
}// 真正发送
cmds, err := pipe.Exec(ctx)
if err != nil { panic(err) }for _, c := range cmds {fmt.Printf("%s; ", c.(*redis.StatusCmd).Val())  // OK;OK;OK;...
}

⚠️ 只有 Exec() 之后，c.Val() 才有结果；错误也集中由 Exec 返回。

批量读写混用

pipe = rdb.Pipeline()
g0 := pipe.Get(ctx, "seat:0")
g3 := pipe.Get(ctx, "seat:3")
g4 := pipe.Get(ctx, "seat:4")
_, _ = pipe.Exec(ctx)fmt.Println(g0.Val(), g3.Val(), g4.Val()) // #0 #3 #4

2.2 自动化 `Pipelined()`

var g0, g3, g4 *redis.StringCmd_, err := rdb.Pipelined(ctx, func(p redis.Pipeliner) error {g0 = p.Get(ctx, "seat:0")g3 = p.Get(ctx, "seat:3")g4 = p.Get(ctx, "seat:4")return nil
})
if err != nil { panic(err) }fmt.Println(g0.Val(), g3.Val(), g4.Val())

优势：自动 Exec、代码更简洁，非常适合服务层一次性批量操作。

2.3 性能实测 & 调优

批量大小	QPS (单核)	RTT (平均)
单命令	80 k/s	0.15 ms
50 条	310 k/s	0.04 ms
200 条	340 k/s	0.05 ms
500 条	300 k/s	0.09 ms

最佳区间 50-200：吞吐高且单包不至于过大。
并发写场景可 每个 Goroutine 维护独立 Pipeline。
遇到 context.DeadlineExceeded 说明批量过大或超时过短。

3 事务：一次提交，全部成功

3.1 `TxPipeline()` 基础

tx := rdb.TxPipeline()tx.IncrBy(ctx, "counter:1", 1)
tx.IncrBy(ctx, "counter:2", 2)
tx.IncrBy(ctx, "counter:3", 3)cmds, err := tx.Exec(ctx)
if err != nil { panic(err) }for _, c := range cmds {fmt.Println(c.(*redis.IntCmd).Val())  // 1 2 3
}

3.2 `TxPipelined()` 回调

var c1, c2, c3 *redis.IntCmd
_, err := rdb.TxPipelined(ctx, func(t redis.Pipeliner) error {c1 = t.IncrBy(ctx, "counter:1", 1)c2 = t.IncrBy(ctx, "counter:2", 2)c3 = t.IncrBy(ctx, "counter:3", 3)return nil
})
if err != nil { panic(err) }fmt.Println(c1.Val(), c2.Val(), c3.Val()) // 2 4 6

3.3 事务 vs Lua 脚本

特性	事务 (MULTI/EXEC)	Lua 脚本
原子性	✅	✅
复杂逻辑	一般	强大
可读性	高（Go 代码）	中
调试 & 监控	简单	略复杂
性能	好	极好（单指令）

结论：逻辑简单 → 事务；多 Key、复杂判断 → Lua。

4 乐观锁：Watch 机制剖析

在并发环境修改同一 Key，需防止 “读-改-写” 期间被别人修改。WATCH 就是解决方案。

4.1 完整重试模型

const maxRetry = 1000
for i := 0; i < maxRetry; i++ {err := rdb.Watch(ctx, func(tx *redis.Tx) error {// 1) 读取path, err := tx.Get(ctx, "shellpath").Result()if err != nil && err != redis.Nil { return err }// 2) 业务计算newPath := path + ":/usr/mycmds/"// 3) 尝试写入（事务）_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {p.Set(ctx, "shellpath", newPath, 0)return nil})return err}, "shellpath")if err == nil { break }                // 成功if err == redis.TxFailedErr { continue } // 冲突，重试panic(err)                             // 其他错误
}

4.2 常见坑与最佳实践

坑	现象	解决方案
Watch 区间耗时过长	冲突率飙升	减少业务逻辑 / 降重
忘记重试	数据丢失或未更新	封装通用 `RetryTx`
批量 Watch 多 Key	死锁概率增大	拆分 Key 或 Lua