维持一致性并同步更新

一分耕耘，一分收获！既然都打开这篇《维持一致性并同步更新》，就坚持看下去，学下去吧！本文主要会给大家讲到等等知识点，如果大家对本文有好的建议或者看到有不足之处，非常欢迎大家积极提出！在后续文章我会继续更新Golang相关的内容，希望对大家都有所帮助！

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

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

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

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

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

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

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

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

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)

解决方案

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

在主例程中

insert  into "product" ("code","price") values ('a',1000) returning "products"."id"

在goroutine中

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”，我会收到未找到错误 - 这对我来说没有意义，因为我认为 readcomlated 事务可以看到其他人应用的更新。

来自 Read Committed Isolation Level 的文档：

更新、删除、选择更新和选择共享命令 在搜索目标行方面与 select 的行为相同：它们 将仅查找自命令开始时提交的目标行 时间。然而，这样的目标行可能已经被更新（或者 被另一个并发事务删除或锁定） 成立。在这种情况下，潜在的更新程序将等待第一个 更新事务以提交或回滚（如果它仍在 进步）。如果第一个更新程序回滚，那么其影响是 被否定，第二个更新程序可以继续更新 原来找到行。如果第一个更新者提交，第二个更新者提交 如果第一个更新者删除了该行，则将忽略该行，否则将忽略该行 尝试将其操作应用于该行的更新版本。

因此 tx2 中的 select * from "product" where (code = 'a') for update 将等待 tx1 完成。此时 tx1 已删除产品 a，因此该行被忽略并且不返回任何结果。现在我知道 tx1 也重新创建了产品 a，但请记住“select 查询（没有 for update/share 子句）只能看到查询开始之前提交的数据；”并且由于选择在 tx1 重新创建记录之前开始，因此不会被看到。

问题2

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

来自 Repeatable Read Isolation Level 的文档（可序列化是更高级别，因此适用这些规则以及一些更严格的规则）：

更新、删除、选择更新和选择共享命令 在搜索目标行方面与 select 的行为相同：它们 将仅查找自事务以来已提交的目标行 开始时间。然而，这样的目标行可能已经被更新 此时已被另一个并发事务（或删除或锁定） 找到了。在这种情况下，可重复读事务将等待 对于第一个更新事务要提交或回滚（如果是 仍在进行中）。如果第一个更新程序回滚，则其影响 被否定并且可重复读取事务可以继续进行 更新最初找到的行。但是如果第一个更新者提交 （并且实际上更新或删除了该行，而不仅仅是锁定它）然后 可重复读取事务将随消息一起回滚

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

如何进行第二次更新？*

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

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

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

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

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

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

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

也许我理解错误 - 我以前没有使用过 gorm。 然而，从您的查询注释来看，您的两个 goroutine 中的两个事务都有一个“select .. for update”，并且它们是并行运行的。在尝试“select.. for update”相同的行之前，您的主 goroutine 不会等待在第二个 goroutine 内启动的事务提交。

根据您的解释，也许您错误地将“for update”包含在第二个 goroutine 中。

或者你可以在第二个 goroutine 中使用 sync.Mutex 锁并在提交后释放它。而主协程则等待获取锁，然后才执行其查询。

今天关于《维持一致性并同步更新》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！