标签
PostgreSQL , 高并发消费 , pg_try_advisory_xact_lock , 秒杀 , 任务分配
背景
给任务分配线程ID，或让线程去抢占任务执行，是任务分配系统中的基本需求。
目的是能够快速的消耗掉所有的任务，同又要保证两点：
1、所有任务都被领取。
2、每个任务只能被一个线程领取。
3、每个线程同一时间只能领取一个任务。
实际上在数据库中， 就是一个高并发的，实时更新系统，设计时要尽量避免冲突，提高处理吞吐。
PostgreSQL的UDF，advisory lock是一个很好的功能点，可以实现高并发、高可靠的任务分配。
其中，秒杀例子：
《HTAP数据库 PostgreSQL 场景与性能测试之 30 - (OLTP) 秒杀 - 高并发单点更新》
给任务分配唯一JAVA线程
例子1
功能描述：
有1000个java线程/进程，需要为具体的某个任务选举出一个master，并把选举结果写入到table中，记录任务ID与master线程/线程ID。
或者说：1每个任务的选举都被投票一次；2每个任务都只有一个master。
如果某个线程已经是某个任务的master，那这个线程/进程不参与选举。
1、JAVA线程与任务ID对应关系表
create table java_pool (
tid int primary key , -- JAVA 线程ID
taskid int unique -- 任务 ID
);
2、插入1000个线程ID
insert into java_pool select generate_series(1,1000);
3、输入任务ID，返回JAVA线程ID，表示这个任务分配给某个JAVA线程ID。
create or replace function set_master(v_taskid int) returns int as $$
declare
res int;
begin
-- set lock_timeout = '10 ms';
-- 使用adlock，消除唯一约束时的等待以及更新时的锁等待。
update java_pool set taskid=$1
where tid in
(select tid from java_pool where pg_try_advisory_xact_lock(tid) and pg_try_advisory_xact_lock($1) and taskid is null limit 1)
and pg_try_advisory_xact_lock($1)
returning tid into res;
return res;
exception when unique_violation then
return -1; -- this task already set other tid
when others then
return -2; -- lock timeout, other session is setting the same taskid.
end;
$$ language plpgsql strict;
释放TID
update java_pool set taskid=null where tid=? and pg_try_advisory_xact_lock(?) and taskid is not null;
4、压测
vi test.sql
\set taskid random(100,100000)
\set tid random(1,1000)
select set_master(:taskid);
update java_pool set taskid=null where tid=:tid and pg_try_advisory_xact_lock(:tid) and taskid is not null;
5、压测结果
pgbench -M prepared -n -r -P 1 -f ./test.sql -c 32 -j 32 -T 120
transaction type: ./test.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 120 s
number of transactions actually processed: 6876049
latency average = 0.558 ms
latency stddev = 0.508 ms
tps = 57298.823054 (including connections establishing)
tps = 57301.470302 (excluding connections establishing)
例子2
功能描述：
在一个table中，每一行记录了一个任务，需要把每个任务分配一个java执行线程/进程。
总的线程数/进程数多于任务数，并要求在table中记录当前任务分配到的线程/进程ID。
或者说，是多个java线程/进程需要争抢一个任务，需要某个方式实现： 1每个任务都被争抢到；2每个任务只被一个java线程/进程争抢到。
10万任务（已知）
100万线程（ID未知）
1、建表
create table task_pool (
taskid int primary key , -- 任务ID
tid int unique -- JAVA 线程ID
);
2、插入10万任务ID
insert into task_pool select generate_series(1,100000);
3、输入JAVA线程ID，返回任务ID。表示这个任务分配给某个JAVA线程ID。
create or replace function set_tid(v_tid int) returns int as $$
declare
res int;
begin
-- set lock_timeout = '10 ms';
update task_pool set tid=$1
where taskid in
(select taskid from task_pool where pg_try_advisory_xact_lock(taskid) and pg_try_advisory_xact_lock($1) and tid is null limit 1)
and pg_try_advisory_xact_lock($1)
returning taskid into res;
return res;
exception when unique_violation then
return -1; -- this task already set other tid
when others then
return -2; -- lock timeout, other session is setting the same taskid.
end;
$$ language plpgsql strict;
4、压测
vi test.sql
\set tid random(1,1000000)
select set_tid(:tid);
5、压测结果，约6秒分配完10万任务。
pgbench -M prepared -n -r -P 1 -f ./test.sql -c 32 -j 32 -T 120
progress: 1.0 s, 42693.2 tps, lat 0.733 ms stddev 0.333
progress: 2.0 s, 41831.7 tps, lat 0.765 ms stddev 0.285
progress: 3.0 s, 38475.0 tps, lat 0.832 ms stddev 1.544
progress: 4.0 s, 39560.5 tps, lat 0.809 ms stddev 0.276
progress: 5.0 s, 36850.0 tps, lat 0.868 ms stddev 0.317
progress: 6.0 s, 32344.5 tps, lat 0.989 ms stddev 0.720
progress: 7.0 s, 16541.2 tps, lat 1.934 ms stddev 0.579
progress: 8.0 s, 17078.0 tps, lat 1.875 ms stddev 0.575
例子3
如果任务ID和JAVA 线程ID都不是预先生成的，那么同样可以使用类似的功能点提高并发和可靠性。
使用pg_try_advisory_xact_lock来提高并发，降低等待。
insert into tbl select $1,$2 where pg_try_advisory_xact_lock($1) and pg_try_advisory_xact_lock($2) returning *;
根据结果判定是否锁定任务和JAVA线程ID