使用 slick 在 postgres 更新中避免竞争条件答案

【问题标题】：Avoiding race condition in postgres updates using slick使用 slick 在 postgres 更新中避免竞争条件
【发布时间】：2021-04-01 23:23:58
【问题描述】：

case class Item(id: String, count: Int).  

class ItemRepo(db: Database) {
  val query = TableQuery[ItemTable]


def updateAmount(id: String, incCount :Int) = {
   val currentRow = db.run(query.filter(_.id === id).result).head
   val updatedRow =  Item(currentRow.id, currentRow.count + incCount)
   db.run((query returning query).insertOrUpdate(updatedRow))
}

上面的代码有一个竞争条件 - 如果两个线程并行运行，它们可能都读取相同的计数，并且只有最后一个更新线程会增加它们的 incCount。

我怎样才能避免这种情况？我尝试在执行query.filter 的行中使用.forUpdate，但它不会阻塞另一个线程。我错过了什么吗？

【问题讨论】：

不要破坏您的帖子。通过在本网站上发布，您已不可撤销地授予 Stack Exchange 网络以CC BY-SA 4.0 license 分发该内容的权利，只要它认为合适即可。有关删除的替代方法，请参阅：I've thought better of my question; can I delete it?

标签： postgresql scala slick

【解决方案1】：

您可以使用一些技巧来改善这种情况。

首先，您要向数据库发送两个独立的查询（两个db.run 调用）。您可以通过将它们组合成单个操作并将其发送到数据库来改进它。例如：

// Danger: I've not tried to compile this. Please excuse typos.

val lookupAction = query.filter(_.id === id).result


val updateAction = lookupAction.flatMap { matchingRows =>
   val newItem = matchingRows.headOption match {
      case Some(Item(_, count)) => Item(id, count + incCount)
      case None => Item(id, 1) // or whatever your default is 
   }
   (query returning query).insertOrUpdate(newItem)
}

// and you db.run(updateAction.transactionally)

这将为您提供一些方法，具体取决于您的数据库的事务保证。我提到它是因为在 Slick 中组合动作是一个重要的概念。这样一来，您的 forUpdate（Laurenz Albe 指出）可能会按预期运行。

但是，您可能更愿意向数据库发送更新。您需要使用 Slick 的普通 SQL 功能来执行此操作：

val action = sqlu"UPDATE items SET count = count + $incCount WHERE id = $id"
// And then you db.run(action)

...并允许您的数据库处理并发（取决于数据库隔离级别）。

如果您真的想在所有客户端执行此操作，那么在 JVM 上的 Scala 代码中，存在并发概念，例如锁、actor 和 refs。 Slick 本身没有任何东西可以为您执行 JVM 锁定。

【讨论】：

【解决方案2】：

当您从数据库中获取数据时，您应该使用SELECT ... FOR UPDATE，以便您在该行上拥有一个排他锁，以防止其他会话在您的事务完成之前更新数据。

在 Slick 中，您可以使用 forUpdate 构造 available since version 3.2.0 来做到这一点。

【讨论】：