回滚在使用Go语言的事务包装器时效果不佳。

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

Rollback does not work well with Go language transactional wrapper

问题

我最近开始学习Go语言。

我找到了一个用于数据库事务处理的包装器的Github实现，并决定尝试一下。

（源代码）https://github.com/oreilly-japan/practical-go-programming/blob/master/ch09/transaction/wrapper/main.go

我正在使用PostgreSQL作为数据库。

最初，数据库中包含以下数据。

  1. testdb=> select * from products;
  2.  product_id | price
  3. ------------+-------
  4.  0001       |   200
  5.  0002       |   100
  6.  0003       |   150
  7.  0004       |   300
  8. (4 rows)

在A进程成功后，故意使B进程失败，并期望回滚A的事务。然而，当我们运行它时，回滚没有发生，结果如下所示：

事实上，由于B失败，进程A应该被回滚，数据库值不应发生任何更改。

我在代码中插入了日志以确认这一点，但我不确定为什么回滚没有执行？

请告诉我如何使用这个包装器来正确处理事务。

PS. 没有包装器的以下代码可以正常工作。

  1. package main
  3. import (
  4. 	"context"
  5. 	"database/sql"
  6. 	"log"
  8. 	_ "github.com/jackc/pgx/v4/stdlib"
  9. )
  11. type Service struct {
  12. 	db *sql.DB
  13. }
  15. func (s *Service) UpdateProduct(ctx context.Context, productID string) (err error) {
  16. 	tx, err := s.db.Begin()
  17. 	if err != nil {
  18. 		return err
  19. 	}
  20. 	defer tx.Rollback()
  22. 	if _, err = tx.ExecContext(ctx, "UPDATE products SET price = 200 WHERE product_id = $1", productID); err != nil {
  23. 		log.Println("update err")
  24. 		return err
  25. 	}
  27. 	if _, err = tx.ExecContext(ctx, "...", productID); err != nil {
  28. 		log.Println("update err")
  29. 		return err
  30. 	}
  32. 	return tx.Commit()
  33. }
  35. func main() {
  36. 	var database Service
  37. 	dbConn, err := sql.Open("pgx", "host=localhost port=5432 user=testuser dbname=testdb password=passs sslmode=disable")
  38. 	if nil != err {
  39. 		log.Fatal(err)
  40. 	}
  41. 	database.db = dbConn
  43. 	ctx := context.Background()
  45. 	database.UpdateProduct(ctx, "0004")
  46. }
英文:

I have recently started learning Go.

I found the following Github implementation of a wrapper for database transaction processing and decided to try it out.

(source) https://github.com/oreilly-japan/practical-go-programming/blob/master/ch09/transaction/wrapper/main.go

I am using PostgreSQL as the database.

Initially, it contains the following data.

  1. testdb=> select * from products;
  2.  product_id | price
  3. ------------+-------
  4.  0001       |   200
  5.  0002       |   100
  6.  0003       |   150
  7.  0004       |   300
  8. (4 rows)

After Process A succeeds, Process B is intentionally made to fail, and a rollback of transaction A is expected. However, when we run it, the rollback does not occur and we end up with the following

In truth, since B failed, the process A should be rolled back and there should be no change in the database value.

I have inserted Logs in places to confirm this, but I am not sure. Why is the rollback not executed?

  1. package main
  3. import (
  4. 	"context"
  5. 	"database/sql"
  6. 	"fmt"
  7. 	"log"
  9. 	_ "github.com/jackc/pgx/v4/stdlib"
  10. )
  12. // transaction-wrapper-start
  13. type txAdmin struct {
  14. 	*sql.DB
  15. }
  17. type Service struct {
  18. 	tx txAdmin
  19. }
  21. func (t *txAdmin) Transaction(ctx context.Context, f func(ctx context.Context) (err error)) error {
  22. 	log.Printf("transaction")
  23. 	tx, err := t.BeginTx(ctx, nil)
  24. 	if err != nil {
  25. 		return err
  26. 	}
  27. 	defer tx.Rollback()
  28. 	if err := f(ctx); err != nil {
  29. 		log.Printf("transaction err")
  30. 		return fmt.Errorf("transaction query failed: %w", err)
  31. 	}
  33. 	log.Printf("commit")
  34. 	return tx.Commit()
  35. }
  37. func (s *Service) UpdateProduct(ctx context.Context, productID string) error {
  38. 	updateFunc := func(ctx context.Context) error {
  39. 		log.Printf("first process")
  40.         // Process A
  41. 		if _, err := s.tx.ExecContext(ctx, "UPDATE products SET price = 200 WHERE product_id = $1", productID); err != nil {
  42. 			log.Printf("first err")
  43. 			return err
  44. 		}
  45. 		log.Printf("second process")
  47.         // Process B(They are intentionally failing.)
  48. 		if _, err := s.tx.ExecContext(ctx, "...", productID); err != nil {
  49. 			log.Printf("second err")
  50. 			return err
  51. 		}
  52. 		return nil
  53. 	}
  54. 	log.Printf("update")
  55. 	return s.tx.Transaction(ctx, updateFunc)
  56. }
  58. // transaction-wrapper-end
  59. func main() {
  61. 	data, err := sql.Open("pgx", "host=localhost port=5432 user=testuser dbname=testdb password=password sslmode=disable")
  62. 	if nil != err {
  63. 		log.Fatal(err)
  64. 	}
  66. 	database := Service {tx: txAdmin{data}}
  68. 	ctx := context.Background()
  70. 	database.UpdateProduct(ctx, "0004")
  71. }

