作业队列作为具有多个消费者的 SQL 表 (PostgreSQL)答案

【问题标题】：Job queue as SQL table with multiple consumers (PostgreSQL)作业队列作为具有多个消费者的 SQL 表 (PostgreSQL)
【发布时间】：2011-09-24 08:22:10
【问题描述】：

我有一个典型的生产者-消费者问题：

多个生产者应用程序将作业请求写入 PostgreSQL 数据库上的作业表。

作业请求的状态字段在创建时开始包含 QUEUED。

当生产者插入新记录时，规则会通知多个消费者应用程序：

CREATE OR REPLACE RULE "jobrecord.added" AS
  ON INSERT TO jobrecord DO 
  NOTIFY "jobrecordAdded";

他们将尝试通过将状态设置为 RESERVED 来保留新记录。当然，只有一个消费者应该成功。所有其他消费者不应该能够保留相同的记录。他们应该使用 state=QUEUED 保留其他记录。

示例：一些生产者在表 jobrecord 中添加了以下记录：

id state  owner  payload
------------------------
1 QUEUED null   <data>
2 QUEUED null   <data>
3 QUEUED null   <data>
4 QUEUED null   <data>

现在，两个消费者A、B想要处理它们。它们同时开始运行。一个应该保留 id 1，另一个应该保留 id 2，然后第一个完成的应该保留 id 3，依此类推..

在纯多线程世界中，我会使用互斥锁来控制对作业队列的访问，但消费者是可能在不同机器上运行的不同进程。它们只访问同一个数据库，因此所有同步都必须通过数据库进行。

我阅读了很多关于 PostgreSQL 中并发访问和锁定的文档，例如http://www.postgresql.org/docs/9.0/interactive/explicit-locking.html Select unlocked row in Postgresql PostgreSQL and locking

从这些主题中，我了解到，以下 SQL 语句应该满足我的需要：

UPDATE jobrecord
  SET owner= :owner, state = :reserved 
  WHERE id = ( 
     SELECT id from jobrecord WHERE state = :queued 
        ORDER BY id  LIMIT 1 
     ) 
  RETURNING id;  // will only return an id when they reserved it successfully

不幸的是，当我在多个消费者进程中运行它时，大约 50% 的时间里，它们仍然保留相同的记录，既处理它，又覆盖另一个的更改。

我错过了什么？如何编写 SQL 语句，使多个消费者不会保留相同的记录？

【问题讨论】：

你在交易中这样做，不是吗？有一个 BEGIN 和一个 COMMIT？
不，没有 BEGIN 和 COMMIT。我尝试在事务中执行此操作，但是什么也没发生，我的消费者不处理任何事情，表保持不变。我应该使用交易吗？
看来，事务的问题在于我在消费者应用程序（用 Qt 编写）中使用它的方式。
好的，在下面看到了您的答案，也许您可以使用最终的工作代码进行编辑，因为这是一个非常常见的问题
如果您在事务中执行此操作但什么都没有发生，我的猜测是您没有提交，因为在这种情况下 tx 会回滚。

标签： sql postgresql queue producer-consumer

【解决方案1】：

我也将 postgres 用于 FIFO 队列。我最初使用 ACCESS EXCLUSIVE，它在高并发下产生正确的结果，但不幸的是与 pg_dump 互斥，后者在执行期间获取了一个 ACCESS SHARE 锁。这会导致我的 next() 函数锁定很长时间（pg_dump 的持续时间）。这是不可接受的，因为我们是 24x7 的商店，客户不喜欢半夜排队等候的时间。

我认为必须有一个限制较少的锁，它仍然是并发安全的，并且在 pg_dump 运行时不会锁定。我的搜索将我带到了这篇 SO 帖子。

然后我做了一些研究。

对于 FIFO 队列 NEXT() 函数，以下模式已经足够了，该函数会将作业的状态从 queued 更新为 running，而不会出现任何并发失败，也不会阻止 pg_dump：

SHARE UPDATE EXCLUSIVE
SHARE ROW EXCLUSIVE
EXCLUSIVE

