Golang 數據庫事務實踐
作者：俞凡
事務是很多業務的基礎，本文介紹了如何在Golang里實現數據庫事務操作，并以一個用戶注冊場景給出了完整實現。
Go 是一種年輕而強大的語言，專為編寫小型、簡單的服務而創建。但隨著時間推移，越來越多復雜應用和系統也在采用 Go 進行開發，這就出現了一些問題：如何處理事務？
為了深入探討這個問題，我們假設一個簡單的業務場景：用戶注冊。
作為一個系統，我希望在注冊時創建用戶和個人資料。
RDBMS/DBMS 的現代 Go 庫不像 C# 和 Java 的 Hibernate、Entity Framework 那樣強大，因此我們必須自己處理。為了實現用戶注冊業務場景，我們將創建并評估幾種處理事務的方法。
由于每種事務處理方法都必須與 sql.DB 和 sql.Tx 配合使用，因此需要引入接口來封裝對數據庫的訪問。
生成的應用有兩個域實體和一個用于訪問數據庫的 DB 低級接口。
package model
type User struct {
Email string
}
type Profile struct {
Name string
}package transaction
type DB interface {
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
}準備工作完成后，就可以采用如下兩種方法。
1. 事務感知上下文
工作原理：transaction.Manager啟動事務并將其放入上下文。當存儲庫執行查詢時，助手會檢查上下文中是否有事務，并使用創建的事務來執行查詢，或者如果上下文為空，則不使用事務來執行查詢。
為了啟動事務，我們需要實體：Manager
package transaction
type Manager interface {
Run(
ctx context.Context,
callback func(ctx context.Context) error,
) error
}transaction.Manager 實現：
package transaction
import (
"context"
"database/sql"
"github.com/pkg/errors"
"go.uber.org/multierr"
)
type txKey string
var ctxWithTx = txKey("tx")
type SQLTransactionManager struct {
db *sql.DB
}
func NewManager(db *sql.DB) *SQLTransactionManager {
return &SQLTransactionManager{db: db}
}
func (m *SQLTransactionManager) Run(
ctx context.Context,
callback func(ctx context.Context) error,
) (rErr error) {
tx, err := m.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return errors.WithStack(err)
}
defer func() {
if rErr != nil {
rErr = multierr.Combine(rErr, errors.WithStack(tx.Rollback()))
}
}()
defer func() {
if rec := recover(); rec != nil {
if e, ok := rec.(error); ok {
rErr = e
} else {
rErr = errors.Errorf("%s", rec)
}
}
}()
if err = callback(putTxToContext(ctx, tx)); err != nil {
return err
}
return errors.WithStack(tx.Commit())
}
func ExtractTxFromContext(ctx context.Context) (*sql.Tx, bool) {
tx := ctx.Value(ctxWithTx)
if t, ok := tx.(*sql.Tx); ok {
return t, true
}
return nil, false
}
func putTxToContext(ctx context.Context, tx *sql.Tx) context.Context {
return context.WithValue(ctx, ctxWithTx, tx)
}DB實現：
package storage
import (
"brand/transaction/example1/transaction"
"context"
"database/sql"
)
type DB struct {
db *sql.DB
}
func NewDB(db *sql.DB) *DB {
return &DB{db: db}
}
func (d *DB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
tx, ok := transaction.ExtractTxFromContext(ctx)
if !ok {
return d.db.QueryRowContext(ctx, query, args...)
}
return tx.QueryRowContext(ctx, query, args...)
}
func (d *DB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
tx, ok := transaction.ExtractTxFromContext(ctx)
if !ok {
return d.db.QueryContext(ctx, query, args...)
}
return tx.QueryContext(ctx, query, args...)
}
func (d *DB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
tx, ok := transaction.ExtractTxFromContext(ctx)
if !ok {
return d.db.ExecContext(ctx, query, args...)
}
return tx.ExecContext(ctx, query, args...)
}
func (d *DB) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
tx, ok := transaction.ExtractTxFromContext(ctx)
if !ok {
return d.db.PrepareContext(ctx, query)
}
return tx.PrepareContext(ctx, query)
}RegistrationService 負責用戶注冊業務場景
package service
import (
"brand/transaction/example1/model"
"brand/transaction/example1/transaction"
"context"
)
type UserRepository interface {
Create(ctx context.Context, user *model.User) error
}
type ProfileRepository interface {
Create(ctx context.Context, user *model.Profile) error
}
type RegistrationData struct {
Email string
Name string
}
type RegistrationService struct {
transactionManager transaction.Manager
userRepository UserRepository
profileRepository ProfileRepository
}
func NewRegistrationService(
transactionManager transaction.Manager,
userRepository UserRepository,
profileRepository ProfileRepository,
) *RegistrationService {
return &RegistrationService{
transactionManager: transactionManager,
userRepository: userRepository,
profileRepository: profileRepository,
}
}
func (s *RegistrationService) Register(ctx context.Context, data RegistrationData) error {
return s.transactionManager.Run(ctx, func(ctx context.Context) error {
if err := s.userRepository.Create(ctx, &model.User{
Email: data.Email,
}); err != nil {
return err
}
if err := s.profileRepository.Create(ctx, &model.Profile{
Name: data.Name,
}); err != nil {
return err
}
return nil
})
}User和ProfileRepository的實現：
package storage
import (
"brand/transaction"
"brand/transaction/example1/model"
"context"
)
type ProfileRepository struct {
db transaction.DB
}
func NewProfileRepository(db transaction.DB) *ProfileRepository {
return &ProfileRepository{db: db}
}
func (r *ProfileRepository) Create(ctx context.Context, profile *model.Profile) error {
_, err := r.db.ExecContext(ctx, "INSERT ...", profile.Name)
return err
}package storage
import (
"brand/transaction"
"brand/transaction/example1/model"
"context"
)
type UserRepository struct {
db transaction.DB
}
func NewUserRepository(db transaction.DB) *UserRepository {
return &UserRepository{db: db}
}
func (r *UserRepository) Create(ctx context.Context, user *model.User) error {
_, err := r.db.ExecContext(ctx, "INSERT ...", user.Email)
return err
}優點：
- 簡單：存儲庫會自動使用由 TransactionManager 啟動的事務
- 與存儲無關：客戶端代碼對存儲類型一無所知
缺點
- 不符合Go的使用習慣
- 控制較少：無法防止在事務中啟動事務，可能會產生意想不到的副作用，代碼審查時必須考慮到這一點
2. 事務感知存儲庫
工作原理：事務管理器啟動事務并將事務放入回調，存儲庫工廠方法使用事務創建自己。
為了啟動事務，我們需要實體：Manager
type Manager interface {
Run(
ctx context.Context,
callback func(ctx context.Context, tx *sql.Tx) error,
) error
}transaction.Manager 實現：
package transaction
import (
"context"
"database/sql"
"github.com/pkg/errors"
"go.uber.org/multierr"
)
type txKey string
var ctxWithTx = txKey("tx")
type SQLTransactionManager struct {
db *sql.DB
}
func NewManager(db *sql.DB) *SQLTransactionManager {
return &SQLTransactionManager{db: db}
}
func (m *SQLTransactionManager) Run(
ctx context.Context,
callback func(ctx context.Context, tx *sql.Tx) error,
) (rErr error) {
tx, err := m.db.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
return errors.WithStack(err)
}
defer func() {
if rErr != nil {
rErr = multierr.Combine(rErr, errors.WithStack(tx.Rollback()))
}
}()
defer func() {
if rec := recover(); rec != nil {
if e, ok := rec.(error); ok {
rErr = e
} else {
rErr = errors.Errorf("%s", rec)
}
}
}()
if err = callback(ctx, tx); err != nil {
return err
}
return errors.WithStack(tx.Commit())
}DB實現：
package storage
import (
"context"
"database/sql"
)
type DB struct {
db *sql.DB
}
func NewDB(db *sql.DB) *DB {
return &DB{db: db}
}
func (d *DB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
return d.db.QueryRowContext(ctx, query, args...)
}
func (d *DB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
return d.db.QueryContext(ctx, query, args...)
}
func (d *DB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
return d.db.ExecContext(ctx, query, args...)
}
func (d *DB) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
return d.db.PrepareContext(ctx, query)
}RegistrationService 負責用戶注冊業務場景
有兩種方法可以創建帶有事務的存儲庫：
- 存儲庫帶有結構方法 WithTransaction（示例中使用了該方法）
- 存儲庫工廠 userRepositoryFactory.CreateFromTransaction(tx)
package service
import (
"brand/transaction/example2/model"
"brand/transaction/example2/transaction"
"context"
"database/sql"
)
type UserRepository interface {
Create(ctx context.Context, user *model.User) error
WithTransaction(tx *sql.Tx) UserRepository
}
type ProfileRepository interface {
Create(ctx context.Context, user *model.Profile) error
WithTransaction(tx *sql.Tx) ProfileRepository
}
type RegistrationData struct {
Email string
Name string
}
type RegistrationService struct {
transactionManager transaction.Manager
userRepository UserRepository
profileRepository ProfileRepository
}
func NewRegistrationService(
transactionManager transaction.Manager,
userRepository UserRepository,
profileRepository ProfileRepository,
) *RegistrationService {
return &RegistrationService{
transactionManager: transactionManager,
userRepository: userRepository,
profileRepository: profileRepository,
}
}
func (s *RegistrationService) Register(ctx context.Context, data RegistrationData) error {
return s.transactionManager.Run(ctx, func(ctx context.Context, tx *sql.Tx) error {
userRepository := s.userRepository.WithTransaction(tx)
profileRepository := s.profileRepository.WithTransaction(tx)
if err := userRepository.Create(ctx, &model.User{
Email: data.Email,
}); err != nil {
return err
}
if err := profileRepository.Create(ctx, &model.Profile{
Name: data.Name,
}); err != nil {
return err
}
return nil
})
}User和ProfileRepository的實現：
package storage
import (
"brand/transaction"
"brand/transaction/example2/model"
"brand/transaction/example2/service"
"context"
"database/sql"
)
type ProfileRepository struct {
db transaction.DB
}
func NewProfileRepository(db transaction.DB) *ProfileRepository {
return &ProfileRepository{db: db}
}
func (r *ProfileRepository) Create(ctx context.Context, profile *model.Profile) error {
_, err := r.db.ExecContext(ctx, "INSERT ...", profile.Name)
return err
}
func (r *ProfileRepository) WithTransaction(tx *sql.Tx) service.ProfileRepository {
return NewProfileRepository(tx)
}package storage
import (
"brand/transaction"
"brand/transaction/example2/model"
"brand/transaction/example2/service"
"context"
"database/sql"
)
type UserRepository struct {
db transaction.DB
}
func NewUserRepository(db transaction.DB) *UserRepository {
return &UserRepository{db: db}
}
func (r *UserRepository) Create(ctx context.Context, user *model.User) error {
_, err := r.db.ExecContext(ctx, "INSERT ...", user.Email)
return err
}
func (r *UserRepository) WithTransaction(tx *sql.Tx) service.UserRepository {
return NewUserRepository(tx)
}優點：
- 更明確：在注冊服務內部創建事務，可避免副作用
缺點：
- 客戶端代碼知道存儲類型
- 客戶端代碼負責創建新的存儲庫
我相信任何一種方法都能使代碼更易讀、更簡單，但建議使用第一種方法，從而可以隱藏存儲細節，使我們能夠在一個項目中使用多個存儲，而無需考慮實現和存儲細節。
package storage
import (
"brand/transaction"
"brand/transaction/example2/model"
"brand/transaction/example2/service"
"context"
"database/sql"
)
type UserRepository struct {
db transaction.DB
}
func NewUserRepository(db transaction.DB) *UserRepository {
return &UserRepository{db: db}
}
func (r *UserRepository) Create(ctx context.Context, user *model.User) error {
_, err := r.db.ExecContext(ctx, "INSERT ...", user.Email)
return err
}
func (r *UserRepository) WithTransaction(tx *sql.Tx) service.UserRepository {
return NewUserRepository(tx)
}責任編輯：趙寧寧
來源：
DeepNoMind
