在 postgres 中删除和插入的竞争条件答案

【问题标题】：Race condition for delete and insert in postgres在 postgres 中删除和插入的竞争条件
【发布时间】：2021-12-21 12:07:55
【问题描述】：

我正在开发一个节点项目，我在其中导入了 pg 库以进行数据库操作。我有一个 Kafka 队列，我从中获取事件并将它们存储在数据库中。我正在从 kafka 获取订单，每次更新订单时都会生成一个新事件，我需要删除旧的订单详细信息并用新的替换它们。

下面是代码

async function saveOrders(orders: Array<TransactionOnOrders>) {
  const client = await pool.connect()
 
  try {
    await client.query('BEGIN')
    if (orders.length) {
      const deleted = await deleteOrders(client, orders[0].orderId)
      logger.debug(`deleted rowCount ${deleted.rowCount} ${orders[0].orderId}`)
    }
    const queries = orders.map(ord => saveTransactionOnOrders(client, ord))
    await Promise.all(queries)
    await client.query('COMMIT')
  } catch (e) {
    await client.query('ROLLBACK')
    throw e
  } finally {
    client.release()
  }
}

订单更新非常频繁，我们收到大量事件，造成竞争条件，导致记录不被删除并插入额外记录。例如：假设我们收到了 order123 的事件并且事务正在处理中，直到它完成 order123 的另一个事件，所以删除查询返回 0 行受影响，插入查询插入另一行导致 2 行应该只存在一条记录。

我曾尝试更改隔离级别，但效果不佳并导致错误

await client.query('BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ')
await client.query('BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE')

我在这里做错了什么还是有更好的方法来处理上述情况？

【问题讨论】：

得到一个序列化错误没什么大不了的，你只需要抓住它并重试。

标签： node.js database postgresql race-condition

【解决方案1】：

正如其他人所建议的，这里的理想选择是使用INSERT ... ON CONFLICT 以原子方式执行此操作。如果没有看到 deleteOrders 和 saveTransactionOrders 的内容，我无能为力。

如果这不是一个选项，您应该使用SERIALIZABLE 作为隔离级别。然后你会得到一些序列化错误，但这些是安全的重试。如果你使用了@databases (https://www.atdatabases.org/docs/pg-guide-transactions)，你可以通过 retrySerializationFailures: true 重试：

async function saveOrders(orders: Array<TransactionOnOrders>) {
  await pool.tx(async client => 
    if (orders.length) {
      const deleted = await deleteOrders(client, orders[0].orderId)
      logger.debug(`deleted rowCount ${deleted.rowCount} ${orders[0].orderId}`)
    }
    const queries = orders.map(ord => saveTransactionOnOrders(client, ord))
    await Promise.all(queries)
  }, {
    isolationLevel: IsolationLevel.SERIALIZABLE,
    retrySerializationFailures: true,
  })
}

@databases 处理启动事务，并在异步回调结束时提交/回滚。它还会在序列化失败时重试。

如果您要处理大量事件，您可能会因为序列化失败的高频率而遇到性能问题并因此重试。

您可以在 node.js 中使用“锁定”来确保一次只有一个进程更新给定的一组订单。 https://www.atdatabases.org/docs/lock 应该使这个实现起来相当简单。不过，这只会锁定单个 node.js 进程上的并发事务，因此您仍然需要事务处理来处理多个 node.js 进程。

【讨论】：

【解决方案2】：

使用INSERT ... ON CONFLICT 来“插入”传入的行。这是原子的，没有竞争条件。

【讨论】：

【解决方案3】：

如果您更新行而不是删除它们并重新创建它们，这可能会更容易。在这种情况下，您可以依靠行锁来防止并发更新。

【讨论】：