go-redis 框架基本使用

redis使用场景

• 缓存系统，减轻主数据库（MySQL）的压力。
• 计数场景，比如微博、抖音中的关注数和粉丝数。
• 热门排行榜，需要排序的场景特别适合使用ZSET。
• 利用 LIST 可以实现队列的功能。
• 利用 HyperLogLog 统计UV、PV等数据。
• 使用 geospatial index 进行地理位置相关查询。

下载框架和连接redis

Go 社区中目前有很多成熟的 redis client 库，比如redigo和go-redis，读者可以自行选择适合自己的库。本文章使用 go-redis 这个库来操作 Redis 数据库。

1. 安装go-redis

# redis 6
go get github.com/go-redis/redis/v8
# redis 7
go get github.com/go-redis/redis/v9

2. 连接redis

var Rdb *redis.Clientfunc Connect() {Rdb = redis.NewClient(&redis.Options{Addr:     "localhost:6379",Password: "",DB:       0,PoolSize: 10,})
}

字符串操作

只要Redis命令足够熟悉，那么对于这个框架的API的学习基本就没有什么问题。由于Redis命令太多，在此只列出了字符串和有序集合这两种数据类型的操作示例。

func String() {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()//set命令_, err := connect.Rdb.Set(ctx, "name", "bing", 0).Result()if err != nil {fmt.Println(err.Error())}name, err := connect.Rdb.Get(ctx, "name").Result()fmt.Println(name)//GetSet命令v1, _ := connect.Rdb.GetSet(ctx, "name", "xyz").Result()fmt.Println("旧值: " + v1) //bingname, err = connect.Rdb.Get(ctx, "name").Result()fmt.Println("新值: " + name) //xyz//MSet和MGet命令connect.Rdb.MSet(ctx, "age", 18, "password", "1234")v2 := connect.Rdb.MGet(ctx, "name", "age", "password").Val()for _, v := range v2 {fmt.Println(v)}//IncrBy命令v3 := connect.Rdb.IncrBy(ctx, "age", 2).Val() //20fmt.Println(v3)//append命令connect.Rdb.Append(ctx, "password", "abc")v4 := connect.Rdb.Get(ctx, "password").Val() //1234abcfmt.Println(v4)//SetRange命令connect.Rdb.SetRange(ctx, "password", 0, "987654")v5 := connect.Rdb.Get(ctx, "password").Val() //987654cfmt.Println(v5)//GetRange命令v6 := connect.Rdb.GetRange(ctx, "password", 4, -1).Val() //54cfmt.Println(v6)v7 := connect.Rdb.Get(ctx, "password").Val() //987654cfmt.Println(v7)//StrLen命令v8 := connect.Rdb.StrLen(ctx, "name").Val() //3fmt.Println(v8)//获取编码方式v9 := connect.Rdb.ObjectEncoding(ctx, "age").Val() //intfmt.Println(v9)//redis.Nil的用法v10, err := connect.Rdb.Get(ctx, "no_existing").Result()if redis.Nil == err {fmt.Println("key不存在")} else if err != nil {fmt.Println(err.Error())} else {fmt.Println(v10)}
}

有序集合操作

func ZSet() {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()ZSetKey := "languages"languages := []redis.Z{{Score: 90, Member: "Go"},{Score: 85, Member: "Python"},{Score: 99, Member: "C"},{Score: 95, Member: "Java"},{Score: 99, Member: "Rust"},{Score: 80, Member: "PHP"},}err := connect.Rdb.ZAdd(ctx, ZSetKey, languages...).Err()if err != nil {fmt.Println(err.Error())}//按照分数从低到高遍历v1 := connect.Rdb.ZRange(ctx, ZSetKey, 0, -1).Val()fmt.Println(v1) //[PHP Python Go Java C Rust]v2 := connect.Rdb.ZRangeWithScores(ctx, ZSetKey, 0, -1).Val()fmt.Println(v2) //[{80 PHP} {85 Python} {90 Go} {95 Java} {99 C} {99 Rust}]opt1 := &redis.ZRangeBy{Min:    "0",  //查询的最小分数值Max:    "95", //查询的最大分数值Offset: 0,    //查询的起始位置Count:  6,    //需要查询的元素个数}v3 := connect.Rdb.ZRangeByScoreWithScores(ctx, ZSetKey, opt1).Val()fmt.Println(v3) //[{80 PHP} {85 Python} {90 Go} {95 Java}]opt2 := &redis.ZRangeBy{Min:    "[K", //查询的最小字典序值Max:    "[X", //查询的最大字典序值Offset: 0,    //查询的起始位置Count:  5,    //需要查询的元素个数}v4 := connect.Rdb.ZRangeByLex(ctx, ZSetKey, opt2).Val()fmt.Println(v4) //[PHP Python Go Java C]v5 := connect.Rdb.ZCard(ctx, ZSetKey).Val()fmt.Println("集合长度: " + strconv.FormatInt(v5, 10)) // 6
}

流水线

使用流水线就是将多个执行的命令放入 pipeline 中，然后使用1次读写操作就像执行单个命令一样执行它们，就相当于把多个命令打包，然后一起发送给redis服务器，让redis服务器一次性执行完毕。这样做的好处是节省了执行命令的网络往返时间（RTT）。