查询：

begin;
lock table tx_test_queue in exclusive mode;
update 
    tx_test_queue
set 
    status='running'
where
    job_id in (
        select
            job_id
        from
            tx_test_queue
        where
            status='queued'
        order by 
            job_id asc
        limit 1
    )
returning job_id;
commit;

结果如下：

UPDATE 1
 job_id
--------
     98
(1 row)

这是一个 shell 脚本，它以高并发 (30) 测试所有不同的锁定模式。

#!/bin/bash
# RESULTS, feel free to repro yourself
#
# noLock                    FAIL
# accessShare               FAIL
# rowShare                  FAIL
# rowExclusive              FAIL
# shareUpdateExclusive      SUCCESS
# share                     FAIL+DEADLOCKS
# shareRowExclusive         SUCCESS
# exclusive                 SUCCESS
# accessExclusive           SUCCESS, but LOCKS against pg_dump

#config
strategy="exclusive"

db=postgres
dbuser=postgres
queuecount=100
concurrency=30

# code
psql84 -t -U $dbuser $db -c "create table tx_test_queue (job_id serial, status text);"
# empty queue
psql84 -t -U $dbuser $db -c "truncate tx_test_queue;";
echo "Simulating 10 second pg_dump with ACCESS SHARE"
psql84 -t -U $dbuser $db -c "lock table tx_test_queue in ACCESS SHARE mode; select pg_sleep(10); select 'pg_dump finished...'" &

echo "Starting workers..."
# queue $queuecount items
seq $queuecount | xargs -n 1 -P $concurrency -I {} psql84 -q -U $dbuser $db -c "insert into tx_test_queue (status) values ('queued');"
#psql84 -t -U $dbuser $db -c "select * from tx_test_queue order by job_id;"
# process $queuecount w/concurrency of $concurrency
case $strategy in
    "noLock")               strategySql="update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "accessShare")          strategySql="lock table tx_test_queue in ACCESS SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "rowShare")             strategySql="lock table tx_test_queue in ROW SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "rowExclusive")         strategySql="lock table tx_test_queue in ROW EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "shareUpdateExclusive") strategySql="lock table tx_test_queue in SHARE UPDATE EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "share")                strategySql="lock table tx_test_queue in SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "shareRowExclusive")    strategySql="lock table tx_test_queue in SHARE ROW EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "exclusive")            strategySql="lock table tx_test_queue in EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "accessExclusive")      strategySql="lock table tx_test_queue in ACCESS EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    *) echo "Unknown strategy $strategy";;
esac
echo $strategySql
seq $queuecount | xargs -n 1 -P $concurrency -I {} psql84 -U $dbuser $db -c "$strategySql"
#psql84 -t -U $dbuser $db -c "select * from tx_test_queue order by job_id;"
psql84 -U $dbuser $db -c "select count(distinct(status)) as should_output_100 from tx_test_queue;"
psql84 -t -U $dbuser $db -c "drop table tx_test_queue;";

如果您想编辑，代码也在这里：https://gist.github.com/1083936

我正在更新我的应用程序以使用 EXCLUSIVE 模式，因为它是 a) 正确且 b) 与 pg_dump 不冲突的最严格的模式。我选择了最严格的，因为在不成为 postgres 锁定方面的超级专家的情况下，将应用程序从 ACCESS EXCLUSIVE 更改为风险最小的。

我对我的测试设备和答案背后的一般想法感到很自在。我希望分享这有助于为其他人解决这个问题。

【讨论】：

谢谢，这很有用。我今天会研究一下。
我已经对此解决方案进行了大量测试，并且 100% 确信这是执行此操作的正确方法。我已将其部署到我的生产系统（使用 EXCLUSIVE 模式出列）。享受吧！
不应该只需要一个表锁一个行锁：stackoverflow.com/a/30315387/492548

【解决方案2】：

不需要为此做一个全表锁定：\。

使用for update 创建的行锁可以正常工作。

