在同一事务中更新多个表格且使用不同的goroutine时出现Gorm错误。

huangapple
go评论119阅读模式
英文:

Gorm error when updating multiple tables in the same transaction and different goroutines

问题

我有这段代码示例：

  1. err = transaction.WithTransaction(context.Background(), func(txCtx context.Context) error {
  2.   errorGroup := &errgroup.Group{}
  3.   errorGroup.Go(func() error {
  4.     return s.addTotable1(txCtx, *model)
  5.   })
  6.   errorGroup.Go(func() error {
  7.     return s.updateTable1(txCtx, *model)
  8.   })
  9.   errorGroup.Go(func() error {
  10.     return s.updateTable2(txCtx, *model)
  11.   })
  12.   errorGroup.Go(func() error {
  13.     return s.updateTable3(txCtx, *model)
  14.   })
  15.   errorGroup.Go(func() error {
  16.     return s.updateTable4(txCtx, *model)
  17.   })
  18.   errorGroup.Go(func() error {
  19.     return s.updateTable5(txCtx, *model)
  20.   })
  21.   if err := errorGroup.Wait(); err != nil {
  22.     transactionError = err
  23.     return err
  24.   }
  25. }, func(trCtx context.Context) error {
  26.   return transactionError
  27. })

WithTransaction 方法在这里定义：

  1. type txKey struct{}
  3. func injectTx(ctx context.Context, tx *gorm.DB) context.Context {
  4.    return context.WithValue(ctx, txKey{}, tx)
  5. }
  7. func ExtractTx(ctx context.Context) *gorm.DB {
  8.    if tx, ok := ctx.Value(txKey{}).(*gorm.DB); ok {
  9.       return tx
  10.    }
  11.    return nil
  12. }
  14. func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error) error {
  15.    gormConnect := postgresGorm.DbConnection
  17.    if gormConnect == nil {
  18.       return commonerrors.InternalServerError{
  19.          ErrorResponse: commonerrors.ErrorResponse{
  20.             Message: "error",
  21.          },
  22.       }
  23.    }
  25.    tx := gormConnect.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  27.    if err := tx.Error; err != nil {
  28.       return err
  29.    }
  31.    err := txFunc(injectTx(ctx, tx))
  33.    if err != nil {
  34.       tx.Rollback()
  35.       tx = postgresGorm.DbConnection.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  36.       err = trFunc(injectTx(ctx, tx))
  37.       tx.Commit()
  38.       return err
  39.    }
  41.    tx.Commit()
  43.    return nil
  44. }

还有我们的 Gorm 配置与 PostgresDB：

  1. var (
  2.    DbConnection *gorm.DB = nil
  3. )
  5. const (
  6.    connectionFailedMsg = "postgres-gorm connection failed: %s"
  7. )
  9. // OpenConnection open postgres connection
  10. func OpenConnection() {
  12.    // PostgreSQL Connection, uncomment to use.
  13.    // connection string format: user=USER password=PASSWORD host=/cloudsql/PROJECT_ID:REGION_ID:INSTANCE_ID/[ dbname=DB_NAME]
  14.    dbURI := fmt.Sprintf("host=%s%s port=%d user=%s "+
  15.       "password=%s dbname=%s sslmode=disable",
  16.       configs.PostgresqlGormConfigs.CloudSqlPrefix, configs.PostgresqlGormConfigs.Host,
  17.       configs.PostgresqlGormConfigs.Port, configs.PostgresqlGormConfigs.User,
  18.       configs.PostgresqlGormConfigs.Password, configs.PostgresqlGormConfigs.DbName)
  19.    config := &gorm.Config{
  20.       NamingStrategy: schema.NamingStrategy{
  21.          TablePrefix:   configs.PostgresqlGormConfigs.TableGormPrefix,
  22.          SingularTable: true,
  23.       }}
  25.    var err error
  27.    DbConnection, err = gorm.Open(postgres.Open(dbURI), config)
  29.    if err != nil {
  30.       panic(err)
  31.    }
  33.    sqlDB, err := DbConnection.DB()
  35.    if err != nil {
  36.       log.Errorf(connectionFailedMsg, err)
  37.       panic(err)
  38.    }
  40.    if configs.PostgresqlGormConfigs.GormLoggin {
  41.       DbConnection.Config.Logger = gormLogger.Default.LogMode(gormLogger.Info)
  42.    }
  44.    err = sqlDB.Ping()
  46.    if err != nil {
  47.       log.Errorf(connectionFailedMsg, err)
  48.    } else {
  49.       log.Info("postgres-gorm connection successfully established")
  50.    }
  51. }

