处理PostgreSQL表的并发进程

Question

我有一个简单的过程，需要处理一个表的记录，理想情况下运行多个流程实例，而不处理同一记录。我在MySQL中这样做的方式相当常见（尽管我认为令牌字段更像是一种黑客行为）：

向表中添加几个字段：

CREATE TABLE records (
    id INTEGER PRIMARY KEY AUTO_INCREMENT,
    ...actual fields...

    processed_at DATETIME DEFAULT NULL,
    process_token TEXT DEFAULT NULL
);

然后是一个简单的处理脚本：

process_salt = md5(rand()) # or something like a process id

def get_record():
    token = md5(microtime + process_salt)
    db.exec("UPDATE records SET process_token = ?
             WHERE processed_at IS NULL LIMIT 1", token)
    return db.exec("SELECT * FROM records WHERE token = ?", token)

while (row = get_record()) is valid:
    # ...do processing on row...

    db.exec("UPDATE records SET processed_at = NOW(), token = NULL
             WHERE id = ?", row.id)

我正在使用PostgreSQL数据库的系统中实现这样的过程。我知道Pg在锁定方面可以被认为是比MySQL更成熟的，这要归功于MVCC - 我可以在Pg中使用行锁定或其他一些功能而不是令牌字段吗？

Answer 1

您可以使用<代码>选择...FOR UPDATE NOWAIT，它将获得该行上的排他锁，或者如果该行已被锁定，则报告错误。

Answer 2

这种方法适用于PostgreSQL，但它往往会非常低效，因为你要更新每行两次 - 每次更新需要两个事务，两个提交。通过使用commit_delay并可能禁用synchronous_commit，可以在一定程度上减轻其成本，但除非您在存储子系统上具有非易失性回写缓存，否则它仍然不会很快。

更重要的是，由于您正在提交第一次更新，因此无法区分仍在工作的工作人员和崩溃的工作人员之间的区别。如果所有工作线程都在本地计算机上，则可以将令牌设置为工作线程的进程 ID，然后偶尔扫描丢失的 PID，但这很麻烦且容易出现争用条件，更不用说 pid 重用的问题了。

我建议你采用一个真正的队列解决方案来解决这些问题，比如ActiveMQ、RabbitMQ、ZeroMQ等等。PGQ也可能是非常重要的。

在事务性关系数据库中进行队列处理应该很容易，但在实践中，很难做好并正确处理。大多数“解决方案”乍一看是合理的，但在详细检查时，它们实际上是序列化了所有工作（因此，在任何给定时间，许多队列工作者中只有一个在做任何事情）。