Go Redis 管道和事务之 go-redis

Go Redis 管道和事务之 go-redis

Go Redis 管道和事务官方文档介绍

Redis pipelines(管道) 允许一次性发送多个命令来提高性能，go-redis支持同样的操作， 你可以使用go-redis一次性发送多个命令到服务器，并一次读取返回结果，而不是一个个命令的操作。
Go Redis 管道和事务: https://redis.uptrace.dev/zh/guide/go-redis-pipelines.html

• 管道
  
• Watch 监听
  
• 事务
  

#管道

通过 go-redis Pipeline 一次执行多个命令并读取返回值:

1. pipe := rdb.Pipeline()
3. incr := pipe.Incr(ctx, "pipeline_counter")
4. pipe.Expire(ctx, "pipeline_counter", time.Hour)
6. cmds, err := pipe.Exec(ctx)
7. if err != nil {
8.         panic(err)
9. }
11. // 结果你需要再调用 Exec 后才可以使用
12. fmt.Println(incr.Val())

复制代码

或者你也可以使用 Pipelined 方法，它将自动调用 Exec:

1. var incr *redis.IntCmd
3. cmds, err := rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
4.         incr = pipe.Incr(ctx, "pipelined_counter")
5.         pipe.Expire(ctx, "pipelined_counter", time.Hour)
6.         return nil
7. })
8. if err != nil {
9.         panic(err)
10. }
12. fmt.Println(incr.Val())

复制代码

同时会返回每个命令的结果，你可以遍历结果集：

1. cmds, err := rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
2.         for i := 0; i < 100; i++ {
3.                 pipe.Get(ctx, fmt.Sprintf("key%d", i))
4.         }
5.         return nil
6. })
7. if err != nil {
8.         panic(err)
9. }
11. for _, cmd := range cmds {
12.     fmt.Println(cmd.(*redis.StringCmd).Val())
13. }

复制代码

#Watch 监听

使用 Redis 事务， 监听key的状态，仅当key未被其他客户端修改才会执行命令， 这种方式也被成为 乐观锁。
Redis 事务：https://redis.io/docs/manual/transactions/
乐观锁：

1. WATCH mykey
3. val = GET mykey
4. val = val + 1
6. MULTI
7. SET mykey $val
8. EXEC

复制代码

#事务

你可以使用 TxPipelined 和 TxPipeline 方法，把命令包装在 MULTI 、 EXEC 中， 但这种做法没什么意义：

1. cmds, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
2.         for i := 0; i < 100; i++ {
3.                 pipe.Get(ctx, fmt.Sprintf("key%d", i))
4.         }
5.         return nil
6. })
7. if err != nil {
8.         panic(err)
9. }
11. // MULTI
12. // GET key0
13. // GET key1
14. // ...
15. // GET key99
16. // EXEC

复制代码

你应该正确的使用 Watch + 事务管道， 比如以下示例，我们使用 GET, SET 和 WATCH 命令，来实现 INCR 操作， 注意示例中使用 redis.TxFailedErr 来判断失败：

1. const maxRetries = 1000
3. // increment 方法，使用 GET + SET + WATCH 来实现Key递增效果，类似命令 INCR
4. func increment(key string) error {
5.         // 事务函数
6.         txf := func(tx *redis.Tx) error {
7.    // // 获得当前值或零值
8.                 n, err := tx.Get(ctx, key).Int()
9.                 if err != nil && err != redis.Nil {
10.                         return err
11.                 }
13.                 n++  // 实际操作
15.     // 仅在监视的Key保持不变的情况下运行
16.                 _, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
17.                         pipe.Set(ctx, key, n, 0)
18.                         return nil
19.                 })
20.                 return err
21.         }
22.        
23.         for i := 0; i < maxRetries; i++ {
24.                 err := rdb.Watch(ctx, txf, key)
25.                 if err == nil {
26.                         // Success.
27.                         return nil
28.                 }
29.                 if err == redis.TxFailedErr {
30.                         // 乐观锁失败
31.                         continue
32.                 }
33.                 return err
34.         }
36.         return errors.New("increment reached maximum number of retries")
37. }

复制代码

Go Redis 管道和事务 实操

1. package main
3. import (
4.         "context"
5.         "fmt"
6.         "github.com/redis/go-redis/v9"
7.         "time"
8. )
10. // 声明一个全局的 rdb 变量
11. var rdb *redis.Client
13. // 初始化连接
14. func initRedisClient() (err error) {
15.         // NewClient将客户端返回给Options指定的Redis Server。
16.         // Options保留设置以建立redis连接。
17.         rdb = redis.NewClient(&redis.Options{
18.                 Addr:     "localhost:6379",
19.                 Password: "", // 没有密码，默认值
20.                 DB:       0,  // 默认DB 0 连接到服务器后要选择的数据库。
21.                 PoolSize: 20, // 最大套接字连接数。 默认情况下，每个可用CPU有10个连接，由runtime.GOMAXPROCS报告。
22.         })
24.         // Background返回一个非空的Context。它永远不会被取消，没有值，也没有截止日期。
25.         // 它通常由main函数、初始化和测试使用，并作为传入请求的顶级上下文
26.         ctx := context.Background()
28.         _, err = rdb.Ping(ctx).Result()
29.         if err != nil {
30.                 return err
31.         }
32.         return nil
33. }
35. // watchDemo 在key值不变的情况下将其值+1
36. func watchKeyDemo(key string) error {
37.         ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
38.         defer cancel()
40.         // Watch准备一个事务，并标记要监视的密钥，以便有条件执行(如果有密钥的话)。
41.         // 当fn退出时，事务将自动关闭。
42.         // func (c *Client) Watch(ctx context.Context, fn func(*Tx) error, keys ...string)
43.         return rdb.Watch(ctx, func(tx *redis.Tx) error {
44.                 // Get Redis `GET key` command. It returns redis.Nil error when key does not exist.
45.                 // 获取 Key 的值 n
46.                 n, err := tx.Get(ctx, key).Int()
47.                 if err != nil && err != redis.Nil {
48.                         fmt.Printf("redis get failed, err: %v\n", err)
49.                         return err
50.                 }
51.                 // 假设操作耗时5秒
52.                 // 5秒内我们通过其他的客户端修改key，当前事务就会失败
53.                 time.Sleep(5 * time.Second)
54.                 // txpipeline 执行事务中fn队列中的命令。
55.                 // 当使用WATCH时，EXEC只会在被监视的键没有被修改的情况下执行命令，从而允许检查和设置机制。
56.                 // Exec总是返回命令列表。如果事务失败，则返回TxFailedErr。否则Exec返回第一个失败命令的错误或nil
57.                 _, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
58.                         // 业务逻辑 如果 Key 没有变化，则在原来的基础上加 1
59.                         pipe.Set(ctx, key, n+1, time.Hour)
60.                         return nil
61.                 })
62.                 return err
63.         }, key)
64. }
66. func main() {
67.         if err := initRedisClient(); err != nil {
68.                 fmt.Printf("initRedisClient failed: %v\n", err)
69.                 return
70.         }
71.         fmt.Println("initRedisClient started successfully")
72.         defer rdb.Close() // Close 关闭客户端，释放所有打开的资源。关闭客户端是很少见的，因为客户端是长期存在的，并在许多例程之间共享。
75.         err := watchKeyDemo("watch_key")
76.         if err != nil {
77.                 fmt.Printf("watchKeyDemo failed: %v\n", err)
78.                 return
79.         }
80.         fmt.Printf("watchKeyDemo succeeded!\n")
81. }

复制代码

运行
[code]Code/go/redis_demo via
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！