Go Redis 管道和事務(wù)之 go-redis

Go Redis 管道和事務(wù)官方文檔介紹

Redis pipelines(管道) 允許一次性發(fā)送多個(gè)命令來提高性能，go-redis支持同樣的操作， 你可以使用go-redis一次性發(fā)送多個(gè)命令到服務(wù)器，并一次讀取返回結(jié)果，而不是一個(gè)個(gè)命令的操作。

Go Redis 管道和事務(wù): https://redis.uptrace.dev/zh/guide/go-redis-pipelines.html

• 管道

• Watch 監(jiān)聽

• 事務(wù)

#管道

通過 go-redis Pipeline 一次執(zhí)行多個(gè)命令并讀取返回值:

pipe := rdb.Pipeline()

incr := pipe.Incr(ctx, "pipeline_counter")
pipe.Expire(ctx, "pipeline_counter", time.Hour)

cmds, err := pipe.Exec(ctx)
if err != nil {
	panic(err)
}

// 結(jié)果你需要再調(diào)用 Exec 后才可以使用
fmt.Println(incr.Val())

或者你也可以使用 Pipelined 方法，它將自動(dòng)調(diào)用 Exec:

var incr *redis.IntCmd

cmds, err := rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
	incr = pipe.Incr(ctx, "pipelined_counter")
	pipe.Expire(ctx, "pipelined_counter", time.Hour)
	return nil
})
if err != nil {
	panic(err)
}

fmt.Println(incr.Val())

同時(shí)會(huì)返回每個(gè)命令的結(jié)果，你可以遍歷結(jié)果集：

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

for _, cmd := range cmds {
    fmt.Println(cmd.(*redis.StringCmd).Val())
}

#Watch 監(jiān)聽

使用 Redis 事務(wù)， 監(jiān)聽key的狀態(tài)，僅當(dāng)key未被其他客戶端修改才會(huì)執(zhí)行命令， 這種方式也被成為 樂觀鎖。

Redis 事務(wù)：https://redis.io/docs/manual/transactions/

樂觀鎖：

WATCH mykey

val = GET mykey
val = val + 1

MULTI
SET mykey $val
EXEC

#事務(wù)

你可以使用 TxPipelined 和 TxPipeline 方法，把命令包裝在 MULTI 、 EXEC 中， 但這種做法沒什么意義：

cmds, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
	for i := 0; i < 100; i++ {
		pipe.Get(ctx, fmt.Sprintf("key%d", i))
	}
	return nil
})
if err != nil {
	panic(err)
}

// MULTI
// GET key0
// GET key1
// ...
// GET key99
// EXEC

你應(yīng)該正確的使用 Watch + 事務(wù)管道， 比如以下示例，我們使用 GET, SET 和 WATCH 命令，來實(shí)現(xiàn) INCR 操作， 注意示例中使用 redis.TxFailedErr 來判斷失?。?/p>

const maxRetries = 1000

// increment 方法，使用 GET + SET + WATCH 來實(shí)現(xiàn)Key遞增效果，類似命令 INCR
func increment(key string) error {
	// 事務(wù)函數(shù)
	txf := func(tx *redis.Tx) error {
   // // 獲得當(dāng)前值或零值 
		n, err := tx.Get(ctx, key).Int()
		if err != nil && err != redis.Nil {
			return err
		}

		n++  // 實(shí)際操作

    // 僅在監(jiān)視的Key保持不變的情況下運(yùn)行
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Set(ctx, key, n, 0)
			return nil
		})
		return err
	}
	
	for i := 0; i < maxRetries; i++ {
		err := rdb.Watch(ctx, txf, key)
		if err == nil {
			// Success.
			return nil
		}
		if err == redis.TxFailedErr {
			// 樂觀鎖失敗
			continue
		}
		return err
	}

	return errors.New("increment reached maximum number of retries")
}

Go Redis 管道和事務(wù) 實(shí)操

package main

import (
	"context"
	"fmt"
	"github.com/redis/go-redis/v9"
	"time"
)

// 聲明一個(gè)全局的 rdb 變量
var rdb *redis.Client