注意：如果redis采用了分布式集群模式，不可以直接使用pipeline命令进行操作，因为访问的key可能并不在同一个节点上。

下面的示例代码中演示了使用 pipeline 将pipeline_counter键的值加1和设置过期时间。

func PipeLine() {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()//创建一个Pipeline对象:pipepipe := connect.Rdb.Pipeline()//将名为"pipeline_counter"的键的值加1incr := pipe.Incr(ctx, "pipeline_counter")//设置"pipeline_counter"键的过期时间为1分钟pipe.Expire(ctx, "pipeline_counter", time.Minute)//执行所有的命令。_, err := pipe.Exec(ctx)if err != nil {panic(err)}// 在执行pipe.Exec之后才能获取到结果fmt.Println(incr.Val())
}

上面的代码相当于将以下两个redis命令一次发给 Redis Server 端执行，与不使用 Pipeline 相比能减少一次RTT。

INCR pipeline_counter
EXPIRE pipeline_counts 60

或者，你也可以使用Pipelined 方法，它会在当前函数退出时调用 Exec。

func PipeLine() {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()var incr *redis.IntCmdcmdS, err := connect.Rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {incr = pipe.Incr(ctx, "pipelined_counter")pipe.Expire(ctx, "pipelined_counter", time.Minute)return nil})if err != nil {panic(err)}// 在pipeline执行后获取到结果fmt.Println(incr.Val())//使用类型断言特性来对 cmd 进行类型检查for _, cmd := range cmdS {switch v := cmd.(type) {case *redis.StringCmd:fmt.Println(v.Val())case *redis.IntCmd:fmt.Println(v.Val())case *redis.BoolCmd:fmt.Println(v.Val())default:fmt.Printf("unexpected type %T\n", v)}}
}

运行结果如下：

所以，在那些我们需要一次性执行多个命令的场景下，就可以考虑使用 pipeline 来优化。

事务

1. 普通事务

Redis 是单线程执行命令的，因此单个命令始终是原子的，但是来自不同客户端的两个给定命令可以依次执行，例如在它们之间交替执行。使用事务后，Redis会按照命令的顺序执行这些命令，并且在执行过程中不会立即返回结果，只有在所有命令都执行完毕后，才会一次性返回所有命令的执行结果。也就是在执行过程中保证了原子性，即要么所有命令都执行成功，要么所有命令都不执行。

同时，Redis事务还支持WATCH命令，可以在事务执行之前监视一个或多个键，如果在事务执行期间这些键发生了改变，事务会被中断。这样可以确保在执行事务期间，被监视的键没有被其他客户端修改。

"Tx"是"Transaction"的缩写，意为"事务”。TxPipeline 和 TxPipelined 的使用方法如下所示：

func Work() {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()pipe := connect.Rdb.TxPipeline()incr := pipe.Incr(ctx, "tx_pipeline_counter")pipe.Expire(ctx, "tx_pipeline_counter", time.Minute)_, err := pipe.Exec(ctx)fmt.Println(incr.Val(), err)var incr2 *redis.IntCmd_, err = connect.Rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {incr2 = pipe.Incr(ctx, "tx_pipeline_counter")pipe.Expire(ctx, "tx_pipeline_counter", time.Minute)return nil})fmt.Println(incr2.Val(), err)
}

运行结果如下：

2. Watch

我们通常搭配 WATCH命令来执行事务操作。从使用WATCH命令监视某个 key 开始，直到执行EXEC命令的这段时间里，如果有其他用户抢先对被监视的 key 进行了替换、更新、删除等操作，那么当用户尝试执行EXEC的时候，事务将失败并返回一个错误，用户可以根据这个错误选择重试事务或者放弃事务。

Watch方法接收一个函数和一个或多个key作为参数。

Watch(fn func(*Tx) error, keys ...string) error

假设我们有一个应用程序，它需要保持用户的积分。我们需要一个函数，可以安全地减少用户的积分。为了避免并发问题，我们将使用WATCH命令来监视用户的积分，并在事务中更新积分。

func WatchUserPoints(userID string, points int) error {ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)defer cancel()for {// 监控err := connect.Rdb.Watch(ctx, func(tx *redis.Tx) error {// 得到当前用户的积分nn, err := tx.Get(ctx, userID).Int()//扣除积分时开启事务,points表示要扣除的积分_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {err := pipe.Set(ctx, userID, n-points, 0).Err()return err})return err}, userID) //监控的键为userID,也就是当这个键的值(积分)如果在事务执行过程中被其他客户端修改,那么当前事务就会执行失败。//对错误的判断if err == redis.TxFailedErr {//表示监视的键在事务执行过程中被其他客户端修改了，因此事务执行失败了。continue} else if err != nil {//其他类型的错误return err} else {//没有错误break}}//能够跳出循环说明一切正常return nil
}

这段代码的目的是监视用户的当前积分，如果在事务执行过程中，其他客户端改变了这个键的值（也就是用户的积分），那么 Watch 会发现这个变化并使得事务失败，返回 redis.TxFailedErr 错误。

总的来说，这段代码的目的是确保在减少用户积分的过程中，用户的积分没有被其他客户端修改。这是通过Redis的 WATCH 命令来实现的，这个命令可以将一个或多个键标记为监视，然后在执行事务之前检查这些键是否已经被修改。