Очередь заданий в виде таблицы SQL с несколькими потребителями (PostgreSQL)

Прочтите мой пост здесь:

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

Если вы используете транзакцию и LOCK TABLE, у вас не будет проблем.

Я также использую postgres для очереди FIFO. Первоначально я использовал ACCESS EXCLUSIVE, который дает правильные результаты при высоком уровне параллелизма, но имеет неудачный эффект взаимоисключаемости с pg_dump, который получает блокировку ACCESS SHARE во время своего выполнения. Это приводит к тому, что моя функция next () блокируется на очень долгое время (продолжительность pg_dump). Это было неприемлемо, так как мы работаем круглосуточно, и покупателям не нравилось время простоя в очереди посреди ночи.

Я подумал, что должна быть менее строгая блокировка, которая по-прежнему будет безопасной для одновременного доступа и не блокируется во время работы pg_dump. Мой поиск привел меня к этому сообщению ТАК.

Затем я провел небольшое исследование.

Следующие режимы достаточны для функции очереди FIFO NEXT (), которая обновит статус задания с в очереди на работает без сбоев параллелизма, а также не блокирует 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)

Вот сценарий оболочки, который проверяет все различные режимы блокировки при высоком уровне параллелизма (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

Я обновляю свое приложение, чтобы использовать ИСКЛЮЧИТЕЛЬНЫЙ режим, поскольку это наиболее ограничительный режим, который а) является правильным и б) не конфликтует с pg_dump. Я выбрал наиболее ограничительный, поскольку это кажется наименее рискованным с точки зрения изменения приложения с ACCESS EXCLUSIVE, не будучи сверхэкспертом по блокировке postgres.

Я чувствую себя довольно комфортно со своим испытательным стендом и с общими идеями, лежащими в основе ответа. Я надеюсь, что это поможет решить эту проблему другим.

Для этого не нужно делать блокировку всей таблицы: \.

Блокировка строки, созданная с помощью 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;

как насчет просто выбора?

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

Возможно, вы захотите посмотреть, как это делает queue_classic. https://github.com/ryandotsmith/queue_classic

Код довольно короткий и простой для понимания.

Хорошо, вот решение, которое работает для меня, на основе ссылки от 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 таблица журнала остается пустой :-)

Попробуйте PgQ вместо того, чтобы изобретать колесо.