go-redis Pipeline与事务的实现示例

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

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

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

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