示例说明
可串行化隔离级别下数据库主动检测冲突并回滚其中一个事务，这不是失败而是并发控制的正常反馈——正确响应是重放整个事务单元。BeginFunc 把事务体收拢成函数，任何返回错误自动回滚；40001 错误码被识别后按递增延迟重试，业务规则不满足的错误直接上抛不浪费预算。
技术要点
- 40001 错误码识别序列化冲突
- BeginFunc 函数式事务自动回滚
- for update 行锁锁定读改写窗口
代码解读
事务体内先 select for update 锁定目标行再读余额——锁保证读到提交后的最新值；余额不足返回普通错误走失败路径。runWithRetry 捕获 BeginFunc 的错误分类处理：序列化冲突退避后 continue 重放全部语句，四次预算耗尽或其他错误原样返回。
使用建议
重试次数按业务时延预算设定。热点行冲突率高时考虑队列化削峰。事务体必须无副作用外泄，重放才能安全。
代码示例
package main
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
)
func isSerializationFailure(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) && pgErr.Code == "40001"
}
func runWithRetry(ctx context.Context, pool *pgxpool.Pool, attempts int) (int, error) {
var balance int
for attempt := 1; attempt <= attempts; attempt++ {
err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
row := tx.QueryRow(ctx,
"select balance from accounts where id = 1 for update")
if err := row.Scan(&balance); err != nil {
return fmt.Errorf("lock read: %w", err)
}
if balance < 100 {
return errors.New("insufficient funds")
}
_, err := tx.Exec(ctx,
"update accounts set balance = balance - 100 where id = 1")
return err
})
if err == nil {
return balance - 100, nil
}
if isSerializationFailure(err) && attempt < attempts {
time.Sleep(time.Duration(attempt*50) * time.Millisecond)
continue // 序列化冲突：整块重放
}
return 0, err
}
return 0, errors.New("unreachable")
}
func main() {
config, _ := pgxpool.ParseConfig("postgres://user:pass@localhost/app")
pool, err := pgxpool.NewWithConfig(context.Background(), config)
if err != nil {
fmt.Println(err)
return
}
defer pool.Close()
final, err := runWithRetry(context.Background(), pool, 4)
fmt.Println("final balance:", final, "err:", err)
}