go-redis Pipeline 与事务

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 性能实测 & 调优

• 最佳区间 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 脚本

结论：逻辑简单 → 事务；多 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 常见坑与最佳实践

5 生产级 Checklist

1. Pipeline 批量：50-200 条最优；阻塞命令 (BLPOP) 另开连接。
2. 事务重试：封装带退避 (exponential back-off) 的 Retry。
3. 连接池：PoolSize = CPU*10，MinIdleConns ≈ 20% PoolSize。
4. 超时：DialTimeout 100ms、Read/WriteTimeout 200ms 典型值。
5. 可观测：redisotel.InstrumentTracing/Metrics 接入 OTel。
6. 幂等命令：重试需确保无副作用。
7. Lua 脚本：库存扣减、抢红包等使用脚本更稳。
8. RESP3：如 Redis ≥ 6.0，可设置 Protocol: 3 享受 Map/Push 类型。

6 结语

• Pipeline 带来吞吐提升，适合大量写入与批量读写。
• 事务 提供原子操作，确保数据一致。
• Watch 则在并发场景下守护一致性。

合理组合三者，配合连接池调优与可观测监控，你就能构建 既快又稳 的 Redis 访问层。祝编码愉快，TPS 飙升！