Hàng đợi công việc dưới dạng bảng SQL với nhiều người tiêu dùng (PostgreSQL)

Tôi cũng sử dụng postgres cho hàng đợi FIFO. Ban đầu, tôi đã sử dụng ACCESS EXCLUSIVE, mang lại kết quả chính xác trong tính đồng thời cao, nhưng có hậu quả đáng tiếc là loại trừ lẫn nhau với pg_dump, khóa này có được khóa ACCESS SHARE trong quá trình thực thi. Điều này khiến hàm next () của tôi bị khóa trong một thời gian rất dài (khoảng thời gian của pg_dump). Điều này là không thể chấp nhận được vì chúng tôi là một cửa hàng 24x7 và khách hàng không thích thời gian chờ đợi khi xếp hàng vào nửa đêm.

Tôi đã nghĩ rằng phải có một khóa ít hạn chế hơn mà vẫn sẽ an toàn đồng thời và không khóa trong khi pg_dump đang chạy. Tìm kiếm của tôi đã dẫn tôi đến bài đăng SO này.

Sau đó, tôi đã thực hiện một số nghiên cứu.

Các chế độ sau là đủ cho hàm NEXT () hàng đợi FIFO sẽ cập nhật trạng thái của công việc từ đã xếp hàng để chạy không có bất kỳ lỗi đồng thời nào và cũng không chặn pg_dump:

SHARE UPDATE EXCLUSIVE
SHARE ROW EXCLUSIVE
EXCLUSIVE

Truy vấn:

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;

Kết quả giống như sau:

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

Đây là một tập lệnh shell kiểm tra tất cả các chế độ khóa khác nhau ở mức đồng thời cao (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;";

Mã cũng ở đây nếu bạn muốn chỉnh sửa:https://gist.github.com/1083936

Tôi đang cập nhật ứng dụng của mình để sử dụng chế độ ĐỘC QUYỀN vì đây là chế độ hạn chế nhất mà a) đúng và b) không xung đột với pg_dump. Tôi đã chọn hạn chế nhất vì nó có vẻ ít rủi ro nhất về mặt thay đổi ứng dụng từ TRUY CẬP ĐỘC QUYỀN mà không phải là chuyên gia uber trong việc khóa postgres.

Tôi cảm thấy khá thoải mái với hệ thống thử nghiệm của mình và với những ý tưởng chung đằng sau câu trả lời. Tôi hy vọng rằng việc chia sẻ điều này sẽ giúp giải quyết vấn đề này cho những người khác.