使用 Python 在 PostgreSQL 中行锁定和更新的并发问题

Question

我在设计用于在高度并发环境中更新 PostgreSQL 数据库中的行的 Python 函数时遇到问题。该函数的目的是使用 FOR UPDATE SKIP LOCKED 锁定领域表中最早的未分配行，更新它，然后提交事务。但是，我注意到，当同时运行大约 50 个请求时，数据库中更新的行数小于函数执行的总数。我已经验证该函数每次都成功执行，没有任何问题。

我正在使用

psycopg2

与 Postgres 数据库交互

def allocate_realm(self):
    try:
        with self.db_conn.cursor() as cursor:
            cursor.execute("""
                SELECT * FROM realms
                WHERE is_assigned IS NULL
                LIMIT 1
                FOR UPDATE SKIP LOCKED;
            """)
            realm = cursor.fetchone()

            if realm:
                cursor.execute("UPDATE realms SET is_assigned = 'some_value' WHERE id = %s;", (realm[0],))
                self.db_conn.commit()
            else:
                # Handle the case when no unassigned realm is found
                pass
    except Exception:
        self.db_conn.rollback()
        # Error handling

并发：大约 50 个并发请求

在此并发设置中，什么可能导致更新的行数与函数执行总数之间存在差异？我处理事务或锁的方式是否需要改进，以确保准确反映数据库中的更新？

同时，我甚至检查了 5 个同样失败的并行请求。

Answer 1

我能够通过在 UPDATE 查询中使用子查询来解决这个问题，而不是在同一事务中的 2 个不同查询中获取和更新。这是我的查询如下 -

UPDATE realms SET is_assigned = 'some_value'
WHERE id IN (
    SELECT id FROM realms
    WHERE is_assigned IS NULL
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
RETURNING *;

更新查询后，我使用 JMeter 对超过 1000 个并行请求进行了测试，这就像一个魅力，没有太多/任何性能影响。我仍然很困惑为什么这个有效而其他人失败了。尽管有一种解释，但我无法想出一个合理的解决方案来解释为什么我的原始查询不起作用。