保持同一行并发更新的完整性

Question

在下面的代码片段中，我尝试查找、删除和创建相同的项目，但在 2 个不同线程的 2 个不同事务中。

在线程 1 中，我创建事务 1，找到该项目并将其删除。

完成后，我允许线程 2 创建事务 2，并尝试查找该项目。该Find()方法在这里阻塞，因为我使用了选项FOR UPDATE。

回到线程 1，重新创建该项目并提交事务 1，这允许Find()线程 2 完成。以下是那里出现的问题：

如果我使用隔离级别“ReadCommissed”，我会收到未找到错误- 这对我来说没有意义，因为我认为 ReadCommited 事务可以看到其他事务应用的更新。

如果我使用隔离级别“可序列化”，则会收到错误：pq: could not serialize access due to concurrent update。

为什么我会看到这种行为？我认为在第二次查找解锁后，它应该为我提供最新的行。

如何才能使当一行正在被修改的过程中时，任何其他读取都将锁定，并在其他线程完成后解锁返回最新数据？

db, err := gorm.Open("postgres", "host=localHost port=5432 user=postgres dbname=test-rm password=postgres sslmode=disable")
if err != nil { panic("failed to connect database") }
db.SingularTable(true)
db.DropTableIfExists(&Product{})
db.AutoMigrate(&Product{})

db.Create(&Product{Code: "A", Price: 1000})
// SQL: INSERT  INTO "product" ("code","price") VALUES ('A',1000) RETURNING "products"."id"

txOpt := &sql.TxOptions{Isolation: sql.LevelSerializable}

doneTrans1 := make(chan struct{})

go func(){
    item1 := &Product{}

    tx1 := db.BeginTx(context.Background(), txOpt)

    err = tx1.Set("gorm:query_option", "FOR UPDATE").Find(item1, "code = ?", "A").Error
    // SQL: SELECT * FROM "product"  WHERE (code = 'A') FOR UPDATE

    item1.Price = 3000

    err = tx1.Delete(item1).Error
    // SQL: DELETE FROM "product"  WHERE "product"."id" = 1

    doneTrans1<-struct{}{}
    time.Sleep(time.Second * 3)

    err = tx1.Create(item1).Error
    // SQL: INSERT  INTO "product" ("id","code","price") VALUES (1,'A',3000) RETURNING "product"."id"

    tx1.Commit()
}()

// Ensure other trans work started
<-doneTrans1
time.Sleep(time.Second * 2)

item2 := &Product{}

tx2 := db.BeginTx(context.Background(), txOpt)

err = tx2.Set("gorm:query_option", "FOR UPDATE").Find(item2, "code = ?", "A").Error
// SQL: SELECT * FROM "product"  WHERE (code = 'A') FOR UPDATE
// ERROR occurs here

item2.Price = 5000
err = tx2.Delete(item2).Error
err = tx2.Create(item2).Error
tx2.Commit()
time.Sleep(time.Second * 5)

Answer 1

为了回答这个问题，我认为最好消除 goroutine 的复杂性（事实上，根本不需要 go）并专注于 SQL。以下是按运行顺序排列的 SQL 语句（我忽略了错误发生后的所有内容，因为这大多无关紧要，而且执行顺序变得复杂/可变！）。

在主程序中

INSERT  INTO "product" ("code","price") VALUES ('A',1000) RETURNING "products"."id"

在 Go 例程中

BEGIN TX1
SELECT * FROM "product"  WHERE (code = 'A') FOR UPDATE
DELETE FROM "product"  WHERE "product"."id" = 1

在主程序中

BEGIN TX2
SELECT * FROM "product"  WHERE (code = 'A') FOR UPDATE -- ERROR occurs here

回答你的问题。

问题1

如果我使用隔离级别“ReadCommissed”，我会收到未找到错误 - 这对我来说没有意义，因为我认为 ReadCommited 事务可以看到其他事务应用的更新。

来自读提交隔离级别的文档：

