Golang如何统一处理数据库事务错误 实现自动回滚与重试机制

package main

import (
    "context"
    "database/sql"
    "errors"
    "fmt"
    "log"
    "math"
    "math/rand"
    "time"

    _ "github.com/go-sql-driver/mysql" // 假设使用MySQL驱动
)

// ErrTransient 是一个标记接口，用于标识可重试的瞬时错误
type TransientError interface {
    error
    IsTransient() bool
}

// transientError 实现 TransientError 接口
type transientError struct {
    err error
}

func (e *transientError) Error() string {
    return e.err.Error()
}

func (e *transientError) IsTransient() bool {
    return true
}

func NewTransientError(err error) error {
    return &transientError{err: err}
}

// isTransientError 检查错误是否是可重试的瞬时错误
func isTransientError(err error) bool {
    var te TransientError
    return errors.As(err, &te) && te.IsTransient()
}

// WithTransaction 是一个事务包装器，支持自动回滚和重试
func WithTransaction(ctx context.Context, db *sql.DB, fn func(tx *sql.Tx) error) error {
    const maxRetries = 3
    var err error

    for i := 0; i < maxRetries; i++ {
        tx, beginErr := db.BeginTx(ctx, nil) // nil for default options
        if beginErr != nil {
            log.Printf("尝试开启事务失败 (第 %d 次): %v", i+1, beginErr)
            if isTransientError(beginErr) {
                time.Sleep(getBackoffDuration(i))
                continue
            }
            return beginErr // 非瞬时错误，直接返回
        }

        // defer 语句确保事务最终被回滚，除非显式提交
        defer func() {
            if r := recover(); r != nil {
                // 捕获 panic，回滚事务并重新抛出 panic
                if rbErr := tx.Rollback(); rbErr != nil && rbErr != sql.ErrTxDone {
                    log.Printf("Panic 时回滚事务失败: %v", rbErr)
                }
                panic(r)
            } else if err != nil { // 如果业务逻辑返回错误
                if rbErr := tx.Rollback(); rbErr != nil && rbErr != sql.ErrTxDone {
                    log.Printf("业务逻辑错误时回滚事务失败: %v", rbErr)
                }
            }
        }()

        // 执行业务逻辑
        execErr := fn(tx)
        if execErr != nil {
            err = execErr // 记录业务逻辑错误
            log.Printf("业务逻辑执行失败 (第 %d 次): %v", i+1, err)
            if isTransientError(err) {
                // 瞬时错误，等待并重试
                time.Sleep(getBackoffDuration(i))
                continue // 进入下一次重试循环
            }
            return err // 非瞬时错误，直接返回
        }

        // 业务逻辑成功，尝试提交事务
        commitErr := tx.Commit()
        if commitErr != nil {
            err = commitErr // 记录提交错误
            log.Printf("提交事务失败 (第 %d 次): %v", i+1, err)
            if isTransientError(err) {
                // 提交失败可能是瞬时错误（如网络闪断），等待并重试
                time.Sleep(getBackoffDuration(i))
                continue // 进入下一次重试循环
            }
            return err // 非瞬时错误，直接返回
        }

        // 事务成功提交
        return nil
    }

    // 达到最大重试次数仍失败
    return fmt.Errorf("事务操作在 %d 次重试后仍失败: %w", maxRetries, err)
}

// getBackoffDuration 计算指数退避时间，并加入随机抖动
func getBackoffDuration(retryCount int) time.Duration {
    baseDelay := 100 * time.Millisecond // 基础延迟
    maxDelay := 5 * time.Second        // 最大延迟

    // 指数退避：baseDelay * 2^retryCount
    delay := baseDelay * time.Duration(math.Pow(2, float64(retryCount)))
    if delay > maxDelay {
        delay = maxDelay
    }

    // 加入随机抖动，避免“惊群效应”
    jitter := time.Duration(rand.Int63n(int64(delay / 2))) // 0 到 delay/2 的随机值
    return delay + jitter
}

// 示例：如何使用 WithTransaction
func main() {
    // 实际应用中应该从配置加载
    db, err := sql.Open("mysql", "user:password@tcp(127.0.0.1:3306)/dbname?parseTime=true")
    if err != nil {
        log.Fatalf("无法连接数据库: %v", err)
    }
    defer db.Close()

    // 设置连接池参数
    db.SetMaxOpenConns(20)
    db.SetMaxIdleConns(10)
    db.SetConnMaxLifetime(5 * time.Minute)

    // 模拟一个瞬时错误
    var simulateTransientError = true
    var transientErrorCount = 0

    err = WithTransaction(context.Background(), db, func(tx *sql.Tx) error {
        // 模拟业务逻辑
        log.Println("执行业务逻辑...")

        // 模拟第一次和第二次失败为瞬时错误，第三次成功
        if simulateTransientError && transientErrorCount < 2 {
            transientErrorCount++
            log.Printf("模拟瞬时错误，第 %d 次", transientErrorCount)
            return NewTransientError(errors.New("模拟数据库死锁或网络瞬断"))
        }

        // 模拟数据库操作
        _, execErr := tx.ExecContext(context.Background(), "INSERT INTO users (name, email) VALUES (?, ?)", "Test User", fmt.Sprintf("test%d@example.com", time.Now().UnixNano()))
        if execErr != nil {
            log.Printf("插入数据失败: %v", execErr)
            // 这里可以根据 execErr 的具体类型判断是否是瞬时错误，例如 MySQL 的死锁错误码
            // if mysqlErr, ok := execErr.(*mysql.MySQLError); ok && mysqlErr.Number == 1213 {
            //     return NewTransientError(execErr)
            // }
            return execErr // 其他非瞬时错误直接返回
        }

        log.Println("业务逻辑执行成功，准备提交。")
        return nil
    })

    if err != nil {
        log.Printf("事务最终失败: %v", err)
    } else {
        log.Println("事务成功完成！")
    }

    // 模拟一个非瞬时错误
    simulateTransientError = false // 禁用瞬时错误模拟
    err = WithTransaction(context.Background(), db, func(tx *sql.Tx) error {
        log.Println("执行业务逻辑 (非瞬时错误模拟)...")
        // 模拟一个非瞬时错误，例如唯一约束冲突
        _, execErr := tx.ExecContext(context.Background(), "INSERT INTO users (name, email) VALUES (?, ?)", "Existing User", "existing@example.com")
        if execErr != nil {
            log.Printf("插入数据失败 (非瞬时错误): %v", execErr)
            // 假设这个错误不是瞬时错误，直接返回
            return errors.New("模拟唯一约束冲突或逻辑错误")
        }
        return nil
    })

    if err != nil {
        log.Printf("事务最终失败 (非瞬时错误): %v", err)
    } else {
        log.Println("事务成功完成 (非瞬时错误)！")
    }
}