当前位置：首页 > news >正文

Go语言数据库事务与并发控制

news 2026/5/24 21:27:23

Go语言数据库事务与并发控制

引言

数据库事务和并发控制是保证数据一致性和系统稳定性的关键技术。Go语言通过database/sql包提供了强大的事务支持。本文将深入探讨Go语言中数据库事务的实现原理、并发控制策略和最佳实践。

一、事务基础

1.1 事务ACID特性

func TransferMoney(db *sql.DB, fromID, toID int, amount float64) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // 扣除转出账户 _, err = tx.Exec( "UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID, ) if err != nil { return err } // 增加转入账户 _, err = tx.Exec( "UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID, ) if err != nil { return err } // 提交事务 return tx.Commit() }

1.2 事务隔离级别

func TransactionWithIsolation(db *sql.DB, level string) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // 设置隔离级别 _, err = tx.Exec(fmt.Sprintf("SET TRANSACTION ISOLATION LEVEL %s", level)) if err != nil { return err } // 执行事务操作... _, err = tx.Exec("INSERT INTO logs (message) VALUES ('transaction')") if err != nil { return err } return tx.Commit() }

二、并发控制机制

2.1 乐观锁

func UpdateWithOptimisticLock(db *sql.DB, id int, expectedVersion int, updates map[string]interface{}) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // 检查版本号 var currentVersion int err = tx.QueryRow("SELECT version FROM users WHERE id = ?", id).Scan(&currentVersion) if err != nil { return err } if currentVersion != expectedVersion { return fmt.Errorf("optimistic lock conflict: expected version %d, got %d", expectedVersion, currentVersion) } // 构建更新语句 setClause := "" args := []interface{}{} i := 1 for k, v := range updates { if i > 1 { setClause += ", " } setClause += fmt.Sprintf("%s = ?", k) args = append(args, v) i++ } setClause += ", version = version + 1" args = append(args, id) _, err = tx.Exec(fmt.Sprintf("UPDATE users SET %s WHERE id = ?", setClause), args...) if err != nil { return err } return tx.Commit() }

2.2 悲观锁

func UpdateWithPessimisticLock(db *sql.DB, id int, updates map[string]interface{}) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // 使用SELECT FOR UPDATE锁定行 var name string err = tx.QueryRow("SELECT name FROM users WHERE id = ? FOR UPDATE", id).Scan(&name) if err != nil { return err } // 执行更新 setClause := "" args := []interface{}{} i := 1 for k, v := range updates { if i > 1 { setClause += ", " } setClause += fmt.Sprintf("%s = ?", k) args = append(args, v) i++ } args = append(args, id) _, err = tx.Exec(fmt.Sprintf("UPDATE users SET %s WHERE id = ?", setClause), args...) if err != nil { return err } return tx.Commit() }

三、分布式事务

3.1 两阶段提交

type TwoPhaseCommit struct { participants []*sql.DB } func NewTwoPhaseCommit(dbs []*sql.DB) *TwoPhaseCommit { return &TwoPhaseCommit{participants: dbs} } func (t *TwoPhaseCommit) Execute(txOps []func(*sql.Tx) error) error { // Phase 1: Prepare transactions := make([]*sql.Tx, len(t.participants)) for i, db := range t.participants { tx, err := db.Begin() if err != nil { return err } transactions[i] = tx if err := txOps[i](tx); err != nil { // 回滚所有已开启的事务 for j := 0; j <= i; j++ { transactions[j].Rollback() } return err } } // Phase 2: Commit for i, tx := range transactions { if err := tx.Commit(); err != nil { // 部分提交失败，需要手动处理 for j := i + 1; j < len(transactions); j++ { transactions[j].Rollback() } return err } } return nil }

3.2 Saga模式

type SagaStep struct { Action func() error Compensate func() error } type Saga struct { steps []SagaStep } func NewSaga(steps []SagaStep) *Saga { return &Saga{steps: steps} } func (s *Saga) Execute() error { executedSteps := make([]int, 0) for i, step := range s.steps { if err := step.Action(); err != nil { // 执行补偿 for j := len(executedSteps) - 1; j >= 0; j-- { s.steps[executedSteps[j]].Compensate() } return err } executedSteps = append(executedSteps, i) } return nil } func TransferSaga(db *sql.DB, fromID, toID int, amount float64) error { saga := NewSaga([]SagaStep{ { Action: func() error { _, err := db.Exec("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID) return err }, Compensate: func() error { _, err := db.Exec("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, fromID) return err }, }, { Action: func() error { _, err := db.Exec("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID) return err }, Compensate: func() error { _, err := db.Exec("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, toID) return err }, }, }) return saga.Execute() }

四、事务优化策略

4.1 批量操作

func BatchInsert(db *sql.DB, items []Item) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() stmt, err := tx.Prepare("INSERT INTO items (name, value) VALUES (?, ?)") if err != nil { return err } defer stmt.Close() for _, item := range items { _, err := stmt.Exec(item.Name, item.Value) if err != nil { return err } } return tx.Commit() }

4.2 分块处理

func ProcessLargeDataset(db *sql.DB, batchSize int) error { offset := 0 for { rows, err := db.Query("SELECT id, data FROM large_table LIMIT ? OFFSET ?", batchSize, offset) if err != nil { return err } count := 0 for rows.Next() { count++ // 处理数据... } rows.Close() if count == 0 { break } offset += batchSize } return nil }

五、最佳实践

5.1 事务边界管理

func WithTransaction(db *sql.DB, fn func(*sql.Tx) error) error { tx, err := db.Begin() if err != nil { return err } defer func() { if r := recover(); r != nil { tx.Rollback() panic(r) } }() if err := fn(tx); err != nil { tx.Rollback() return err } return tx.Commit() } func UseTransaction(db *sql.DB) error { return WithTransaction(db, func(tx *sql.Tx) error { // 执行事务操作 _, err := tx.Exec("INSERT INTO logs (message) VALUES ('test')") return err }) }

5.2 错误处理模式

func HandleTransactionError(err error) error { if strings.Contains(err.Error(), "deadlock") { // 死锁，重试 return fmt.Errorf("deadlock detected, consider retrying") } if strings.Contains(err.Error(), "lock wait timeout") { // 锁等待超时 return fmt.Errorf("lock timeout, consider increasing timeout") } return err }