go

Go操作常见数据库

Posted by Sutdown on February 18, 2025

Go操作MySQL

连接

Go语言中的database/sql包提供了保证SQL或类SQL数据库的泛用接口，并不提供具体的数据库驱动，使用该包时至少注入一个数据库驱动。

初始化连接：Ping()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

go get -u github.com/go-sql-driver/mysql 				  // 下载依赖
func Open(driverName, dataSourceName string) (*DB, error) // 打开指定数据库，指定数据源

void db *sql.DB

func initDB() (err error) {
     // 格式校验
    dsn := "root:root1234@tcp(127.0.0.1:13306)/sql_demo"
    db, err = sql.Open("mysql", dsn)
    if err!= nil {
        panic(err)
    }
    
    // 做完错误检查之后，确保db不为nil
    // CLose() 用来释放数据库连接相关的资源
    defer db.Close()
    
    // 初始化连接
    err = db.Ping()
    if err != nil {
        return err
    }
    
    db.SetConnMaxLifetime(time.Second*10)
    db.SetMaxOpenConns(200)
    db.SetMaxIdleConns(1)
	return nil
}

func main() {
    if err :=initDB(); err != nil {
        fmt.Printf("connect to db failed, err:%v\n", error)
        return
    }
      fmt.Println("connect to db success")
}

其中sql.DB是表示连接的数据库对象（结构体实例），它保存了连接数据库相关的所有信息。它内部维护着一个具有零到多个底层连接的连接池，它可以安全地被多个goroutine同时使用。

1
2
3
4
5

// SetMaxOpenConns设置与数据库建立连接的最大数目。 如果n大于0且小于最大闲置连接数，会将最大闲置连接数减小到匹配最大开启连接数的限制。 如果n<=0，不会限制最大开启连接数，默认为0（无限制）。
func (db *DB) SetMaxOpenConns(n int)

// SetMaxIdleConns设置连接池中的最大闲置连接数。 如果n大于最大开启连接数，则新的最大闲置连接数会减小到匹配最大开启连接数的限制。 如果n<=0，不会保留闲置连接。
func (db *DB) SetMaxIdleConns(n int)

CRUD

mySQL

