Golang 操作 Redis：Pipeline操作用法 - go-redis 使用指南

在上一篇文章中，我们探讨了 Redis 中 HyperLogLog 的操作和使用场景，详细介绍了如何通过 go-redis 实现相关功能。如果你还没有读过，可以点击这里查看。本篇文章将聚焦于 Redis Pipeline 的操作，通过 go-redis 库来高效地执行多个 Redis 命令。我们将介绍 Pipeline 的基本概念、常见使用场景，并通过示例代码详细解析各个操作方法的用法。

👉 点击查看《go-redis使用指南》系列文章目录

在《go-redis使用指南》系列文章中，我们将详细介绍如何在 Golang 项目中使用 redis/go-redis 库与 Redis 进行交互。以下是该系列文章的全部内容：

Redis Pipeline 简介

Redis Pipeline 是一种用来批量执行多个 Redis 命令的技术。它可以将多个命令一起发送到 Redis 服务器，而不必等待每个命令执行完毕后再发送下一个，从而减少了网络往返的次数，提升了操作的效率。这在需要执行大量命令时尤其有效，典型的使用场景包括：

批量写入数据，如批量设置键值对。
批量读取数据，如获取多个键的值。

Pipeline 并不是事务，虽然它可以批量执行命令，但不能保证命令之间的原子性。如果需要原子性操作，应该使用 Redis 事务（MULTI/EXEC）或 Lua 脚本。

go-redis 中 Pipeline 操作的方法

以下是 go-redis 中 Pipeline 操作的相关方法及其功能描述：

Len() - 获取 Pipeline 中未执行的命令数量。
Do(ctx context.Context, args ...interface{}) *Cmd - 执行任意 Redis 命令，适用于 go-redis 尚未支持的命令。
Process(ctx context.Context, cmd Cmder) error - 将要执行的命令放入 Pipeline 缓存中。
Discard() - 丢弃缓存中所有未执行的命令。
Exec(ctx context.Context) ([]Cmder, error) - 将 Pipeline 中缓存的所有命令发送到 Redis 服务器并执行。

go-redis Pipeline 操作方法详细讲解及示例代码

⚠️ 注意：在使用 Redis Pipeline 时，所有命令的返回值在 Exec() 调用之前都是未定义的，只有在 Exec() 调用之后，你才能够获取到实际的返回值。因此，在需要对返回结果进行处理时，确保在 Exec() 之后再访问这些返回值。

Pipeline/Exec 示例:

下面我们通过一个完整的示例代码，展示如何在 Golang 中使用 go-redis 的 Pipeline 操作上述所有方法。

Pipeline 和 Exec 是两个分开的步骤。首先使用 Pipeline() 方法创建一个 Pipeline 对象。然后，通过在这个 Pipeline 对象上连续调用 Redis 命令方法（如 Set、Get 等），将命令添加到 Pipeline 中。最后，使用 Exec() 方法将所有命令发送到 Redis 服务器并执行。

package main

import (
	"context"
	"fmt"

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

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})

	pipe := rdb.Pipeline()

	// 添加命令到Pipeline
	pipe.Set(ctx, "key1", "value1", 0)
	pipe.Set(ctx, "key2", "value2", 0)

	// 使用Len方法获取Pipeline中未执行的命令数量
	fmt.Println("Pipeline length:", pipe.Len())
	// 输出：Pipeline length: 2

	// 使用Do方法添加未支持的Redis命令到Pipeline中
	pipe.Do(ctx, "SET", "key3", "value3")

	// 创建一个新命令并使用Process方法将其添加到Pipeline中
	cmd := redis.NewStringCmd(ctx, "SET", "key4", "value4")
	err := pipe.Process(ctx, cmd)
	if err != nil {
		fmt.Println("Error processing command:", err)
	}

	// 再次使用Len方法查看命令数量
	fmt.Println("Pipeline length after adding commands:", pipe.Len())
	// 输出：Pipeline length after adding commands: 4

	// 使用Discard方法丢弃所有未执行的命令
	pipe.Discard()

	// 丢弃后的Pipeline命令数量应为0
	fmt.Println("Pipeline length after discard:", pipe.Len())
	// 输出：Pipeline length after discard: 0

	// 再次添加命令到Pipeline
	pipe.Set(ctx, "key5", "value5", 0)
	incr := pipe.Incr(ctx, "counter")
    // 在pipeline还没真正执行时是取不到正确值的
	fmt.Println("incr before pipeline exec:", incr.Val())
	// 输出：incr before pipeline exec: 0

	// 使用Exec方法执行Pipeline中的所有命令
	cmds, err := pipe.Exec(ctx)
	if err != nil {
		fmt.Println("Pipeline execution error:", err)
	}

	fmt.Println("incr after pipeline exec:", incr.Val())
	// 输出：incr after pipeline exec: 1

	fmt.Println(cmds)
	// 输出：[set key5 value5: OK incr counter: 1]
}

Pipelined 示例:

Pipelined 是一个单一的函数，它将命令的创建和执行合并在一起。你传递一个回调函数给 Pipelined，在这个回调函数中，你可以直接调用 Redis 命令方法，这些命令会自动加入到 Pipeline 中。

当回调函数返回时，Pipelined 会自动调用 Exec，并返回所有命令的结果。

方法签名:

Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)

参数说明:

ctx：上下文，用于控制命令的执行。
fn：一个回调函数，接受Pipeliner接口，用于定义 Pipeline 中的命令。

返回结果说明: 返回一个[]Cmder数组，包含所有命令的执行结果，以及一个error对象，表示执行过程中的错误信息。

示例代码:

package main

import (
	"context"
	"fmt"
	"time"

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

func main() {
	ctx := context.Background()
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})

	var incr *redis.IntCmd

	// 使用Pipelined方法
	cmds, err := rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		incr = pipe.Incr(ctx, "pipelined_counter")
		// 再次强调，在pipeline还没真正执行时是取不到正确值的
		// 因此此处incr的值是不对的
		fmt.Println(incr.Val())
		// 输出：0

		pipe.Expire(ctx, "pipelined_counter", time.Hour)

		for i := 0; i < 5; i++ {
			pipe.Set(ctx, fmt.Sprintf("key%d", i), i, 0)
		}
		return nil
	})
	if err != nil {
		panic(err)
	}

	// 在Pipeline执行后获取命令结果
	fmt.Println(incr.Val())
	// 输出：1

	// 另一个Pipelined的示例
	cmds, err = rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		for i := 0; i < 5; i++ {
			pipe.Get(ctx, fmt.Sprintf("key%d", i))
		}
		return nil
	})
	if err != nil {
		panic(err)
	}

	// 遍历获取所有命令的结果
	for _, cmd := range cmds {
		fmt.Println(cmd.Args(), cmd.(*redis.StringCmd).Val())
	}
	// 输出：
	// [get key0] 0
	// [get key1] 1
	// [get key2] 2
	// [get key3] 3
	// [get key4] 4
}