请参阅 https://gist.github.com/mackross/a49b72ad8d24f7cefc32 了解我对 apinstein 的回答所做的更改并验证它仍然有效。

最终代码是

update 
    tx_test_queue
set 
    status='running'
where
    job_id in (
        select
            job_id
        from
            tx_test_queue
        where
            status='queued'
        order by 
            job_id asc
        limit 1 for update
    )
returning job_id;

【讨论】：

此解决方案显式删除事务。为什么？
pg 在语句周围隐式添加事务
@mackross 最好使用for update skip locked 而不是for update，因为它不会阻塞其他并发读者。也许您没有使用它的原因是 skip locked 语句在 2015 年不可用。

【解决方案3】：

只选择呢？

SELECT * FROM table WHERE status = 'QUEUED' LIMIT 10 FOR UPDATE SKIP LOCKED;

https://www.postgresql.org/docs/9.5/static/sql-select.html#SQL-FOR-UPDATE-SHARE

【讨论】：

【解决方案4】：

在这里阅读我的帖子：

https://stackoverflow.com/a/6500830/32688

如果你使用 transaction 和 LOCK TABLE 你不会有问题。

【讨论】：

谢谢，添加表锁LOCK TABLE slots IN ACCESS EXCLUSIVE MODE;（并在Qt中修复我的事务语句）是解决方案。现在，它按预期工作。谢谢！
ACCESS EXCLUSIVE 可以工作，但是它有严重的性能副作用。值得注意的是，它会在使用ACCESS SHARE 模式的pg_dump 期间阻塞。这意味着您的队列将阻塞备份所需的时间，在我们的例子中可能是几个小时。

【解决方案5】：

您可能想看看 queue_classic 是如何做到的。 https://github.com/ryandotsmith/queue_classic

代码很短，很容易理解。

【讨论】：

【解决方案6】：

好的，这是对我有用的解决方案，基于 jordani 的链接。由于我的一些问题在于 Qt-SQL 的工作方式，所以我已经包含了 Qt 代码：

QSqlDatabase db = GetDatabase();
db.transaction();
QSqlQuery lockQuery(db);
bool lockResult = lockQuery.exec("LOCK TABLE serverjobrecord IN ACCESS EXCLUSIVE MODE; ");
QSqlQuery query(db);
query.prepare(    
"UPDATE jobrecord "
"  SET \"owner\"= :owner, state = :reserved "
"  WHERE id = ( "
"    SELECT id from jobrecord WHERE state = :queued ORDER BY id LIMIT 1 "
"  ) RETURNING id;"
);
query.bindValue(":owner", pid);
query.bindValue(":reserved", JobRESERVED);
query.bindValue(":queued", JobQUEUED); 
bool result = query.exec();

为了检查多个消费者是否处理同一个作业，我添加了一个规则和一个日志表：

CREATE TABLE serverjobrecord_log
(
  serverjobrecord_id integer,
  oldowner text,
  newowner text
) WITH ( OIDS=FALSE );


CREATE OR REPLACE RULE ownerrule AS ON UPDATE TO jobrecord
WHERE old.owner IS NOT NULL AND new.state = 1 
DO INSERT INTO jobrecord_log     (id, oldowner, newowner) 
    VALUES (new.id, old.owner, new.owner);

如果没有LOCK TABLE serverjobrecord IN ACCESS EXCLUSIVE MODE; 语句，日志表偶尔会填满条目，如果一个消费者覆盖了另一个消费者的值，但使用 LOCK 语句，日志表仍然是空的:-)

【讨论】：

【解决方案7】：

查看PgQ，而不是重新发明轮子。

【讨论】：

我已经检查了 PgQ，但不幸的是，它会向所有消费者发送相同的事件：You can have as many consumers as you want to on the same event queue, but they will all see the same events rather than share the workload. 这正是我不想想要发生的事情。我本来希望使用现有的解决方案，但找不到满足我要求的解决方案。
啊，看，我会使用 PgQ 然后让每个消费者使用 UPDATE ... RETURNING 语句在不同的表中注册一个锁定工作。值得深思。