golang

数据库事务：保证数据一致性的关键

系统讲解 Go 语言中数据库事务的 ACID 特性、隔离级别、锁机制，并结合银行转账和订单系统的实战案例深入理解事务处理

Leeting Yan 2025-08-13 13 分钟阅读 6384 字

数据库事务：保证数据一致性的关键

在数据库操作中，我们经常需要执行多个相关的操作，这些操作要么全部成功，要么全部失败。事务（Transaction）就是用来保证这种原子性的机制。

本文将深入探讨数据库事务的原理和在 Go 中的实现。

什么是事务？

事务是一组数据库操作的逻辑单元，具有以下 ACID 特性：

原子性（Atomicity）：事务中的操作要么全部成功，要么全部失败
一致性（Consistency）：事务执行前后，数据库保持一致状态
隔离性（Isolation）：并发事务之间互不干扰
持久性（Durability）：事务提交后，数据永久保存

基础事务操作

package main

import (
    "context"
    "database/sql"
    "log"
    
    _ "github.com/go-sql-driver/mysql"
)

func main() {
    db, err := sql.Open("mysql", "user:password@/dbname")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    
    // 开始事务
    tx, err := db.Begin()
    if err != nil {
        log.Fatal(err)
    }
    
    // 执行操作
    _, err = tx.Exec("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    if err != nil {
        tx.Rollback()
        log.Fatal(err)
    }
    
    _, err = tx.Exec("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    if err != nil {
        tx.Rollback()
        log.Fatal(err)
    }
    
    // 提交事务
    err = tx.Commit()
    if err != nil {
        log.Fatal(err)
    }
    
    log.Println("Transaction committed")
}

使用 defer 确保回滚

