PostgreSQL 高并发任务分配系统实践__给任务分配线程ID，或让线程去抢占任务执行，是任务分配系统中的基本需求。在数据库中，就是一个高并发的，实时更新系统，设计时要尽量避免冲突，提高处理吞吐。 PostgreSQL的UDF，advisory lock是一个很好的功能点，可以实现高并发、高可靠的任务分配。_

PG中文社区 /

mdi-home

首页

社区新闻

中文文档

加入ACE

mdi-home

首页

mdi-chat-processing

社区新闻

mdi-book-open-variant

中文文档

mdi-account-multiple-check

加入ACE

mdi-file-multiple-outline

相关资料

mdi-blank

mdi-exit-to-app

退出账号

PostgreSQL 高并发任务分配系统实践

原作者：德哥/digoal　　创作时间：2017-12-18 23:09:05+08　　

doudou586 发布于2017-12-19 23:09:05　　

评论: 1　　

顶: 856　

踩: 830　

# PostgreSQL 高并发任务分配系统 实践

#### 作者：德哥/digoal
#### 发布：2017-12-18
#### 欢迎大家踊跃投稿，投稿信箱： press@postgres.cn

----
## 背景
 给任务分配线程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都不是预先生成的，那么同样可以使用类似的功能点提高并发和可靠性。

使用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
```

![PostgreSQL_Community](/images/news/2016/pg_bot_banner.jpg)

评论:1　　

顶: 856　

踩: 830　

请在登录后发表评论，否则无法保存。

1# __ xiaowu 回答于 2024-04-22 09:44:07+08

人心叵测:https://www.nanss.com/yuedu/4002.html 为人民服务心得体会:https://www.nanss.com/gongzuo/3409.html 可爱的游戏名字:https://www.nanss.com/mingcheng/3772.html 英文游戏名字:https://www.nanss.com/mingcheng/3867.html 我的学习生活:https://www.nanss.com/xuexi/3316.html 沙雕女孩的可爱文案:https://www.nanss.com/wenan/3752.html 2022最火的个性签名:https://www.nanss.com/yulu/3816.html 晒美食的说说朋友圈:https://www.nanss.com/wenan/3971.html 开心就好:https://www.nanss.com/yulu/3540.html 入团志愿书:https://www.nanss.com/xuexi/2685.html 农村入党申请书:https://www.nanss.com/shenghuo/2624.html 环卫工人作文:https://www.nanss.com/xuexi/3279.html 卖报歌歌词:https://www.nanss.com/shenghuo/3940.html 给孩子的一封信:https://www.nanss.com/shenghuo/3174.html 荷花作文:https://www.nanss.com/xuexi/2165.html 会变大变小的书包作文四年级:https://www.nanss.com/xuexi/3687.html 管理评审报告:https://www.nanss.com/gongzuo/3843.html 小学作文:https://www.nanss.com/xuexi/3002.html 比较牛的网名:https://www.nanss.com/mingcheng/3730.html 运动健将作文300字三年级下册:https://www.nanss.com/xuexi/3699.html 班级总结:https://www.nanss.com/gongzuo/3484.html 班级典型教育案例:https://www.nanss.com/gongzuo/3570.html 事故调查报告:https://www.nanss.com/shenghuo/3388.html 王者荣耀好听的id:https://www.nanss.com/mingcheng/3738.html 衡是什么单位:https://www.nanss.com/wenti/3794.html 又可爱又沙雕的句子:https://www.nanss.com/yulu/3766.html 对老师说的话:https://www.nanss.com/xuexi/3617.html 个性签名:https://www.nanss.com/shenghuo/3491.html 当幸福来敲门观后感:https://www.nanss.com/shenghuo/3171.html 动员大会发言稿:https://www.nanss.com/shenghuo/3019.html