output

  1. 2022/05/26 13:28:55 update     
  2. 2022/05/26 13:28:55 transaction
  3. 2022/05/26 13:28:55 first process
  4. 2022/05/26 13:28:55 second process
  5. 2022/05/26 13:28:55 second err
  6. 2022/05/26 13:28:55 transaction err

database changes(If the rollback works, the PRICE for id 0004 should remain 300.)

  1. testdb=> select * from products;
  2.  product_id | price
  3. ------------+-------
  4.  0001       |   200
  5.  0002       |   100
  6.  0003       |   150
  7.  0004       |   200
  8. (4 rows)

Please tell me how I can use the wrapper to correctly process transactions.

=========
PS.
The following code without the wrapper worked properly.

  1. package main
  3. import (
  4. 	"context"
  5. 	"database/sql"
  6. 	"log"
  8. 	_ "github.com/jackc/pgx/v4/stdlib"
  9. )
  11. // transaction-defer-start
  12. type Service struct {
  13. 	db *sql.DB
  14. }
  16. func (s *Service) UpdateProduct(ctx context.Context, productID string) (err error) {
  17. 	tx, err := s.db.Begin()
  18. 	if err != nil {
  19. 		return err
  20. 	}
  21. 	defer tx.Rollback()
  23. 	if _, err = tx.ExecContext(ctx, "UPDATE products SET price = 200 WHERE product_id = $1", productID); err != nil {
  24. 		log.Println("update err")
  25. 		return err
  26. 	}
  28. 	if _, err = tx.ExecContext(ctx, "...", productID); err != nil {
  29. 		log.Println("update err")
  30. 		return err
  31. 	}
  33. 	return tx.Commit()
  34. }
  36. // transaction-defer-end
  37. func main() {
  38. 	var database Service
  39. 	dbConn, err := sql.Open("pgx", "host=localhost port=5432 user=testuser dbname=testdb password=passs sslmode=disable")
  40. 	if nil != err {
  41. 		log.Fatal(err)
  42. 	}
  43. 	database.db = dbConn
  45. 	ctx := context.Background()
  47. 	database.UpdateProduct(ctx, "0004")
  48. 	
  49. }

答案1

得分: 0

根据@Richard Huxton的说法，将tx传递给函数f。

以下是步骤：

  1. 在struct txAdmin上添加一个字段以容纳*sql.Tx，因此txAdmin具有DB和Tx字段。
  2. 在Transaction中将tx设置为*txAdmin.Tx。
  3. 在UpdateProduct中，对每个查询使用*Service.tx.Tx。