func transfer(db *sql.DB, fromID, toID int, amount float64) (err error) {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    
    // 确保在函数退出时回滚（如果未提交）
    defer func() {
        if err != nil {
            tx.Rollback()
            return
        }
        err = tx.Commit()
    }()
    
    // 扣款
    _, err = tx.Exec("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID)
    if err != nil {
        return err
    }
    
    // 检查余额
    var balance float64
    err = tx.QueryRow("SELECT balance FROM accounts WHERE id = ?", fromID).Scan(&balance)
    if err != nil {
        return err
    }
    if balance < 0 {
        return fmt.Errorf("insufficient balance")
    }
    
    // 入账
    _, err = tx.Exec("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID)
    if err != nil {
        return err
    }
    
    return nil
}

事务与 Context

func transferWithContext(ctx context.Context, db *sql.DB, fromID, toID int, amount float64) error {
    // 使用 context 开始事务
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 执行操作（带 context）
    _, err = tx.ExecContext(ctx, 
        "UPDATE accounts SET balance = balance - ? WHERE id = ?", 
        amount, fromID)
    if err != nil {
        return err
    }
    
    _, err = tx.ExecContext(ctx, 
        "UPDATE accounts SET balance = balance + ? WHERE id = ?", 
        amount, toID)
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

事务隔离级别

// 设置隔离级别
tx, err := db.BeginTx(ctx, &sql.TxOptions{
    Isolation: sql.LevelSerializable,
})

// 常用隔离级别
// LevelReadUncommitted - 读未提交
// LevelReadCommitted - 读已提交
// LevelRepeatableRead - 可重复读
// LevelSerializable - 串行化

隔离级别与并发问题

// 脏读（Dirty Read）
// 事务 A 读取了事务 B 未提交的数据

// 不可重复读（Non-repeatable Read）
// 事务 A 两次读取同一数据，结果不同（事务 B 修改了数据）

// 幻读（Phantom Read）
// 事务 A 两次查询，结果集不同（事务 B 插入了新数据）

乐观锁与悲观锁

悲观锁

// 使用 SELECT ... FOR UPDATE
func updateWithPessimisticLock(ctx context.Context, db *sql.DB, id int, amount float64) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 锁定行
    var balance float64
    err = tx.QueryRowContext(ctx, 
        "SELECT balance FROM accounts WHERE id = ? FOR UPDATE", id).
        Scan(&balance)
    if err != nil {
        return err
    }
    
    // 更新
    newBalance := balance + amount
    _, err = tx.ExecContext(ctx, 
        "UPDATE accounts SET balance = ? WHERE id = ?", 
        newBalance, id)
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

乐观锁

// 使用版本号
func updateWithOptimisticLock(ctx context.Context, db *sql.DB, id int, amount float64) error {
    for i := 0; i < 3; i++ { // 最多重试 3 次
        // 读取当前值和版本号
        var balance float64
        var version int
        err := db.QueryRowContext(ctx, 
            "SELECT balance, version FROM accounts WHERE id = ?", id).
            Scan(&balance, &version)
        if err != nil {
            return err
        }
        
        // 更新（带版本号检查）
        newBalance := balance + amount
        result, err := db.ExecContext(ctx, 
            "UPDATE accounts SET balance = ?, version = version + 1 WHERE id = ? AND version = ?", 
            newBalance, id, version)
        if err != nil {
            return err
        }
        
        rows, err := result.RowsAffected()
        if err != nil {
            return err
        }
        
        if rows > 0 {
            return nil // 更新成功
        }
        
        // 版本号不匹配，重试
        time.Sleep(100 * time.Millisecond)
    }
    
    return fmt.Errorf("failed to update after retries")
}

嵌套事务（Savepoint）

func nestedTransaction(ctx context.Context, db *sql.DB) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 外层事务操作
    _, err = tx.ExecContext(ctx, "INSERT INTO orders (user_id) VALUES (1)")
    if err != nil {
        return err
    }
    
    // 创建保存点
    _, err = tx.ExecContext(ctx, "SAVEPOINT sp1")
    if err != nil {
        return err
    }
    
    // 内层事务操作
    _, err = tx.ExecContext(ctx, "INSERT INTO order_items (order_id, product_id) VALUES (1, 1)")
    if err != nil {
        // 回滚到保存点
        tx.ExecContext(ctx, "ROLLBACK TO SAVEPOINT sp1")
        log.Println("Rolled back to savepoint")
    }
    
    // 继续外层事务
    _, err = tx.ExecContext(ctx, "UPDATE users SET order_count = order_count + 1 WHERE id = 1")
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

实战：订单系统

package main

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "time"
)

type OrderService struct {
    db *sql.DB
}

func NewOrderService(db *sql.DB) *OrderService {
    return &OrderService{db: db}
}

// CreateOrder 创建订单（事务）
func (s *OrderService) CreateOrder(ctx context.Context, userID int, items []OrderItem) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 1. 创建订单
    var orderID int
    err = tx.QueryRowContext(ctx, 
        "INSERT INTO orders (user_id, created_at) VALUES (?, ?) RETURNING id",
        userID, time.Now()).
        Scan(&orderID)
    if err != nil {
        return fmt.Errorf("create order: %w", err)
    }
    
    // 2. 创建订单项并扣减库存
    totalAmount := 0.0
    for _, item := range items {
        // 检查库存（悲观锁）
        var stock int
        var price float64
        err = tx.QueryRowContext(ctx, 
            "SELECT stock, price FROM products WHERE id = ? FOR UPDATE",
            item.ProductID).
            Scan(&stock, &price)
        if err != nil {
            return fmt.Errorf("check stock: %w", err)
        }
        
        if stock < item.Quantity {
            return fmt.Errorf("insufficient stock for product %d", item.ProductID)
        }
        
        // 扣减库存
        _, err = tx.ExecContext(ctx, 
            "UPDATE products SET stock = stock - ? WHERE id = ?",
            item.Quantity, item.ProductID)
        if err != nil {
            return fmt.Errorf("update stock: %w", err)
        }
        
        // 创建订单项
        amount := price * float64(item.Quantity)
        _, err = tx.ExecContext(ctx, 
            "INSERT INTO order_items (order_id, product_id, quantity, price, amount) VALUES (?, ?, ?, ?, ?)",
            orderID, item.ProductID, item.Quantity, price, amount)
        if err != nil {
            return fmt.Errorf("create order item: %w", err)
        }
        
        totalAmount += amount
    }
    
    // 3. 更新订单总金额
    _, err = tx.ExecContext(ctx, 
        "UPDATE orders SET total_amount = ? WHERE id = ?",
        totalAmount, orderID)
    if err != nil {
        return fmt.Errorf("update order amount: %w", err)
    }
    
    // 4. 记录用户购买历史
    _, err = tx.ExecContext(ctx, 
        "INSERT INTO purchase_history (user_id, order_id, amount) VALUES (?, ?, ?)",
        userID, orderID, totalAmount)
    if err != nil {
        return fmt.Errorf("record purchase: %w", err)
    }
    
    return tx.Commit()
}

type OrderItem struct {
    ProductID int
    Quantity  int
}

func main() {
    db, err := sql.Open("mysql", "user:password@/shop")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    
    service := NewOrderService(db)
    ctx := context.Background()
    
    items := []OrderItem{
        {ProductID: 1, Quantity: 2},
        {ProductID: 2, Quantity: 1},
    }
    
    err = service.CreateOrder(ctx, 123, items)
    if err != nil {
        log.Printf("Create order failed: %v", err)
        return
    }
    
    log.Println("Order created successfully")
}

事务最佳实践

1. 保持事务简短

// ❌ 不好：事务中包含耗时操作
func badTransaction(db *sql.DB) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    _, err = tx.Exec("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    if err != nil {
        return err
    }
    
    // 耗时的外部调用
    err = callExternalService() // 可能很慢
    if err != nil {
        return err
    }
    
    _, err = tx.Exec("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

// ✅ 好：事务只包含数据库操作
func goodTransaction(db *sql.DB) error {
    // 先执行外部调用
    err := callExternalService()
    if err != nil {
        return err
    }
    
    // 再执行事务
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    _, err = tx.Exec("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    if err != nil {
        return err
    }
    
    _, err = tx.Exec("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

2. 处理死锁

func handleDeadlock(db *sql.DB, maxRetries int) error {
    for i := 0; i < maxRetries; i++ {
        err := doTransaction(db)
        if err == nil {
            return nil
        }
        
        // 检查是否是死锁错误
        if isDeadlockError(err) {
            log.Printf("Deadlock detected, retry %d/%d", i+1, maxRetries)
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
            continue
        }
        
        return err
    }
    
    return fmt.Errorf("max retries exceeded")
}

func isDeadlockError(err error) bool {
    // MySQL: Error 1213
    // PostgreSQL: Error 40P01
    return strings.Contains(err.Error(), "deadlock")
}

3. 使用连接池

db.SetMaxOpenConns(25)
db.SetMaxIdleConns(5)
db.SetConnMaxLifetime(5 * time.Minute)

总结

事务是保证数据一致性的关键机制：

ACID 特性：原子性、一致性、隔离性、持久性
隔离级别：从低到高依次为读未提交、读已提交、可重复读、串行化
并发控制：悲观锁（SELECT FOR UPDATE）和乐观锁（版本号）
最佳实践：保持事务简短、处理死锁、使用连接池

记住：事务虽然强大，但也会带来性能开销。要在一致性和性能之间找到平衡。

进阶概念

ACID 特性详述

ACID 是事务的四个核心特性，理解其底层实现有助于更好地设计和优化数据库操作。

原子性（Atomicity）：事务中的所有操作要么全部完成，要么全部不完成。数据库通过 undo log（回滚日志）实现原子性。在事务执行过程中，所有修改先写入 undo log，如果事务失败，数据库利用 undo log 将数据恢复到事务开始前的状态。MySQL 的 InnoDB 引擎在事务提交前不会真正写入数据文件，而是先记录在 redo log 中，保证即使系统崩溃也能恢复。

一致性（Consistency）：事务执行前后，数据库必须从一个一致状态转换到另一个一致状态。一致性不仅依赖数据库自身的约束（外键、唯一索引、CHECK 约束），还依赖应用层面的业务规则。例如，银行转账业务中，转出账户扣款和转入账户加款之和应保持不变，这就是业务层面的一致性保障。

隔离性（Isolation）：并发执行的事务彼此隔离，互不影响。数据库通过锁机制和多版本并发控制（MVCC）实现隔离性。MVCC 允许读操作不加锁，通过保存数据的多个版本实现非阻塞读，大幅提升并发性能。InnoDB 的 Read View 机制就是 MVCC 在 MySQL 中的实现。

持久性（Durability）：一旦事务提交，对数据库的修改就是永久的。即使系统崩溃，已提交的数据也不会丢失。数据库通过 redo log（重做日志）和 binlog（二进制日志）实现持久性。redo log 用于崩溃恢复，binlog 用于数据复制和 point-in-time 恢复。

隔离级别与并发异常详解

数据库定义了四种标准隔离级别，从低到高依次提升一致性，但并发性能逐步下降。

读未提交（Read Uncommitted）：事务可以读取其他事务未提交的数据。这是隔离级别最低的设置，会产生脏读（Dirty Read）、不可重复读和幻读三种问题。实际生产环境几乎不使用此级别。

读已提交（Read Committed）：事务只能读取其他事务已提交的数据。避免了脏读，但不可重复读问题仍然存在。Oracle 和 PostgreSQL 的默认隔离级别。在此级别下，每次 SELECT 都会重新生成 Read View，因此同一事务内两次查询可能看到不同数据。

可重复读（Repeatable Read）：同一事务内多次读取同一数据，结果一致。MySQL InnoDB 的默认隔离级别。通过事务开始时生成 Read View 并在整个事务期间保持不变来实现。InnoDB 在此级别下通过间隙锁（Gap Lock）还解决了幻读问题，因此 MySQL 的可重复读实际上避免了绝大多数幻读场景。

串行化（Serializable）：最高的隔离级别，通过强制事务串行执行来避免所有并发问题。性能最差，通常只在数据一致性要求极高且并发量很小的场景使用。

三种并发异常的具体表现：

脏读：事务 A 读取了事务 B 修改但未提交的数据。如果 B 回滚，A 读取到的就是无效数据。
不可重复读：事务 A 在同一事务内两次读取同一行数据，由于事务 B 修改并提交了该行，导致 A 两次读取结果不同。
幻读：事务 A 执行范围查询，事务 B 在范围内插入新数据并提交，A 再次执行相同查询时结果集增多。

分布式事务：2PC 与 Saga

在微服务架构中，数据往往分散在多个数据库中，本地事务无法满足需求，需要分布式事务方案。

两阶段提交（2PC）：分为准备阶段和提交阶段。协调者询问所有参与者是否可以提交，所有参与者回答是后，协调者再发出正式提交指令。2PC 的最大问题是阻塞：如果协调者在第二阶段崩溃，参与者会持有锁等待协调者恢复，严重影响可用性。

// 2PC 简化示意：Go 中的分布式事务实现
func twoPhaseCommit(ctx context.Context, participants []Participant) error {
    // 第一阶段：准备
    for _, p := range participants {
        if err := p.Prepare(ctx); err != nil {
            // 有参与者准备失败，回滚所有已准备的参与者
            for _, prepared := range participants {
                _ = prepared.Rollback(ctx)
            }
            return fmt.Errorf("prepare failed: %w", err)
        }
    }
    
    // 第二阶段：提交
    for _, p := range participants {
        if err := p.Commit(ctx); err != nil {
            // 记录日志，需要人工或补偿机制处理
            log.Printf("Commit failed for participant: %v", err)
        }
    }
    return nil
}

Saga 模式：将长事务拆分为一系列本地事务，每个本地事务完成后发送事件触发下一个事务。如果某个步骤失败，则执行补偿操作回滚之前的步骤。Saga 有两种实现方式：

编排式（Choreography）：每个服务完成操作后发送事件，其他服务监听事件并执行下一步
编排式（Orchestration）：由 Saga 协调器集中管理服务调用顺序

// Saga 编排式示例
func sagaOrchestrator(ctx context.Context, order Order) error {
    // 步骤1：扣减库存
    if err := inventoryService.Deduct(ctx, order.Items); err != nil {
        return err
    }
    
    // 步骤2：创建订单
    orderID, err := orderService.Create(ctx, order)
    if err != nil {
        // 补偿：恢复库存
        _ = inventoryService.Restore(ctx, order.Items)
        return err
    }
    
    // 步骤3：扣款
    if err := paymentService.Charge(ctx, order.UserID, order.TotalAmount); err != nil {
        // 补偿：恢复库存、取消订单
        _ = inventoryService.Restore(ctx, order.Items)
        _ = orderService.Cancel(ctx, orderID)
        return err
    }
    
    return nil
}

乐观锁与悲观锁深入对比

特性	悲观锁	乐观锁
加锁时机	读取时立即加锁	不加锁，提交时检查冲突
适用场景	写冲突概率高	读多写少
性能影响	持有锁期间阻塞其他事务	无锁，冲突时需要重试
实现复杂度	简单，数据库原生支持	需要版本号或时间戳机制
死锁风险	有	无
Go 实现	SELECT FOR UPDATE	版本号 CAS 更新

// 悲观锁在高并发转账场景中的应用
func pessimisticTransfer(ctx context.Context, db *sql.DB, fromID, toID int, amount float64) error {
    tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 按固定顺序加锁，避免死锁
    firstID, secondID := fromID, toID
    if fromID > toID {
        firstID, secondID = toID, fromID
    }
    
    var fromBalance, toBalance float64
    err = tx.QueryRowContext(ctx, "SELECT balance FROM accounts WHERE id = ? FOR UPDATE", firstID).Scan(&fromBalance)
    if err != nil {
        return err
    }
    err = tx.QueryRowContext(ctx, "SELECT balance FROM accounts WHERE id = ? FOR UPDATE", secondID).Scan(&toBalance)
    if err != nil {
        return err
    }
    
    if firstID == fromID {
        if fromBalance < amount {
            return fmt.Errorf("insufficient balance")
        }
        _, _ = tx.ExecContext(ctx, "UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID)
        _, _ = tx.ExecContext(ctx, "UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID)
    } else {
        if toBalance < amount {
            return fmt.Errorf("insufficient balance")
        }
        _, _ = tx.ExecContext(ctx, "UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID)
        _, _ = tx.ExecContext(ctx, "UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID)
    }
    
    return tx.Commit()
}

实战案例

银行转账完整实现

以下是一个生产级的银行转账实现，包含余额检查、幂等性控制和审计日志：

package main

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "time"
    
    _ "github.com/go-sql-driver/mysql"
)

type TransferService struct {
    db *sql.DB
}

// TransferRequest 转账请求
func (s *TransferService) Transfer(ctx context.Context, requestID string, fromID, toID int, amount float64) error {
    return s.db.Transaction(func(tx *sql.Tx) error {
        // 幂等性检查
        var exists int
        err := tx.QueryRowContext(ctx,
            "SELECT COUNT(*) FROM transfer_log WHERE request_id = ?", requestID).Scan(&exists)
        if err != nil {
            return err
        }
        if exists > 0 {
            log.Printf("Duplicate request: %s", requestID)
            return nil // 已处理过，直接返回成功
        }
        
        // 检查转出账户余额
        var fromBalance float64
        err = tx.QueryRowContext(ctx,
            "SELECT balance FROM accounts WHERE id = ? FOR UPDATE", fromID).Scan(&fromBalance)
        if err != nil {
            return fmt.Errorf("query from account: %w", err)
        }
        if fromBalance < amount {
            return fmt.Errorf("insufficient balance: %.2f < %.2f", fromBalance, amount)
        }
        
        // 执行转账
        _, err = tx.ExecContext(ctx,
            "UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, fromID)
        if err != nil {
            return fmt.Errorf("deduct from account: %w", err)
        }
        
        _, err = tx.ExecContext(ctx,
            "UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, toID)
        if err != nil {
            return fmt.Errorf("add to account: %w", err)
        }
        
        // 记录转账日志
        _, err = tx.ExecContext(ctx,
            "INSERT INTO transfer_log (request_id, from_id, to_id, amount, created_at) VALUES (?, ?, ?, ?, ?)",
            requestID, fromID, toID, amount, time.Now())
        if err != nil {
            return fmt.Errorf("log transfer: %w", err)
        }
        
        return nil
    })
}

// TransferWithTimeout 带超时的转账
func (s *TransferService) TransferWithTimeout(ctx context.Context, requestID string, fromID, toID int, amount float64) error {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()
    return s.Transfer(ctx, requestID, fromID, toID, amount)
}

库存扣减防超卖

电商系统中库存扣减是典型的高并发事务场景，需要防止超卖同时保证性能：

// DeductStock 库存扣减（乐观锁实现）
func DeductStock(ctx context.Context, db *sql.DB, productID int, quantity int) error {
    for attempt := 0; attempt < 5; attempt++ {
        tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
        if err != nil {
            return err
        }
        
        var stock, version int
        err = tx.QueryRowContext(ctx,
            "SELECT stock, version FROM products WHERE id = ?", productID).
            Scan(&stock, &version)
        if err != nil {
            tx.Rollback()
            return err
        }
        
        if stock < quantity {
            tx.Rollback()
            return fmt.Errorf("insufficient stock: %d < %d", stock, quantity)
        }
        
        result, err := tx.ExecContext(ctx,
            "UPDATE products SET stock = stock - ?, version = version + 1 WHERE id = ? AND version = ?",
            quantity, productID, version)
        if err != nil {
            tx.Rollback()
            return err
        }
        
        affected, err := result.RowsAffected()
        if err != nil {
            tx.Rollback()
            return err
        }
        
        if affected > 0 {
            // 扣减成功，记录库存变动日志
            _, _ = tx.ExecContext(ctx,
                "INSERT INTO stock_log (product_id, delta, old_stock, new_stock, created_at) VALUES (?, ?, ?, ?, ?)",
                productID, -quantity, stock, stock-quantity, time.Now())
            return tx.Commit()
        }
        
        tx.Rollback()
        // 版本冲突，重试
        time.Sleep(time.Duration(attempt+1) * 10 * time.Millisecond)
    }
    
    return fmt.Errorf("failed to deduct stock after retries")
}

订单创建完整事务

订单创建涉及 order、order_item、inventory、user_coupon 等多个表的更新：

func CreateOrderWithCoupon(ctx context.Context, db *sql.DB, userID int, cartItems []CartItem, couponID int) (*OrderResult, error) {
    tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
    if err != nil {
        return nil, err
    }
    defer tx.Rollback()
    
    // 1. 验证优惠券
    var couponValue float64
    var couponUsed bool
    err = tx.QueryRowContext(ctx,
        "SELECT value, used FROM user_coupons WHERE id = ? AND user_id = ? FOR UPDATE",
        couponID, userID).Scan(&couponValue, &couponUsed)
    if err != nil {
        return nil, fmt.Errorf("coupon not found: %w", err)
    }
    if couponUsed {
        return nil, fmt.Errorf("coupon already used")
    }
    
    // 2. 扣减库存并计算总价
    var totalAmount float64
    for _, item := range cartItems {
        var stock int
        var price float64
        err = tx.QueryRowContext(ctx,
            "SELECT stock, price FROM products WHERE id = ? FOR UPDATE", item.ProductID).
            Scan(&stock, &price)
        if err != nil {
            return nil, err
        }
        if stock < item.Quantity {
            return nil, fmt.Errorf("product %d out of stock", item.ProductID)
        }
        
        _, err = tx.ExecContext(ctx,
            "UPDATE products SET stock = stock - ? WHERE id = ?", item.Quantity, item.ProductID)
        if err != nil {
            return nil, err
        }
        
        totalAmount += price * float64(item.Quantity)
    }
    
    // 3. 应用优惠券
    finalAmount := totalAmount - couponValue
    if finalAmount < 0 {
        finalAmount = 0
    }
    
    // 4. 创建订单
    result, err := tx.ExecContext(ctx,
        "INSERT INTO orders (user_id, total_amount, final_amount, coupon_id, status, created_at) VALUES (?, ?, ?, ?, ?, ?)",
        userID, totalAmount, finalAmount, couponID, "PENDING", time.Now())
    if err != nil {
        return nil, err
    }
    orderID, _ := result.LastInsertId()
    
    // 5. 创建订单项
    for _, item := range cartItems {
        _, err = tx.ExecContext(ctx,
            "INSERT INTO order_items (order_id, product_id, quantity) VALUES (?, ?, ?)",
            orderID, item.ProductID, item.Quantity)
        if err != nil {
            return nil, err
        }
    }
    
    // 6. 标记优惠券已使用
    _, err = tx.ExecContext(ctx,
        "UPDATE user_coupons SET used = true WHERE id = ?", couponID)
    if err != nil {
        return nil, err
    }
    
    if err := tx.Commit(); err != nil {
        return nil, err
    }
    
    return &OrderResult{
        OrderID:     orderID,
        TotalAmount: totalAmount,
        FinalAmount: finalAmount,
    }, nil
}

各隔离级别性能对比

隔离级别	脏读	不可重复读	幻读	读性能	写性能
读未提交	允许	允许	允许	最优	最优
读已提交	禁止	允许	允许	优	优
可重复读	禁止	禁止	禁止*	良	良
串行化	禁止	禁止	禁止	差	差

*注：MySQL InnoDB 的可重复读通过间隙锁解决了幻读问题，但其他数据库引擎可能不完全解决。

// 隔离级别基准测试示例
func BenchmarkTransactionIsolation(b *testing.B) {
    db, _ := sql.Open("mysql", "user:password@/test")
    defer db.Close()
    
    levels := map[string]sql.IsolationLevel{
        "ReadCommitted":  sql.LevelReadCommitted,
        "RepeatableRead": sql.LevelRepeatableRead,
        "Serializable":   sql.LevelSerializable,
    }
    
    for name, level := range levels {
        b.Run(name, func(b *testing.B) {
            for i := 0; i < b.N; i++ {
                tx, _ := db.BeginTx(context.Background(), &sql.TxOptions{Isolation: level})
                _, _ = tx.Exec("SELECT id, balance FROM accounts WHERE id = 1")
                _ = tx.Commit()
            }
        })
    }
}

FAQ

1. Go 的 sql.Tx 支持嵌套事务吗？

database/sql 包的 Tx 本身不支持嵌套事务，但可以通过 SAVEPOINT 机制模拟。对于复杂业务，建议使用外部库如 jmoiron/sqlx 或直接手动管理 SAVEPOINT。

2. 事务中遇到网络中断怎么办？

如果事务在提交前网络中断，数据库会等待连接超时后自动回滚未提交的事务。如果提交已发送但响应丢失，数据库可能已提交成功，应用需要设计幂等性机制来应对这种不确定性。

3. 长事务有什么危害？

长事务会长时间持有锁，阻塞其他事务的执行，同时可能导致 undo log 膨胀和 MVCC 版本链过长，严重影响并发性能。应将事务限制在最小必要范围内。

4. 乐观锁和悲观锁如何选择？

读多写少的场景优先选择乐观锁，减少锁争用提升并发能力。写冲突频繁或数据竞争严重的场景选择悲观锁，避免大量重试造成的额外开销。

5. 事务隔离级别越高越好吗？

不是。更高的隔离级别意味着更强的锁竞争和更少的并发能力。应根据业务需求选择合适的隔离级别，通常读已提交或 MySQL 的可重复读已能满足大部分需求。

6. 分布式事务有哪些替代方案？

除了 2PC 和 Saga，还可以使用本地消息表、事务消息（如 RocketMQ）、TCC（Try-Confirm-Cancel）模式、最大努力通知等方案，根据业务对一致性的要求灵活选择。

7. 如何撤销已提交的事务？

已提交的事务只能通过补偿事务来回滚。设计业务逻辑时预留反向操作，例如转账的补偿就是反向转账，创建订单的补偿就是取消订单。

8. context 超时后事务会怎样？

使用 BeginTx 时，如果 context 超时或被取消，未提交的事务会自动回滚。BEGIN 语句本身不会超时，但后续查询和提交会受 context 影响。

最佳实践

始终使用 defer 确保回滚：防止因 panic 或提前 return 导致事务未回滚
事务内只包含数据库操作：网络调用、IO 操作移到事务外执行
正确设置连接池参数：根据 QPS 和数据库连接限制调整 MaxOpenConns
加锁顺序一致：多表操作时按固定顺序加锁，避免死锁
核心业务操作记录日志：转账、支付等关键操作保留审计日志
幂等性设计：接口支持幂等调用，应对网络超时和重试场景
合理选择隔离级别：不要一味追求最高隔离级别
监控事务执行时间：慢事务告警，及时发现和优化

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页

数据库事务：保证数据一致性的关键

什么是事务？

基础事务操作

使用 defer 确保回滚

事务与 Context

事务隔离级别

隔离级别与并发问题

乐观锁与悲观锁

悲观锁

乐观锁

嵌套事务（Savepoint）

实战：订单系统

事务最佳实践

1. 保持事务简短

2. 处理死锁

3. 使用连接池

总结

进阶概念

ACID 特性详述

隔离级别与并发异常详解

分布式事务：2PC 与 Saga

乐观锁与悲观锁深入对比

实战案例

银行转账完整实现

库存扣减防超卖

订单创建完整事务

各隔离级别性能对比

FAQ

1. Go 的 sql.Tx 支持嵌套事务吗？

2. 事务中遇到网络中断怎么办？

3. 长事务有什么危害？

4. 乐观锁和悲观锁如何选择？

5. 事务隔离级别越高越好吗？

6. 分布式事务有哪些替代方案？

7. 如何撤销已提交的事务？

8. context 超时后事务会怎样？

最佳实践

延伸阅读

探索更多技术文章

「golang」更多文章