// 初始化連接
func initRedisClient() (err error) {
	// NewClient將客戶端返回給Options指定的Redis Server。
	// Options保留設(shè)置以建立redis連接。
	rdb = redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "", // 沒有密碼，默認(rèn)值
		DB:       0,  // 默認(rèn)DB 0 連接到服務(wù)器后要選擇的數(shù)據(jù)庫。
		PoolSize: 20, // 最大套接字連接數(shù)。 默認(rèn)情況下，每個(gè)可用CPU有10個(gè)連接，由runtime.GOMAXPROCS報(bào)告。
	})

	// Background返回一個(gè)非空的Context。它永遠(yuǎn)不會(huì)被取消，沒有值，也沒有截止日期。
	// 它通常由main函數(shù)、初始化和測(cè)試使用，并作為傳入請(qǐng)求的頂級(jí)上下文
	ctx := context.Background()

	_, err = rdb.Ping(ctx).Result()
	if err != nil {
		return err
	}
	return nil
}

// watchDemo 在key值不變的情況下將其值+1
func watchKeyDemo(key string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// Watch準(zhǔn)備一個(gè)事務(wù)，并標(biāo)記要監(jiān)視的密鑰，以便有條件執(zhí)行(如果有密鑰的話)。
	// 當(dāng)fn退出時(shí)，事務(wù)將自動(dòng)關(guān)閉。
	// func (c *Client) Watch(ctx context.Context, fn func(*Tx) error, keys ...string)
	return rdb.Watch(ctx, func(tx *redis.Tx) error {
		// Get Redis `GET key` command. It returns redis.Nil error when key does not exist.
		// 獲取 Key 的值 n
		n, err := tx.Get(ctx, key).Int()
		if err != nil && err != redis.Nil {
			fmt.Printf("redis get failed, err: %v\n", err)
			return err
		}
		// 假設(shè)操作耗時(shí)5秒
		// 5秒內(nèi)我們通過其他的客戶端修改key，當(dāng)前事務(wù)就會(huì)失敗
		time.Sleep(5 * time.Second)
		// txpipeline 執(zhí)行事務(wù)中fn隊(duì)列中的命令。
		// 當(dāng)使用WATCH時(shí)，EXEC只會(huì)在被監(jiān)視的鍵沒有被修改的情況下執(zhí)行命令，從而允許檢查和設(shè)置機(jī)制。
		// Exec總是返回命令列表。如果事務(wù)失敗，則返回TxFailedErr。否則Exec返回第一個(gè)失敗命令的錯(cuò)誤或nil
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			// 業(yè)務(wù)邏輯 如果 Key 沒有變化，則在原來的基礎(chǔ)上加 1
			pipe.Set(ctx, key, n+1, time.Hour)
			return nil
		})
		return err
	}, key)
}

func main() {
	if err := initRedisClient(); err != nil {
		fmt.Printf("initRedisClient failed: %v\n", err)
		return
	}
	fmt.Println("initRedisClient started successfully")
	defer rdb.Close() // Close 關(guān)閉客戶端，釋放所有打開的資源。關(guān)閉客戶端是很少見的，因?yàn)榭蛻舳耸情L(zhǎng)期存在的，并在許多例程之間共享。


	err := watchKeyDemo("watch_key")
	if err != nil {
		fmt.Printf("watchKeyDemo failed: %v\n", err)
		return
	}
	fmt.Printf("watchKeyDemo succeeded!\n")
}

運(yùn)行

Code/go/redis_demo via ?? v1.20.3 via ?? base 
? go run main.go
initRedisClient started successfully
watchKeyDemo succeeded!

Code/go/redis_demo via ?? v1.20.3 via ?? base took 6.5s 
? go run main.go
initRedisClient started successfully
watchKeyDemo failed: redis: transaction failed

Code/go/redis_demo via ?? v1.20.3 via ?? base took 6.2s 
? 

Redis 操作文章來源地址http://www.zghlxwxcb.cn/news/detail-486505.html

27.0.0.1:6379> get watch_key
(nil)
127.0.0.1:6379> set watch_key 9
OK
127.0.0.1:6379> get watch_key
"9"
127.0.0.1:6379> get watch_key
"10"
127.0.0.1:6379> set watch_key 99
OK
127.0.0.1:6379>

到了這里，關(guān)于Go Redis 管道和事務(wù)之 go-redis的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！