绑定完请刷新页面
取消
刷新

分享好友

×
取消 复制
PostgreSQL 高并发任务分配系统 实践
2019-04-26 14:25:07

作者:德哥/digoal

背景

给任务分配线程ID,或让线程去抢占任务执行,是任务分配系统中的基本需求。目的是能够快速的消耗掉所有的任务,同又要保证3点:

  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都不是预先生成的,那么同样可以使用类似的功能点提高并发和可靠性。

使用pgtryadvisoryxactlock来提高并发,降低等待。

insert into tbl select $1,$2 where pg_try_advisory_xact_lock($1) and pg_try_advisory_xact_lock($2


分享好友

分享这个小栈给你的朋友们,一起进步吧。

PostgreSQL中文社区小栈
创建时间:2019-04-19 17:47:49
PostgreSQL中文社区是一个非盈利的民间组织,已经在中国成功举办过8届技术大会。目前成员都以志愿者身份加入,成立的目的在于构建PG数据库技术生态圈子(内核、用户、培训机构、厂商、服务商、软件开发商、高校形成“业务与利益双向驱动”的良性发展生态圈);帮助企业解决人才培养和企业商用数据库成本问题。社区会在各运营平台发布PG新信息和PG相关技术文章,推动PG技术在中国的发展。
展开
订阅须知

• 所有用户可根据关注领域订阅专区或所有专区

• 付费订阅:虚拟交易,一经交易不退款;若特殊情况,可3日内客服咨询

• 专区发布评论属默认订阅所评论专区(除付费小栈外)

栈主、嘉宾

查看更多
  • wangliyun
    栈主
  • digoal
    嘉宾
  • 飘絮絮絮丶
    嘉宾

小栈成员

查看更多
  • 栈栈
  • 喵呜
  • osdba
  • 一号管理员
戳我,来吐槽~