go-redisPipeline与事务的实现示例

目录

• 1 背景与动机
• 2 Pipeline：降低 RTT 的秘密武器
• 2.1 基础用法
• 2.2 自动化Pipelined()
• 2.3 性能实测 & 调优
• 3 事务：一次提交，全部成功
• 3.1TxPipeline()基础
• 3.2TxPipelined()回调
• 3.3 事务 vs Lua 脚本
• 4 乐观锁：Watch 机制剖析
• 4.1 完整重试模型
• 4.2 常见坑与最佳实践
• 5 生产级 Checklist
• 6 结语

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

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

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

// 初始化
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

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、代码更简洁，非常适合服务层一次性批量操作。

批量大小	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 说明批量过大或超时过短。

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
}

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

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

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

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

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)                             // 其他错误
}

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

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 类型。

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

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

到此这篇关于go-redis Pipeline与事务的实现示例的文章就介绍到这了,更多相关go-redis Pipeline与事务内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• 如何在Go语言中高效使用Redis的Pipeline
• Golang实现自己的Redis(pipeline客户端)实例探索
• golang redis中Pipeline通道的使用详解
• Golang 实现 Redis系列(六)如何实现 pipeline 模式的 redis 客户端