Go Redis 管道与事务批量

示例说明

逐条发送命令时每条都要经历一次网络往返——一百毫秒延迟下十条命令就是一秒。Pipeline 把命令攒成一批一次发出、按序收回，延迟与命令数解耦。TxPipeline 更进一步：命令包进 MULTI/EXEC 块，服务端保证执行期间不被其他客户端插入——计数器加过期时间的成对操作就有了原子保障。

技术要点

• Pipeline 合并网络往返次数
• TxPipeline 走 MULTI/EXEC 原子执行
• 命令结果按序从命令对象取回

代码解读

seedAndMeasure 先用普通管道对三个指标键自增，命令对象数组保存引用；再用事务管道批量设置一小时过期。两次 Exec 分别完成批量往返与原子提交，之后从两类命令对象读取结果并打印。上下文超时统一约束整批操作的总时长。

使用建议

管道过大可拆分防止单次响应超限。事务内命令失败其余仍会执行，需要失败中止语义要配合 Lua。结果丢弃前必须 Exec 否则命令不发出。

代码示例

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

func seedAndMeasure(ctx context.Context, client *redis.Client) error {
	keys := []string{"metric:hits", "metric:misses", "metric:errors"}

	// 独立命令走管道：一次网络往返
	pipe := client.Pipeline()
	incrCmds := make([]*redis.IntCmd, len(keys))
	for i, key := range keys {
		incrCmds[i] = pipe.Incr(ctx, key)
	}
	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("pipeline exec: %w", err)
	}

	// 事务管道：命令打包进 MULTI/EXEC 原子执行
	tx := client.TxPipeline()
	expireCmds := make([]*redis.BoolCmd, len(keys))
	for i, key := range keys {
		expireCmds[i] = tx.Expire(ctx, key, time.Hour)
	}
	if _, err := tx.Exec(ctx); err != nil {
		return fmt.Errorf("tx exec: %w", err)
	}

	for i, key := range keys {
		fmt.Printf("%s=%d expires=%v\n",
			key, incrCmds[i].Val(), expireCmds[i].Val())
	}
	return nil
}

func main() {
	client := redis.NewClient(&redis.Options{
		Addr:        "127.0.0.1:6379",
		PoolSize:    10,
		ReadTimeout: 500 * time.Millisecond,
	})
	defer client.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	if err := seedAndMeasure(ctx, client); err != nil {
		fmt.Println("failed:", err)
	}
}