深入分析Go 实现 MySQL 数据库事务

START TRANSACTION;

COMMIT;

ROLLBACK;

START TRANSACTION;
-- 在事务中执行一系列数据库操作
INSERT INTO users (name, age) VALUES ('Alice', 25);
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
DELETE FROM logs WHERE user_id = 1;
-- 如果一切正常，提交事务
COMMIT;

START TRANSACTION;

-- 在事务中执行一系列数据库操作
INSERT INTO users (name, age) VALUES ('Bob', 30);
UPDATE accounts SET balance = balance - 200 WHERE user_id = 2;

-- 发生错误或异常，回滚事务
ROLLBACK;

CREATE TABLE mytable (
  id INT PRIMARY KEY,
  name VARCHAR(50)
) ENGINE=InnoDB;

// BeginTx starts a transaction.
//
// The provided context is used until the transaction is committed or rolled back.
// If the context is canceled, the sql package will roll back
// the transaction. Tx.Commit will return an error if the context provided to
// BeginTx is canceled.
//
// The provided TxOptions is optional and may be nil if defaults should be used.
// If a non-default isolation level is used that the driver doesn't support,
// an error will be returned.
func (db *DB) BeginTx(ctx context.Context, opts *TxOptions) (*Tx, error) {
	var tx *Tx
	var err error
	err = db.retry(func(strategy connReuseStrategy) error {
		tx, err = db.begin(ctx, opts, strategy)
		return err
	})
	return tx, err
}
// Begin starts a transaction. The default isolation level is dependent on
// the driver.
//
// Begin uses context.Background internally; to specify the context, use
// BeginTx.
func (db *DB) Begin() (*Tx, error) {
	return db.BeginTx(context.Background(), nil)
}

// rollback aborts the transaction and optionally forces the pool to discard
// the connection.
func (tx *Tx) rollback(discardConn bool) error {
	if !tx.done.CompareAndSwap(false, true) {
		return ErrTxDone
	}
	if rollbackHook != nil {
		rollbackHook()
	}
	// Cancel the Tx to release any active R-closemu locks.
	// This is safe to do because tx.done has already transitioned
	// from 0 to 1. Hold the W-closemu lock prior to rollback
	// to ensure no other connection has an active query.
	tx.cancel()
	tx.closemu.Lock()
	tx.closemu.Unlock()
	var err error
	withLock(tx.dc, func() {
		err = tx.txi.Rollback()
	})
	if !errors.Is(err, driver.ErrBadConn) {
		tx.closePrepared()
	}
	if discardConn {
		err = driver.ErrBadConn
	}
	tx.close(err)
	return err
}
// Rollback aborts the transaction.
func (tx *Tx) Rollback() error {
	return tx.rollback(false)
}

// Commit commits the transaction.
func (tx *Tx) Commit() error {
	// Check context first to avoid transaction leak.
	// If put it behind tx.done CompareAndSwap statement, we can't ensure
	// the consistency between tx.done and the real COMMIT operation.
	select {
	default:
	case <-tx.ctx.Done():
		if tx.done.Load() {
			return ErrTxDone
		}
		return tx.ctx.Err()
	}
	if !tx.done.CompareAndSwap(false, true) {
		return ErrTxDone
	}
	// Cancel the Tx to release any active R-closemu locks.
	// This is safe to do because tx.done has already transitioned
	// from 0 to 1. Hold the W-closemu lock prior to rollback
	// to ensure no other connection has an active query.
	tx.cancel()
	tx.closemu.Lock()
	tx.closemu.Unlock()
	var err error
	withLock(tx.dc, func() {
		err = tx.txi.Commit()
	})
	if !errors.Is(err, driver.ErrBadConn) {
		tx.closePrepared()
	}
	tx.close(err)
	return err
}

package main

import (
	"database/sql"
	"fmt"
	"time"

	_ "github.com/go-sql-driver/mysql" // 匿名导入 自动执行 init()
)

var db *sql.DB

func initMySQL() (err error) {
	//DSN (Data Source Name)
	dsn := "root:12345678@tcp(127.0.0.1:3306)/sql_test"
	// 注意：要初始化全局的 db 对象，不要新声明一个 db 变量
	db, err = sql.Open("mysql", dsn) // 只对格式进行校验，并不会真正连接数据库
	if err != nil {
		return err
	}

	// Ping 验证与数据库的连接是否仍处于活动状态，并在必要时建立连接。
	err = db.Ping()
	if err != nil {
		fmt.Printf("connect to db failed, err: %v\n", err)
		return err
	}
	// 数值需要根据业务具体情况来确定
	db.SetConnMaxLifetime(time.Second * 10) // 设置可以重用连接的最长时间
	db.SetConnMaxIdleTime(time.Second * 5)  // 设置连接可能处于空闲状态的最长时间
	db.SetMaxOpenConns(200)                 // 设置与数据库的最大打开连接数
	db.SetMaxIdleConns(10)                  //  设置空闲连接池中的最大连接数
	return nil
}

type user struct {
	id   int
	age  int
	name string
}

// 事务操作
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=? WHERE id=?"
	ret1, err := tx.Exec(sqlStr1, 22, 2)
	if err != nil {
		tx.Rollback() // 回滚
		fmt.Printf("exec sql1 failed, err: %v\n", err)
		return
	}
	// RowsAffected 返回受更新、插入或删除影响的行数。并非每个数据库或数据库驱动程序都支持此功能。
	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=? WHERE id=?"
	ret2, err := tx.Exec(sqlStr2, 100, 5)
	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 ret2.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!")
}

func main() {
	if err := initMySQL(); err != nil {
		fmt.Printf("connect to db failed, err: %v\n", err)
	}
	// 检查完错误之后执行，确保 db 不为 nil
	// Close() 用来释放数据库连接相关的资源
	// Close 将关闭数据库并阻止启动新查询。关闭，然后等待服务器上已开始处理的所有查询完成。
	defer db.Close()

	fmt.Println("connect to database success")
	// db.xx() 去使用数据库操作...

	//	事务
	transactionDemo()
}

Code/go/mysql_demo via 🐹 v1.20.3 via 🅒 base 
➜ go run main.go
connect to database success
1 0
事务回滚...
exec trans success!

Code/go/mysql_demo via 🐹 v1.20.3 via 🅒 base 
➜ go run main.go
connect to database success
1 1
事务提交...
exec trans success!

Code/go/mysql_demo via 🐹 v1.20.3 via 🅒 base 
➜ 

mysql> select * from user;
+----+--------+------+
| id | name   | age  |
+----+--------+------+
|  1 | 小乔   |   16 |
|  2 | 小乔   |   12 |
|  5 | 昭君   |   12 |
|  6 | 黛玉   |   16 |
+----+--------+------+
4 rows in set (0.00 sec)

mysql> select * from user;  # 第一次执行后查询
+----+--------+------+
| id | name   | age  |
+----+--------+------+
|  1 | 小乔   |   16 |
|  2 | 小乔   |   12 |
|  5 | 昭君   |   12 |
|  6 | 黛玉   |   16 |
+----+--------+------+
4 rows in set (0.00 sec)

mysql> select * from user;   # 第二次执行后查询
+----+--------+------+
| id | name   | age  |
+----+--------+------+
|  1 | 小乔   |   16 |
|  2 | 小乔   |   22 |
|  5 | 昭君   |  100 |
|  6 | 黛玉   |   16 |
+----+--------+------+
4 rows in set (0.00 sec)

mysql>