这是使用 Gorm 更新表的方法示例：

  1. func (o ServiceImpl) UpdateTable1(ctx context.Context, model *model) (*model, error) {
  3.    tx := transaction.ExtractTx(ctx)
  4.    injectedTransaction := true
  6.    if tx == nil {
  7.       tx = postgresGorm.DbConnection.Begin()
  8.       injectedTransaction = false
  9.    }
  11.    //Result
  12.    queryResult := tx.Save(&model)
  14.    // Error
  15.    if queryResult.Error != nil {
  16.       if !injectedTransaction {
  17.          tx.Rollback()
  18.       }
  19.       errResp := commonerrors.ErrorResponse{
  20.          Code:    "500",
  21.          Message: "Error",
  22.       }
  23.       return nil, commonerrors.InternalServerError{ErrorResponse: errResp}
  24.    }
  26.    if !injectedTransaction {
  27.       tx.Commit()
  28.    }
  29.    return shipDetail, nil
  30. }

我们的问题是，在使用此服务时，存在一些与多个 goroutine 相关的问题，我们随机地遇到以下错误：driver: bad connection，但这完全是随机的，第一次尝试总是成功的，之后会失败一次，然后再次成功...你明白我的意思。

我们已经尝试升级到最新版本的 gorm 和 gorm postgres 驱动程序，但没有任何改变。通过阅读 gorm 文档，我们正在使用的所有方法都应该是线程安全的，所以我现在有些困惑。如果我找到任何解决方法，我会更新帖子。谢谢。

英文:

I've got this sample of code

  1. err = transaction.WithTransaction(context.Background(), func(txCtx context.Context) error {
  2.   errorGroup := &errgroup.Group{}
  3.   errorGroup.Go(func() error {
  4.     return s.addTotable1(txCtx, *model)
  5.   })
  6.   errorGroup.Go(func() error {
  7.     return s.updateTable1(txCtx, *model)
  8.   })
  9.   errorGroup.Go(func() error {
  10.     return s.updateTable2(txCtx, *model)
  11.   })
  12.   errorGroup.Go(func() error {
  13.     return s.updateTable3(txCtx, *model)
  14.   })
  15.   errorGroup.Go(func() error {
  16.     return s.updateTable4(txCtx, *model)
  17.   })
  18.   errorGroup.Go(func() error {
  19.     return s.updateTable5(txCtx, *model)
  20.   })
  21.   if err := errorGroup.Wait(); err != nil {
  22.     transactionError = err
  23.     return err
  24.   }
  25.   }, func(trCtx context.Context) error {
  26.     return transactionError
  27.   })

WithTransaction method is defined here

  1. type txKey struct{}
  3. func injectTx(ctx context.Context, tx *gorm.DB) context.Context {
  4.    return context.WithValue(ctx, txKey{}, tx)
  5. }
  7. func ExtractTx(ctx context.Context) *gorm.DB {
  8.    if tx, ok := ctx.Value(txKey{}).(*gorm.DB); ok {
  9.       return tx
  10.    }
  11.    return nil
  12. }
  1. func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error) error {
  2.    gormConnect := postgresGorm.DbConnection
  4.    if gormConnect == nil {
  5.       return commonerrors.InternalServerError{
  6.          ErrorResponse: commonerrors.ErrorResponse{
  7.             Message: "error",
  8.          },
  9.       }
  10.    }
  12.    tx := gormConnect.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  14.    if err := tx.Error; err != nil {
  15.       return err
  16.    }
  18.    err := txFunc(injectTx(ctx, tx))
  20.    if err != nil {
  21.       tx.Rollback()
  22.       tx = postgresGorm.DbConnection.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  23.       err = trFunc(injectTx(ctx, tx))
  24.       tx.Commit()
  25.       return err
  26.    }
  28.    tx.Commit()
  30.    return nil
  32. }

