9.8 KiB
9.8 KiB
事务管理方案说明
概述
本方案通过Context传递GORM事务对象,实现简单直接的事务管理,避免了复杂的Saga分布式事务框架。所有事务相关功能统一在 shared/database 包中管理。
架构设计
分层职责
┌─────────────────────────────────────────────────────────────┐
│ 应用服务层 (Application) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 使用 TransactionManager.ExecuteInTx() 进行事务管理 │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 共享层 (Shared) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ shared/database/transaction.go │ │
│ │ - TransactionManager (事务管理器) │ │
│ │ - Context传递机制 │ │
│ │ - 事务选项和统计 │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 基础设施层 (Infrastructure) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ infrastructure/database/database.go │ │
│ │ - 数据库连接管理 │ │
│ │ - 连接池配置 │ │
│ │ - 基础数据库操作 │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
核心组件
1. 事务管理器 (TransactionManager)
// 位置: internal/shared/database/transaction.go
type TransactionManager struct {
db *gorm.DB
}
// 主要方法:
// - ExecuteInTx() - 推荐使用的事务执行方法
// - ExecuteInTxWithTimeout() - 带超时的事务执行
// - ExecuteInTxWithOptions() - 带选项的事务执行
// - BeginTx() - 手动开始事务
// - NewTxWrapper() - 创建事务包装器
2. Context工具函数
// WithTx 将事务对象存储到context中
func WithTx(ctx context.Context, tx *gorm.DB) context.Context
// GetTx 从context中获取事务对象
func GetTx(ctx context.Context) (*gorm.DB, bool)
3. 仓储层支持
所有GORM仓储实现都添加了getDB方法:
// getDB 获取数据库连接,优先使用事务
func (r *GormEnterpriseInfoSubmitRecordRepository) getDB(ctx context.Context) *gorm.DB {
if tx, ok := database.GetTx(ctx); ok {
return tx
}
return r.db
}
使用方式
1. 基础事务执行(推荐)
// 应用服务层
func (s *CertificationApplicationServiceImpl) SubmitEnterpriseInfo(ctx context.Context, cmd *commands.SubmitEnterpriseInfoCommand) (*responses.EnterpriseInfoResponse, error) {
// 1. 验证企业信息
exists, err := s.enterpriseService.CheckUnifiedSocialCodeExists(ctx, cmd.UnifiedSocialCode, "")
if err != nil {
return nil, fmt.Errorf("检查企业信息失败: %w", err)
}
if exists {
return nil, fmt.Errorf("统一社会信用代码已存在")
}
// 2. 获取或创建认证申请
certification, err := s.certManagementService.GetCertificationByUserID(ctx, cmd.UserID)
if err != nil {
// 处理错误...
}
// 3. 使用事务执行状态转换和创建记录
var recordID string
var createdRecord entities.EnterpriseInfoSubmitRecord
err = s.txManager.ExecuteInTx(ctx, func(txCtx context.Context) error {
// 步骤1:创建企业信息提交记录
record := entities.NewEnterpriseInfoSubmitRecord(
certification.ID,
cmd.UserID,
cmd.CompanyName,
cmd.UnifiedSocialCode,
cmd.LegalPersonName,
cmd.LegalPersonID,
)
var err error
createdRecord, err = s.enterpriseRecordRepo.Create(txCtx, *record)
if err != nil {
return fmt.Errorf("创建企业信息提交记录失败: %w", err)
}
recordID = createdRecord.ID
// 步骤2:状态转换
err = s.certWorkflowService.SubmitEnterpriseInfo(txCtx, certification.ID)
if err != nil {
return fmt.Errorf("状态转换失败: %w", err)
}
return nil
})
if err != nil {
s.logger.Error("事务执行失败", zap.Error(err))
return nil, fmt.Errorf("企业信息提交失败: %w", err)
}
// 返回成功响应...
}
2. 带超时的事务执行
// 设置30秒超时
err = s.txManager.ExecuteInTxWithTimeout(ctx, 30*time.Second, func(txCtx context.Context) error {
// 事务操作...
return nil
})
3. 带选项的事务执行
options := &database.TransactionOptions{
Timeout: 30 * time.Second,
ReadOnly: false,
}
err = s.txManager.ExecuteInTxWithOptions(ctx, options, func(txCtx context.Context) error {
// 事务操作...
return nil
})
4. 手动事务管理(高级用法)
// 手动管理事务
txWrapper := s.txManager.NewTxWrapper()
defer func() {
if err != nil {
txWrapper.Rollback()
}
}()
// 使用事务
txCtx := database.WithTx(ctx, txWrapper.GetDB())
err = s.enterpriseRecordRepo.Create(txCtx, record)
if err != nil {
return err
}
// 提交事务
err = txWrapper.Commit()
5. 仓储层自动事务支持
仓储层会自动从context中获取事务对象:
// Create 创建企业信息提交记录
func (r *GormEnterpriseInfoSubmitRecordRepository) Create(ctx context.Context, record entities.EnterpriseInfoSubmitRecord) (entities.EnterpriseInfoSubmitRecord, error) {
r.logger.Info("创建企业信息提交记录", zap.String("certification_id", record.CertificationID))
err := r.getDB(ctx).WithContext(ctx).Create(&record).Error
return record, err
}
优势
- 统一管理: 所有事务相关功能集中在
shared/database包中 - 简单直接: 不需要复杂的状态管理和补偿逻辑
- 自动回滚: 任何步骤失败都会自动回滚整个事务
- 类型安全: 通过Context传递,类型安全
- 易于理解: 代码逻辑清晰,易于维护
- 性能好: 避免了分布式事务的开销
- 灵活配置: 支持超时、只读等选项
- 向后兼容: 保留旧接口,平滑迁移
适用场景
- 单数据库事务
- 简单的业务流程编排
- 需要原子性操作的场景
- 对性能要求较高的场景
- 需要事务超时控制的场景
注意事项
- 只适用于单数据库事务
- 不支持跨服务的分布式事务
- 需要确保所有仓储都支持事务传递
- 事务超时需要合理设置
- 避免在事务中执行长时间操作
与Saga的对比
| 特性 | 事务管理方案 | Saga方案 |
|---|---|---|
| 复杂度 | 简单 | 复杂 |
| 性能 | 高 | 中等 |
| 适用场景 | 单数据库 | 分布式 |
| 维护成本 | 低 | 高 |
| 错误处理 | 自动回滚 | 需要补偿逻辑 |
| 超时控制 | 内置支持 | 需要额外实现 |
| 统计监控 | 预留接口 | 复杂 |
迁移指南
从旧的事务管理迁移
-
替换基础设施层的事务调用:
// 旧方式 err := db.WithTx(func(tx *gorm.DB) error { // 事务操作 return nil }) // 新方式 txManager := database.NewTransactionManager(db) err := txManager.ExecuteInTx(ctx, func(txCtx context.Context) error { // 事务操作 return nil }) -
更新仓储层: 确保所有仓储都使用
getDB(ctx)方法 -
更新应用服务: 使用
TransactionManager替代直接的事务调用
未来扩展
- 事务统计: 实现事务执行统计和监控
- 分布式事务: 在需要时扩展为分布式事务支持
- 事务链路追踪: 集成链路追踪系统
- 事务重试: 添加自动重试机制