如何使用“SELECT FOR UPDATE”锁定一组行

Question

我有一个包含需要处理任务的“jobs”表。

一组工作人员可以使用像 SELECT * FROM jobs WHERE status='new' LIMIT 1 FOR UPDATE SKIP LOCKED 这样简单的查询，逐个并行地处理行。

但是我需要锁定由单个工作者链接的所有行。

类似这样的查询 SELECT * FROM jobs WHERE status='new' GROUP BY person_id LIMIT 1 FOR UPDATE SKIP LOCKED 可以完成该任务，但聚合函数无法使用锁定。

如果使用子查询 SELECT * FROM jobs WHERE person_id in (SELECT person_id FROM jobs WHERE status='new' LIMIT 1 FOR UPDATE SKIP LOCKED) FOR UPDATE SKIP LOCKED ，我无法保证所有行都会被同一线程锁定。

而将“人”行在“persons”表中锁定，而不是在作业中进行锁定，不是首选解决方案，因为其他进程也要使用此表。

请建议是否有一种简单的方法来实现此功能。

- Andrew Rozdolsky

---

你为什么想要显式锁定行呢？如果你能分享更广泛的问题陈述，我认为有更好的替代方案。此外，关系型数据库并不是构建作业队列的最佳工具 - 你考虑过使用专用的 MQ 服务器，如 RabbitMQ、Kafka 或 SQS 吗？ - Dai

这个设计有点“奇怪”。你想要像这样锁定行的事实暗示你可能遇到了XY问题。 - Mitch Wheat

我不能保证所有行都会被同一个线程锁定。无论如何，如果一个已经被锁定的人的新工作在该人的现有工作之后被插入，会怎样呢？ - jjanes

你说得对，我在这里用来说明问题的案例有点人为... 但尽管如此，我只是在寻求建议，是否可能编写一条锁定并返回与数据库中第一个可用的"person"相关的所有记录的PostgreSQL查询。 - Andrew Rozdolsky

Answer 1

你可以使用咨询锁来独占锁定“人”实体。

- Jonathan Jacobson

---

那样做是可行的，但我需要查询每个工人的每个人，这会产生很多额外的请求。我希望有一些优雅的解决方案，可以通过单个查询从数据库中获取并锁定记录。 - Andrew Rozdolsky

@AndrewRozdolsky，您可以将对pg_try_advisory_lock的调用与主查询组合使用。如果函数调用的结果为“true”，则表示工作进程可以继续进行。 - Jonathan Jacobson

好的，SELECT * FROM jobs WHERE status='new' AND pg_try_advisory_xact_lock(pic.person_id) = true LIMIT 1似乎正是我需要的。我还会添加GROUP BY以消除在相同id上重复尝试获取锁的情况。 - Andrew Rozdolsky

如果您打算在同一实例中的其他任何地方使用咨询锁，请记住，避免不同模块尝试锁定具有相同ID的不同资源之间的冲突由您负责。 - Jonathan Jacobson