CREATE DATABASE sql_test;
use sql_test;
CREATE TABLE `user` (
    `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
    `name` VARCHAR(20) DEFAULT '',
    `age` INT(11) DEFAULT '0',
    PRIMARY KEY(`id`)
)ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

单行查询

QueryRow会在连接池里调用连接，达到最大连接数后停止，scan()会关闭连接。

1

func (db *DB) QueryRow(query string, args ...interface{}) *Row

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

type user struct {
    id int
    age int
    name string
}

func queryROwDemo() {
    sqlStr := "select id, name, age from user where id=?"
    var u user
    row := db.QueryRow(sqlStr, 1)
    err := row.Scan(&u.id, &u.name, &u.age) // 关闭连接
    if err!=nil {
        fmt.Printf("scan failed, err:%v\n", err)
        return
    }
}

多行查询

1

func (db *DB) Query(query string, args ...interface{}) (*Rows, error)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

// 查询多条数据示例
func queryMultiRowDemo() {
	sqlStr := "select id, name, age from user where id > ?"
	rows, err := db.Query(sqlStr, 0)
	if err != nil {
		fmt.Printf("query failed, err:%v\n", err)
		return
	}
	// 非常重要：关闭rows释放持有的数据库链接
	defer rows.Close()

	// 循环读取结果集中的数据
	for rows.Next() {
		var u user
		err := rows.Scan(&u.id, &u.name, &u.age)
		if err != nil {
			fmt.Printf("scan failed, err:%v\n", err)
			return
		}
		fmt.Printf("id:%d name:%s age:%d\n", u.id, u.name, u.age)
	}
}

插入数据，更新数据

1
2
3
4

func (db *DB) Exec(query string, args ...interface{}) (Result, error)

LastInsertId()
RowsAffected()

MySQL预处理

什么是预处理

普通SQL语句执行过程：

  1. 客户端对SQL语句进行占位符替换得到完整的SQL语句。
  2. 客户端发送完整SQL语句到MySQL服务端
  3. MySQL服务端执行完整的SQL语句并将结果返回给客户端。

预处理执行过程：

  1. 把SQL语句分成两部分，命令部分与数据部分。
  2. 先把命令部分发送给MySQL服务端，MySQL服务端进行SQL预处理。
  3. 然后把数据部分发送给MySQL服务端，MySQL服务端对SQL语句进行占位符替换。
  4. MySQL服务端执行完整的SQL语句并将结果返回给客户端。

为什么要预处理？

  1. 优化MySQL服务器重复执行SQL的方法，可以提升服务器性能，提前让服务器编译，一次编译多次执行，节省后续编译的成本。
  2. 避免SQL注入问题。

Go实现MySQL预处理

Prepare方法会先将sql语句发送给MySQL服务端，返回一个准备好的状态用于之后的查询和命令。返回值可以同时执行多个查询和命令。

1
2

// database/sql
func (db *DB) Prepare(query string) (*Stmt, error)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// 预处理查询示例
func prepareQueryDemo() {
	sqlStr := "select id, name, age from user where id > ?"
	stmt, err := db.Prepare(sqlStr)
	if err != nil {
		fmt.Printf("prepare failed, err:%v\n", err)
		return
	}
	defer stmt.Close()
	rows, err := stmt.Query(0)
	if err != nil {
		fmt.Printf("query failed, err:%v\n", err)
		return
	}
	defer rows.Close()
	// 遍历结果集的每一行
	for rows.Next() {
		var u user
    // Scan 方法用于从数据库查询结果中提取数据并存储到变量中。
		err := rows.Scan(&u.id, &u.name, &u.age)
		if err != nil {
			fmt.Printf("scan failed, err:%v\n", err)
			return
		}
		fmt.Printf("id:%d name:%s age:%d\n", u.id, u.name, u.age)
	}
}

SQL注入

我们任何时候都不应该自己拼接SQL语句！

Go实现MySQL事务

在MySQL中只有使用了Innodb数据库引擎的数据库或表才支持事务。事务处理可以用来维护数据库的完整性，保证成批的SQL语句要么全部执行，要么全部不执行。事务必须满足4个条件（ACID）：原子性（Atomicity，或称不可分割性）、一致性（Consistency）、隔离性（Isolation，又称独立性）、持久性（Durability）。

事务相关方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

func (db *DB) Begin() (*Tx, error) 	// 开始事务
func (tx *Tx) Commit() error				// 提交事务
func (tx *Tx) Rollback() error			// 回滚事务

// 实例
// 事务操作示例
func transactionDemo() {
	tx, err := db.Begin() // 开启事务
	if err != nil {
		if tx != nil {
			tx.Rollback() 		// 回滚
		}
		fmt.Printf("begin trans failed, err:%v\n", err)
		return
	}
	sqlStr1 := "Update user set age=30 where id=?"
	ret1, err := tx.Exec(sqlStr1, 2)
	if err != nil {
		tx.Rollback() // 回滚
		fmt.Printf("exec sql1 failed, err:%v\n", err)
		return
	}
	affRow1, err := ret1.RowsAffected()
	if err != nil {
		tx.Rollback() // 回滚
		fmt.Printf("exec ret1.RowsAffected() failed, err:%v\n", err)
		return
	}

	sqlStr2 := "Update user set age=40 where id=?"
	ret2, err := tx.Exec(sqlStr2, 3)
	if err != nil {
		tx.Rollback() // 回滚
		fmt.Printf("exec sql2 failed, err:%v\n", err)
		return
	}
	affRow2, err := ret2.RowsAffected()
	if err != nil {
		tx.Rollback() // 回滚
		fmt.Printf("exec ret1.RowsAffected() failed, err:%v\n", err)
		return
	}

	fmt.Println(affRow1, affRow2)
	if affRow1 == 1 && affRow2 == 1 {
		fmt.Println("事务提交啦...")
		tx.Commit() // 提交事务
	} else {
		tx.Rollback()
		fmt.Println("事务回滚啦...")
	}

	fmt.Println("exec trans success!")
}

sqlx库使用指南

连接

1

go get github.com/jmoiron/sqlx # 下载sqlx依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import (
	"fmt"
	_ "github.com/go-sql-driver/mysql"  // 不要忘了导入数据库驱动
	"github.com/jmoiron/sqlx"
)

var db *sqlx.DB

func initDB() (err error) {
	dsn := "user:password@tcp(127.0.0.1:3306)/sql_test?charset=utf8mb4&parseTime=True"
	// 也可以使用MustConnect连接不成功就panic
	db, err = sqlx.Connect("mysql", dsn)
	if err != nil {
		fmt.Printf("connect DB failed, err:%v\n", err)
		return
	}
	db.SetMaxOpenConns(20)
	db.SetMaxIdleConns(10)
	return
}

CRUD

查询

1
2

err := db.Get(&u, sqlStr, 1)				  // select id, name, age from user where id=?
err := db.Select(&users, sqlStr, 0)	  // select id, name, age from user where id > ?

插入，更新和删除

1
2
3
4
5
6
7
8
9

/* Exec：用于执行不返回行数据的SQL语句，例如INSERT、UPDATE和DELETE。 */
ret, err := db.Exec(sqlStr, "沙河小王子", 19) 	  // insert into user(name, age) values (?,?)
ret, err := db.Exec(sqlStr, 39, 6)			 			 // update user set age=? where id = ?
ret, err := db.Exec(sqlStr, 6)								 // delete from user where id = ?

/* 用于执行SQL语句，并使用命名参数（结构体或map）替代?占位符 */
// DB.NamedExec方法用来绑定SQL语句与结构体或map中的同名字段。
_, err = db.NamedExec(sqlStr, map[string]interface{}{ "name": "七米", "age": 28, })	// INSERT、UPDATE、DELETE 
rows, err := db.NamedQuery(sqlStr, map[string]interface{}{ "name": "七米" })	// select

事务操作

可以使用sqlx中提供的db.Beginx()和tx.Exec()方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

func transactionDemo2()(err error) {
	tx, err := db.Beginx() // 开启事务
	if err != nil {
		fmt.Printf("begin trans failed, err:%v\n", err)
		return err
	}
  
	defer func() {
		if p := recover(); p != nil {
			tx.Rollback() // 捕获到panic，回滚
			panic(p) 			// re-throw panic after Rollback
		} else if err != nil {
			fmt.Println("rollback")	// 发生一个预期的错误，回滚
			tx.Rollback() 					// err is non-nil; don't change it
		} else {
			err = tx.Commit() 		// err is nil; if Commit returns error update err
			fmt.Println("commit") // 没有任何错误
		}
	}()

	sqlStr1 := "Update user set age=20 where id=?"
	rs, err := tx.Exec(sqlStr1, 1)
	if err!= nil{
		return err
	}
	n, err := rs.RowsAffected()
	if err != nil {
		return err
	}
	if n != 1 {
		return errors.New("exec sqlStr1 failed")
	}
  
	sqlStr2 := "Update user set age=50 where i=?"
	rs, err = tx.Exec(sqlStr2, 5)
	if err!=nil{
		return err
	}
	n, err = rs.RowsAffected()
	if err != nil {
		return err
	}
	if n != 1 {
		return errors.New("exec sqlStr1 failed")
	}
  
	return err
}

sqlx.In的批量插入实例

bindvars（绑定变量）

查询占位符?在内部称为bindvars（查询占位符）,它非常重要。你应该始终使用它们向数据库发送值，因为它们可以防止SQL注入攻击。database/sql不尝试对查询文本进行任何验证；它与编码的参数一起按原样发送到服务器。除非驱动程序实现一个特殊的接口，否则在执行之前，查询是在服务器上准备的。因此bindvars是特定于数据库的:

  • MySQL中使用?
  • PostgreSQL使用枚举的$1、$2等bindvar语法
  • SQLite中?和$1的语法都支持
  • Oracle中使用:name的语法

bindvars的一个常见误解是，它们用来在sql语句中插入值。它们其实仅用于参数化，不允许更改SQL语句的结构。

1
2
3
4
5

// ？不能用来插入表名（做SQL语句中表名的占位符）
db.Query("SELECT * FROM ?", "mytable")
 
// ？也不能用来插入列名（做SQL语句中列名的占位符）
db.Query("SELECT ?, ? FROM people", "name", "location")

自己拼接语句实现批量插入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

// BatchInsertUsers 自行构造批量插入的语句
func BatchInsertUsers(users []*User) error {
	// 存放 (?, ?) 的slice
	valueStrings := make([]string, 0, len(users))
	// 存放values的slice
	valueArgs := make([]interface{}, 0, len(users) * 2)
	// 遍历users准备相关数据
	for _, u := range users {
		// 此处占位符要与插入值的个数对应
		valueStrings = append(valueStrings, "(?, ?)")
		valueArgs = append(valueArgs, u.Name)
		valueArgs = append(valueArgs, u.Age)
	}
	// 自行拼接要执行的具体语句
	stmt := fmt.Sprintf("INSERT INTO user (name, age) VALUES %s",
		strings.Join(valueStrings, ","))
	_, err := DB.Exec(stmt, valueArgs...)
	return err
}

使用sqlx.In实现批量插入

结构体实现driver.Valuer接口，参数通常为切片

  • NamedExec 更适合处理结构化的输入（如结构体或命名字段的 map），其优势在于能自动匹配结构体字段与 SQL 语句中的命名参数。映射更清晰，但是结构体也更复杂。
  • sqlx.In 则更适合处理位置参数，适合用于将一组数据传递给 SQL 语句，并自动生成批量插入的 SQL 语句。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

func (u User) Value() (driver.Value, error) {
	return []interface{}{u.Name, u.Age}, nil
}

// BatchInsertUsers2 使用sqlx.In帮我们拼接语句和参数, 注意传入的参数是[]interface{}
func BatchInsertUsers2(users []interface{}) error {
	query, args, _ := sqlx.In(
		"INSERT INTO user (name, age) VALUES (?), (?), (?)",
		users..., // 如果arg实现了 driver.Valuer, sqlx.In 会通过调用 Value()来展开它
	)
	fmt.Println(query) // 查看生成的querystring
	fmt.Println(args)  // 查看生成的args
	_, err := DB.Exec(query, args...) // 执行查询
	return err
}

使用NameExec实现批量插入

1
2
3
4
5
6

// BatchInsertUsers3 使用NamedExec实现批量插入
// 参数通常是结构体或者map
func BatchInsertUsers3(users []*User) error {
	_, err := DB.NamedExec("INSERT INTO user (name, age) VALUES (:name, :age)", users)
	return err
}

sqlx.In的查询示例

关于sqlx.In这里再补充一个用法，在sqlx查询语句中实现In查询和FIND_IN_SET函数。即实现SELECT * FROM user WHERE id in (3, 2, 1);和SELECT * FROM user WHERE id in (3, 2, 1) ORDER BY FIND_IN_SET(id, '3,2,1');。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// in查询 查询id在给定id集合中的数据。
// QueryByIDs 根据给定ID查询
func QueryByIDs(ids []int)(users []User, err error){
	// 动态填充id
	query, args, err := sqlx.In("SELECT name, age FROM user WHERE id IN (?)", ids)
	if err != nil {
		return
	}
	// sqlx.In 返回带 `?` bindvar的查询语句, 我们使用Rebind()重新绑定。
	// 重新生成对应数据库的查询语句（如PostgreSQL 用 `$1`, `$2` bindvar）
	query = DB.Rebind(query)

	err = DB.Select(&users, query, args...)
	return
}

// in查询和FIND_IN_SET函数 查询id在给定id集合的数据并维持给定id集合的顺序。
// QueryAndOrderByIDs 按照指定id查询并维护顺序
func QueryAndOrderByIDs(ids []int)(users []User, err error){
	// 动态填充id
	strIDs := make([]string, 0, len(ids))
	for _, id := range ids {
		strIDs = append(strIDs, fmt.Sprintf("%d", id))
	}
	query, args, err := sqlx.In("SELECT name, age FROM user WHERE id IN (?) ORDER BY FIND_IN_SET(id, ?)", ids, strings.Join(strIDs, ","))
	if err != nil {
		return
	}

	// sqlx.In 返回带 `?` bindvar的查询语句, 我们使用Rebind()重新绑定它
	query = DB.Rebind(query)

	err = DB.Select(&users, query, args...)
	return
}

go-redis

redis介绍

Redis（Remote Dictionary Server）是一个开源的内存数据库，支持键值（Key-Value）存储，可以用作缓存、消息队列、分布式锁等。它的特点是高性能、支持多种数据结构、持久化、分布式，适用于各种高并发场景。

1
2

docker run --name redis507 -p 6379:6379 -d redis:5.0.7 	 // 名为 redis507 的 5.0.7 版本的 redis server环境。
docker run -it --network host --rm redis:5.0.7 redis-cli // 启动一个 redis-cli 连接上面的 redis server。

go-redis库

go-redis 这个库来操作 Redis 数据库。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

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

// 普通连接
rdb := redis.NewClient(&redis.Options{
	Addr:     "localhost:6379",
	Password: "", // 密码
	DB:       0,  // 数据库
	PoolSize: 20, // 连接池大小
})

// 解析数据源字符串
opt, err := redis.ParseURL("redis://<user>:<pass>@localhost:6379/<db>")
if err != nil {
	panic(err)
}

rdb := redis.NewClient(opt)

// TLS连接
rdb := redis.NewClient(&redis.Options{
	TLSConfig: &tls.Config{
		MinVersion: tls.VersionTLS12,
		// Certificates: []tls.Certificate{cert},
    // ServerName: "your.domain.com",
	},
})

// Redis Sentinel模式
rdb := redis.NewFailoverClient(&redis.FailoverOptions{
    MasterName:    "master-name",
    SentinelAddrs: []string{":9126", ":9127", ":9128"},
})
// Redis Cluster模式
rdb := redis.NewClusterClient(&redis.ClusterOptions{
    Addrs: []string{":7000", ":7001", ":7002", ":7003", ":7004", ":7005"},

    // 若要根据延迟或随机路由命令，请启用以下命令之一
    // RouteByLatency: true,
    // RouteRandomly: true,
})

基本操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

// doCommand go-redis基本使用示例
func doCommand() {
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// 执行命令获取结果
	val, err := rdb.Get(ctx, "key").Result()
	fmt.Println(val, err)

	// 先获取到命令对象
	cmder := rdb.Get(ctx, "key")
	fmt.Println(cmder.Val()) // 获取值
	fmt.Println(cmder.Err()) // 获取错误

	// 直接执行命令获取错误
	err = rdb.Set(ctx, "key", 10, time.Hour).Err()

	// 直接执行命令获取值
	value := rdb.Get(ctx, "key").Val()
	fmt.Println(value)
}

go-redis 还提供了一个执行任意命令或自定义命令的 Do 方法，特别是一些 go-redis 库暂时不支持的命令都可以使用该方法执行。具体使用方法如下。

1
2
3
4
5
6
7
8
9
10
11
12
13

// doDemo rdb.Do 方法使用示例
func doDemo() {
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// 直接执行命令获取错误
	err := rdb.Do(ctx, "set", "key", 10, "EX", 3600).Err()
	fmt.Println(err)

	// 执行命令获取结果
	val, err := rdb.Do(ctx, "get", "key").Result()
	fmt.Println(val, err)
}

扫描or遍历所有key

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

vals, err := rdb.Keys(ctx, "user:*").Result()

// 将redis中所有以prefix:为前缀的key都扫描出来
keys, cursor, err = rdb.Scan(ctx, cursor, "prefix:*", 0).Result()

// 但是如果需要扫描数百万的 key ，那速度就会比较慢。这种场景下你可以使用Scan命令来遍历所有符合要求的 key。
keys, cursor, err = rdb.Scan(ctx, cursor, "prefix:*", 0).Result()
if err != nil {
			panic(err)
}

// 针对这种需要遍历大量key的场景，go-redis中提供了一个简化方法——Iterator
// delKeysByMatch 按match格式扫描所有key并删除
func delKeysByMatch(match string, timeout time.Duration) {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	iter := rdb.Scan(ctx, 0, match, 0).Iterator()
	for iter.Next(ctx) {
		err := rdb.Del(ctx, iter.Val()).Err()
		if err != nil {
			panic(err)
		}
	}
	if err := iter.Err(); err != nil {
		panic(err)
	}
}

Redis Pipeline

Redis Pipeline 允许通过使用单个 client-server-client 往返执行多个命令来提高性能。区别于一个接一个地执行100个命令，你可以将这些命令放入 pipeline 中，然后使用1次读写操作像执行单个命令一样执行它们。这样做的好处是节省了执行命令的网络往返时间（RTT）。

pipeline和exec的区别是什么

Pipeline：用于将多个命令打包并发送到 Redis，但并不立即执行命令。它是将命令排队的过程。

Exec：用于执行 pipeline 中的命令，最终将这些命令发送到 Redis 服务器，并等待其执行结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

/* exec */
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)
}

// 在执行pipe.Exec之后才能获取到结果
fmt.Println(incr.Val())


/* pipeline */
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)
}

// 在pipeline执行后获取到结果
fmt.Println(incr.Val())

事务

Redis 是单线程执行命令的，因此单个命令始终是原子的，但是来自不同客户端的两个给定命令可以依次执行，例如在它们之间交替执行。但是，Multi/exec能够确保在multi/exec两个语句之间的命令之间没有其他客户端正在执行命令。

在这种场景我们需要使用 TxPipeline 或 TxPipelined 方法将 pipeline 命令使用 MULTI 和EXEC包裹起来。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// TxPipeline demo
pipe := rdb.TxPipeline()
incr := pipe.Incr(ctx, "tx_pipeline_counter")
pipe.Expire(ctx, "tx_pipeline_counter", time.Hour)
_, err := pipe.Exec(ctx)
fmt.Println(incr.Val(), err)

// TxPipelined demo
var incr2 *redis.IntCmd
_, err = rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
	incr2 = pipe.Incr(ctx, "tx_pipeline_counter")
	pipe.Expire(ctx, "tx_pipeline_counter", time.Hour)
	return nil
})
fmt.Println(incr2.Val(), err)

watch

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

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

// watchDemo 在key值不变的情况下将其值+1
func watchDemo(ctx context.Context, key string) error {
	return rdb.Watch(ctx, func(tx *redis.Tx) error {
		n, err := tx.Get(ctx, key).Int()
		if err != nil && err != redis.Nil {
			return err
		}
		// 假设操作耗时5秒
		// 5秒内我们通过其他的客户端修改key，当前事务就会失败
		time.Sleep(5 * time.Second)
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Set(ctx, key, n+1, time.Hour)
			return nil
		})
		return err
	}, key)
}

last，go-redis 官方文档中使用 GET 、SET和WATCH命令实现一个 INCR 命令的完整示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

// 此处rdb为初始化的redis连接客户端
const routineCount = 100

// 设置5秒超时
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

// increment 是一个自定义对key进行递增（+1）的函数
// 使用 GET + SET + WATCH 实现，类似 INCR
increment := func(key string) error {
	txf := func(tx *redis.Tx) error {
		// 获得当前值或零值
		n, err := tx.Get(ctx, key).Int()
		if err != nil && err != redis.Nil {
			return err
		}

		// 实际操作（乐观锁定中的本地操作）
		n++

		// 仅在监视的Key保持不变的情况下运行
		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
			// pipe 处理错误情况
			pipe.Set(ctx, key, n, 0)
			return nil
		})
		return err
	}

	// 最多重试100次
	for retries := routineCount; retries > 0; retries-- {
		err := rdb.Watch(ctx, txf, key)
		if err != redis.TxFailedErr {
			return err
		}
		// 乐观锁丢失
	}
	return errors.New("increment reached maximum number of retries")
}

// 开启100个goroutine并发调用increment
// 相当于对key执行100次递增
var wg sync.WaitGroup
wg.Add(routineCount)
for i := 0; i < routineCount; i++ {
	go func() {
		defer wg.Done()

		if err := increment("counter3"); err != nil {
			fmt.Println("increment error:", err)
		}
	}()
}
wg.Wait()

n, err := rdb.Get(ctx, "counter3").Int()
fmt.Println("最终结果：", n, err)
CATALOG
    FEATURED TAGS
    algorithm C++ levelDB 操作系统 Blog go mysql coroutine 计算机网络 recreation
    FRIENDS