因此，最终的代码如下所示：

  1. package main
  3. import (
  4. 	"context"
  5. 	"database/sql"
  6. 	"fmt"
  7. 	"log"
  9. 	_ "github.com/jackc/pgx/v4/stdlib"
  10. )
  12. // transaction-wrapper-start
  13. type txAdmin struct {
  14. 	*sql.DB
  15. 	*sql.Tx
  16. }
  18. type Service struct {
  19. 	tx txAdmin
  20. }
  22. func (t *txAdmin) Transaction(ctx context.Context, f func(ctx context.Context) (err error)) error {
  23. 	log.Printf("transaction")
  24. 	tx, err := t.DB.BeginTx(ctx, nil)
  25. 	if err != nil {
  26. 		return err
  27. 	}
  29. 	// 将tx设置为Tx
  30. 	t.Tx = tx
  32. 	defer tx.Rollback()
  33. 	if err := f(ctx); err != nil {
  34. 		log.Printf("transaction err")
  35. 		return fmt.Errorf("transaction query failed: %w", err)
  36. 	}
  38. 	log.Printf("commit")
  39. 	return tx.Commit()
  40. }
  42. func (s *Service) UpdateProduct(ctx context.Context, productID string) error {
  43. 	updateFunc := func(ctx context.Context) error {
  44. 		log.Printf("first process")
  45. 		// Process A
  46. 		if _, err := s.tx.Tx.ExecContext(ctx, "UPDATE products SET price = 200 WHERE product_id = $1", productID); err != nil {
  47. 			log.Printf("first err")
  48. 			return err
  49. 		}
  50. 		log.Printf("second process")
  52. 		// Process B(They are intentionally failing.)
  53. 		if _, err := s.tx.Tx.ExecContext(ctx, "...", productID); err != nil {
  54. 			log.Printf("second err")
  55. 			return err
  56. 		}
  57. 		return nil
  58. 	}
  59. 	log.Printf("update")
  60. 	return s.tx.Transaction(ctx, updateFunc)
  61. }
  63. // transaction-wrapper-end
  64. func main() {
  65. 	data, err := sql.Open("pgx", "host=localhost port=5432 user=testuser dbname=testdb password=password sslmode=disable")
  66. 	if nil != err {
  67. 		log.Fatal(err)
  68. 	}
  70. 	database := Service{tx: txAdmin{DB: data}}
  72. 	ctx := context.Background()
  74. 	database.UpdateProduct(ctx, "0004")
  75. }
英文:

As @Richard Huxton said, pass tx into a function f

here are the steps:

  1. add one field on struct txAdmin to accommodate *sql.Tx, so txAdmin have DB and Tx fields
  2. inside Transaction set tx to *txAdmin.Tx
  3. inside UpdateProduct use *Service.tx.Tx for every query

so the final code looks like this:

  1. package main
  3. import (
  4. 	"context"
  5. 	"database/sql"
  6. 	"fmt"
  7. 	"log"
  9. 	_ "github.com/jackc/pgx/v4/stdlib"
  10. )
  12. // transaction-wrapper-start
  13. type txAdmin struct {
  14. 	*sql.DB
  15. 	*sql.Tx
  16. }
  18. type Service struct {
  19. 	tx txAdmin
  20. }
  22. func (t *txAdmin) Transaction(ctx context.Context, f func(ctx context.Context) (err error)) error {
  23. 	log.Printf("transaction")
  24. 	tx, err := t.DB.BeginTx(ctx, nil)
  25. 	if err != nil {
  26. 		return err
  27. 	}
  29.     // set tx to Tx
  30. 	t.Tx = tx
  32. 	defer tx.Rollback()
  33. 	if err := f(ctx); err != nil {
  34. 		log.Printf("transaction err")
  35. 		return fmt.Errorf("transaction query failed: %w", err)
  36. 	}
  38. 	log.Printf("commit")
  39. 	return tx.Commit()
  40. }
  42. func (s *Service) UpdateProduct(ctx context.Context, productID string) error {
  43. 	updateFunc := func(ctx context.Context) error {
  44. 		log.Printf("first process")
  45. 		// Process A
  46. 		if _, err := s.tx.Tx.ExecContext(ctx, "UPDATE products SET price = 200 WHERE product_id = $1", productID); err != nil {
  47. 			log.Printf("first err")
  48. 			return err
  49. 		}
  50. 		log.Printf("second process")
  52. 		// Process B(They are intentionally failing.)
  53. 		if _, err := s.tx.Tx.ExecContext(ctx, "...", productID); err != nil {
  54. 			log.Printf("second err")
  55. 			return err
  56. 		}
  57. 		return nil
  58. 	}
  59. 	log.Printf("update")
  60. 	return s.tx.Transaction(ctx, updateFunc)
  61. }
  63. // transaction-wrapper-end
  64. func main() {
  65. 	data, err := sql.Open("pgx", "host=localhost port=5432 user=testuser dbname=testdb password=password sslmode=disable")
  66. 	if nil != err {
  67. 		log.Fatal(err)
  68. 	}
  70. 	database := Service{tx: txAdmin{DB: data}}
  72. 	ctx := context.Background()
  74. 	database.UpdateProduct(ctx, "0004")
  75. }

huangapple
  • 本文由 发表于 2022年5月26日 12:47:05
  • 转载请务必保留本文链接：https://go.coder-hub.com/72386821.html
  • go
  • postgresql
匿名

发表评论

匿名网友

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

确定