在搜索目标行方面，UPDATE、DELETE、SELECT FOR UPDATE 和 SELECT FOR SHARE 命令的行为与 SELECT 相同：它们只会查找截至命令开始时间已提交的目标行。然而，这样的目标行在被发现时可能已经被另一个并发事务更新（或删除或锁定）。在这种情况下，潜在的更新程序将等待第一个更新事务提交或回滚（如果仍在进行中）。如果第一个更新程序回滚，则其效果将被否定，第二个更新程序可以继续更新最初找到的行。如果第一个更新程序提交，则第二个更新程序将忽略第一个更新程序删除的行，否则它将尝试将其操作应用于该行的更新版本。

所以SELECT * FROM "product" WHERE (code = 'A') FOR UPDATETX2中的会等待TX1完成。此时 TX1 已删除产品 A，因此该行被忽略并且不返回任何结果。现在我知道 TX1 也重新创建了产品 A，但请记住“SELECT 查询（没有 FOR UPDATE/SHARE 子句）只能看到查询开始之前提交的数据；” 并且由于选择在 TX1 重新创建记录之前开始，因此不会被看到。

问题2

如果我使用隔离级别“可序列化”，则会收到错误：pq：由于并发更新而无法序列化访问。

从可重复读隔离级别的文档（可串行化是更高的级别，因此这些规则以及一些更严格的规则适用）：

在搜索目标行方面，UPDATE、DELETE、SELECT FOR UPDATE 和 SELECT FOR SHARE 命令的行为与 SELECT 相同：它们只会查找截至事务开始时间已提交的目标行。然而，这样的目标行在被发现时可能已经被另一个并发事务更新（或删除或锁定）。在这种情况下，可重复读事务将等待第一个更新事务提交或回滚（如果仍在进行中）。如果第一个更新程序回滚，则其影响被否定，并且可重复读取事务可以继续更新最初找到的行。但是，如果第一个更新程序提交（并且实际更新或删除了该行，而不仅仅是锁定它），则可重复读取事务将回滚并显示以下消息

在您的代码中，TX1 更新产品 A，这意味着 TX2 中的查询将被延迟，直到 TX1 提交，此时它将因错误而中止（如果 TX1 回滚，则它将继续）。

我怎样才能进行第二次更新？*

维护事务完整性是一个难题，PostgreSQL 中的功能是一些非常聪明的人大量工作的结果。如果您发现自己与数据库作斗争，通常最好退后一步，考虑是否需要改变方法（或者您认为的问题是否是一个真正的问题）。

在您的示例中，您有两个例程删除并重新创建相同的记录；我无法预见您希望两项交易都继续进行的情况。在可能发生这种情况的真实系统中，您不会仔细安排计时器来确保首先启动一个事务。这意味着事务完成后数据库的状态将取决于哪个事务到达第一个事务SELECT * FROM "product" WHERE (code = 'A') FOR UPDATE。因此，实际上，如果失败并不重要（因为无论如何结果都是随机的）；这实际上是一个更好的结果，因为您可以建议用户（他们可以检查记录并在需要时重新运行任务）。

因此，在阅读本文的其余部分之前，我建议您考虑一下这是否是一个问题（我对您想要实现的目标没有背景，因此很难发表评论）。

如果您确实想确保更新继续进行，您有以下几种选择：

• 如果使用“可序列化”，您需要检测故障并重试事务（如果这是业务逻辑的要求）
• 如果使用“已提交读”，则用 UPDATE 替换 DELETE/INSERT（在这种情况下，当第一个事务锁被释放时，PostgreSQL 将重新评估 WHERE 子句）。

然而，我认为更好的方法是消除大部分这种情况，并尝试在一个步骤中执行这样的更新（这可能意味着绕过 ORM）。如果您想最大限度地减少出现此类问题的可能性，那么最大限度地减少锁定的数量/持续时间非常重要，并且在单个步骤中执行操作会带来很大帮助。对于复杂的操作，使用存储过程可以加快速度，但仍然（减少）与其他同时运行的操作发生冲突的机会。

您可能还想看看乐观锁定，因为在某些情况下这更有意义（例如，您读取信息，将其显示给用户并等待更改，但与此同时另一个用户可能已进行更改）。