Also Gorm configuration with our PostgresDB

  1. var (
  2.    DbConnection *gorm.DB = nil
  3. )
  5. const (
  6.    connectionFailedMsg = "postgres-gorm connection failed: %s"
  7. )
  9. // OpenConnection open postgres connection
  10. func OpenConnection() {
  12.    // PostgreSQL Connection, uncomment to use.
  13.    // connection string format: user=USER password=PASSWORD host=/cloudsql/PROJECT_ID:REGION_ID:INSTANCE_ID/[ dbname=DB_NAME]
  14.    dbURI := fmt.Sprintf("host=%s%s port=%d user=%s "+"password=%s dbname=%s sslmode=disable",
  15.       configs.PostgresqlGormConfigs.CloudSqlPrefix, configs.PostgresqlGormConfigs.Host,
  16.       configs.PostgresqlGormConfigs.Port, configs.PostgresqlGormConfigs.User,
  17.       configs.PostgresqlGormConfigs.Password, configs.PostgresqlGormConfigs.DbName)
  18.    config := &gorm.Config{
  19.       NamingStrategy: schema.NamingStrategy{
  20.          TablePrefix:   configs.PostgresqlGormConfigs.TableGormPrefix,
  21.          SingularTable: true,
  22.       }}
  24.    var err error
  26.    DbConnection, err = gorm.Open(postgres.Open(dbURI), config)
  28.    if err != nil {
  29.       panic(err)
  30.    }
  32.    sqlDB, err := DbConnection.DB()
  34.    if err != nil {
  35.       log.Errorf(connectionFailedMsg, err)
  36.       panic(err)
  37.    }
  39.    if configs.PostgresqlGormConfigs.GormLoggin {
  40.       DbConnection.Config.Logger = gormLogger.Default.LogMode(gormLogger.Info)
  41.    }
  43.    err = sqlDB.Ping()
  45.    if err != nil {
  46.       log.Errorf(connectionFailedMsg, err)
  47.    } else {
  48.       log.Info("postgres-gorm connection successfully established")
  50.    }
  51. }

And here an example of the method to update a table using Gorm

  1. func (o ServiceImpl) UpdateTable1(ctx context.Context, model *model) (*model, error) {
  3.    tx := transaction.ExtractTx(ctx)
  4.    injectedTransaction := true
  6.    if tx == nil {
  7.       tx = postgresGorm.DbConnection.Begin()
  8.       injectedTransaction = false
  9.    }
  11.    //Result
  12.    queryResult := tx.Save(&model)
  14.    // Error
  15.    if queryResult.Error != nil {
  16.       if !injectedTransaction {
  17.          tx.Rollback()
  18.       }
  19.       errResp := commonerrors.ErrorResponse{
  20.          Code:    "500",
  21.          Message: "Error",
  22.       }
  23.       return nil, commonerrors.InternalServerError{ErrorResponse: errResp}
  24.    }
  26.    if !injectedTransaction {
  27.       tx.Commit()
  28.    }
  29.    return shipDetail, nil
  30. }

Our problem is that something is going on with multiple goroutines and we are getting this error randomly when using this service: driver: bad connection But it is completly random, first try always is successful and after that it fails once, then success again... You get the idea.

We've tried upgrading to the latest version of gorm and gorm postgres driver but it didnt change a thing. Reading through gorm doc, all the methods we are using should be thread safe, so Im kinda stuck right now. If I find any fix to this I will update the post. Thanks.

答案1

得分: 1

我将为您翻译以下内容：

我在这里发布我们目前正在使用的答案，以防将来有人遇到这个问题。主要问题是在所有线程中重用相同的gorm.Session，所以我创建了一个通用的解决方法：

首先对WithTransaction函数进行轻微修改：

  1. func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error, nestedTx bool) error {
  2.     gormConnect := postgresGorm.DbConnection
  4.     if gormConnect == nil {
  5.         return commonerrors.InternalServerError{
  6.             ErrorResponse: commonerrors.ErrorResponse{
  7.                 Message: "Error intentando conseguir la conexión con bbdd",
  8.             },
  9.         }
  10.     }
  12.     tx := gormConnect.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  14.     if err := tx.Error; err != nil {
  15.         return err
  16.     }
  18.     err := txFunc(injectTx(ctx, tx))
  20.     if err != nil {
  21.         tx.Rollback()
  22.         tx = postgresGorm.DbConnection.Session(&gorm.Session{SkipDefaultTransaction: true}).Begin()
  23.         err = trFunc(injectTx(ctx, tx))
  24.         tx.Commit()
  25.         return err
  26.     }
  28.     if !nestedTx {
  29.         tx.Commit()
  30.     }
  32.     return nil
  33. }

现在，如果我们有嵌套事务，它将不会提交。然后，我添加了以下函数来处理errorGroup并为每个Goroutine创建一个新的会话：

  1. func RoutineTransaction(errorGroup *errgroup.Group, transactions chan<- *gorm.DB, subroutineFuncTx func(txCtx context.Context) error) {
  2.     errorGroup.Go(func() error {
  3.         var routineTxErr error
  4.         routineTxErr = WithTransaction(context.Background(), func(txCtx context.Context) error {
  5.             transactions <- ExtractTx(txCtx)
  6.             routineTxErr = subroutineFuncTx(txCtx)
  7.             return routineTxErr
  8.         }, func(trCtx context.Context) error {
  9.             return routineTxErr
  10.         }, true)
  11.         return routineTxErr
  12.     })
  13. }
  15. func RoutinesTransactionsCommit(transactions chan *gorm.DB) {
  16.     close(transactions)
  17.     for tx := range transactions {
  18.         tx.Commit()
  19.     }
  20. }
  22. func RoutinesTransactionsRollback(transactions chan *gorm.DB) {
  23.     close(transactions)
  24.     for tx := range transactions {
  25.         tx.Rollback()
  26.     }
  27. }

