使用 Twisted Python 锁定 Postgres 表

Question

我有一个扭曲的守护进程，它可以解析一些 xml 提要。

我通过 twisted.enterprise.adbapi 将数据存储在 PostgreSQL 中，IIRC 正在包装 psycopg2

我在将数据存储到数据库时遇到了一些问题——重复数据会定期进入数据库。

说实话，我的实现存在一些潜在问题，应该重做并设计得更好。不过我没有时间和资源来做这件事——所以我们现在处于“保持运行”模式。

我认为问题可能来自我对 deferToThread 的使用或我在开始时生成服务器的方式。

作为我认为有问题的功能的快速概述：

Twisted 在 Postgres 中查询应该被分析的账户，并在它们上设置一个块

SELECT 
    id 
FROM 
    xml_sources 
WHERE 
    timestamp_last_process < ( CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - INTERVAL '4 HOUR' ) 
    AND
    is_processing_block IS NULL ;

lock_ids = [ i['id'] for i in results ]

UPDATE xml_sources SET is_processing_block = True WHERE id IN %(lock_ids)s

我认为正在发生的事情是（意外地）运行多个服务器或各种其他问题导致多个线程处理这些数据。

我认为如果我将此快速部分包含在独占表锁中，这可能会得到解决 - 或至少排除为问题。

不过，我以前从来没有通过 twisted 完成表锁定。谁能指出我正确的方向？

Answer 1

您可以使用SELECT FOR UPDATE 而不是普通的SELECT，这将锁定查询返回的行。如果您确实想要 table 锁定，您可以发出 LOCK 语句，但根据您的其余问题，我认为您想要行锁定。

如果您使用adbapi，请记住，如果您想在事务中运行多个语句，则需要使用runInteraction。传递给runInteraction 的函数将在一个线程中运行，因此您可能需要使用callFromThread 或blockingCallFromThread 才能从数据库交互返回到reactor。

但是，锁定可能不是您的问题。一方面，如果您将deferToThread 和adbapi 混合使用，则可能有问题。 adbapi 已经在为你做相当于deferToThread for 的工作。你应该可以在主线程上做所有事情。

不过，您必须提供一个具有代表性的示例才能获得更具体的答案。考虑一下您的问题：基本上是“有时我会收到重复的数据，并且执行过程自认存在问题，这很大，我无法修复，也无法向您展示。”这不是一个可以回答的问题。