然后，您只需在需要的地方简单地使用它：

  1. errorGroup := &errgroup.Group{}
  2. transactions := make(chan *gorm.DB, 6)
  3. transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  4.     return s.addToTable1(txCtx, *model)
  5. })
  6. transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  7.     return s.updateTable1(txCtx, *model)
  8. })
  9. transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  10.     return s.updateTable2(txCtx, *model)
  11. })
  12. transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  13.     return s.updateTable3(txCtx, *model)
  14. })
  15. if err := errorGroup.Wait(); err != nil {
  16.     transaction.RoutinesTransactionsRollback(transactions)
  17.     transactionError = err
  18.     return err
  19. }
  20. transaction.RoutinesTransactionsCommit(transactions)
英文:

Im posting the answer that is currently working for us in case someone enconters this problem in the future. The main problem was reusing the same gorm.Session for all the threads, so I've created a generic workarround:

First a slight modification to the WithTransaction function

  1. func WithTransaction(ctx context.Context, txFunc func(ctx context.Context) error, trFunc func(ctx context.Context) error, nestedTx bool) error {
  2. gormConnect := postgresGorm.DbConnection
  4. if gormConnect == nil {
  5. 	return commonerrors.InternalServerError{
  6. 		ErrorResponse: commonerrors.ErrorResponse{
  7. 			Message: &quot;Error intentando conseguir la conexi&#243;n con bbdd&quot;,
  8. 		},
  9. 	}
  10. }
  12. tx := gormConnect.Session(&amp;gorm.Session{SkipDefaultTransaction: true}).Begin()
  14. if err := tx.Error; err != nil {
  15. 	return err
  16. }
  18. err := txFunc(injectTx(ctx, tx))
  20. if err != nil {
  21. 	tx.Rollback()
  22. 	tx = postgresGorm.DbConnection.Session(&amp;gorm.Session{SkipDefaultTransaction: true}).Begin()
  23. 	err = trFunc(injectTx(ctx, tx))
  24. 	tx.Commit()
  25. 	return err
  26. }
  28. if !nestedTx {
  29. 	tx.Commit()
  30. }
  32. return nil
  33. }

Now if we have nested transactions it will not commit.
Then I've added this functions to handle an errorGroup and create a new session for each Goroutine:

  1. func RoutineTransaction(errorGroup *errgroup.Group, transactions chan&lt;- *gorm.DB, subroutineFuncTx func(txCtx context.Context) error) {
  2. errorGroup.Go(func() error {
  3. 	var routineTxErr error
  4. 	routineTxErr = WithTransaction(context.Background(), func(txCtx context.Context) error {
  5. 		transactions &lt;- ExtractTx(txCtx)
  6. 		routineTxErr = subroutineFuncTx(txCtx)
  7. 		return routineTxErr
  8. 	}, func(trCtx context.Context) error {
  9. 		return routineTxErr
  10. 	}, true)
  11. 	return routineTxErr
  12. })
  15. func RoutinesTransactionsCommit(transactions chan *gorm.DB) {
  16. close(transactions)
  17. for tx := range transactions {
  18. 	tx.Commit()
  19. }
  20. func RoutinesTransactionsRollback(transactions chan *gorm.DB) {
  21. close(transactions)
  22. for tx := range transactions {
  23. 	tx.Rollback()
  24. }
  25. }

And then you just simply use it like this wherever you need it:

  1.     errorGroup := &amp;errgroup.Group{}
  2. 	transactions := make(chan *gorm.DB, 6)
  3. 	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  4. 		return s.addToTable1(txCtx, *model)
  5. 	})
  6. 	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  7. 		return s.updateTable1(txCtx, *model)
  8. 	})
  9. 	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  10. 		return s.updateTable2(txCtx, *model)
  11. 	})
  12. 	transaction.RoutineTransaction(errorGroup, transactions, func(txCtx context.Context) error {
  13. 		return s.updateTable3(txCtx, *model)
  14. 	})
  15. 	if err := errorGroup.Wait(); err != nil {
  16. 		transaction.RoutinesTransactionsRollback(transactions)
  17. 		transactionError = err
  18. 		return err
  19. 	}
  20. 	transaction.RoutinesTransactionsCommit(transactions)

huangapple
  • 本文由 发表于 2022年10月27日 04:10:08
  • 转载请务必保留本文链接：https://go.coder-hub.com/74213692.html
  • go
  • go-gorm
  • goroutine
  • postgresql
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定