沉浸式学习PostgreSQL|PolarDB 2: 电商高并发秒杀业务、跨境电商高并发队列消费业务

drop table IF EXISTS tbl;  
create unlogged table tbl (   -- 测试使用unlogged table减少redo   
  id int primary key,  -- 商品id  
  cnt int,             -- 库存  
  ts timestamp        -- 修改时间  
);

insert into tbl values (1, 2000000000, now());

postgres=# alter system set max_connections =2000;  
ALTER SYSTEM  
docker stop pg  
docker start pg  
docker exec -ti pg bash

vi t1.sql  
update tbl set cnt=cnt-1, ts=now() where id=1;

pgbench -M prepared -n -r -f ./t1.sql -P 1 -c 1920 -j 8 -T 120

transaction type: ./t1.sql  
scaling factor: 1  
query mode: prepared  
number of clients: 1920  
number of threads: 8  
duration: 120 s  
number of transactions actually processed: 31875  
latency average = 7729.207 ms  
latency stddev = 15626.227 ms  
initial connection time = 1784.073 ms  
tps = 230.270300 (without initial connection time)  
statement latencies in milliseconds:  
      7722.072  update tbl set cnt=cnt-1, ts=now() where id=1;

vi t1.sql  
with tmp as (  
  select id from tbl where id=1 for update skip locked  
)  
update tbl set cnt=cnt-1, ts=now() from tmp where tbl.id=tmp.id returning tbl.id;

pgbench -M prepared -n -r -f ./t1.sql -P 1 -c 1920 -j 8 -T 120

transaction type: ./t1.sql  
scaling factor: 1  
query mode: prepared  
number of clients: 1920  
number of threads: 8  
duration: 120 s  
number of transactions actually processed: 2847721  
latency average = 77.784 ms  
latency stddev = 138.536 ms  
initial connection time = 1703.949 ms  
tps = 23915.353056 (without initial connection time)

vacuum full tbl;

vi t2.sql  
update tbl set cnt=cnt-1, ts=now() where id=1 and pg_try_advisory_xact_lock(1) returning id;

pgbench -M prepared -n -r -f ./t2.sql -P 1 -c 1920 -j 8 -T 120

transaction type: ./t2.sql  
scaling factor: 1  
query mode: prepared  
number of clients: 1920  
number of threads: 8  
duration: 120 s  
number of transactions actually processed: 13917081  
latency average = 12.053 ms  
latency stddev = 30.646 ms  
initial connection time = 1879.213 ms  
tps = 116928.030422 (without initial connection time)  
statement latencies in milliseconds:  
        12.056  update tbl set cnt=cnt-1, ts=now() where id=1 and pg_try_advisory_xact_lock(1) returning id;

drop table if exists t_order_q;  
create unlogged table t_order_q (    
  id serial8 primary key,   -- 自增主键    
  order_id uuid unique,     -- 上游传递过来的订单号    
  cts timestamp not null    -- 上游传递过来的订单创建时间     
);     
-- create index on t_order_q (cts); -- 如果按订单时间先后取出处理, 则需要创建时间字段索引.  本实验按自增主键顺序处理, 则不需要时间索引.

drop table if exists t_order_u;  
create unlogged table t_order_u (    
  id serial8 primary key,   -- 自增主键    
  order_id uuid unique,     -- 上游传递过来的订单号    
  cts timestamp not null,    -- 上游传递过来的订单创建时间     
  uts timestamp not null,   -- 订单处理时间    
  status int not null       -- 订单处理状态标记     
);

truncate t_order_q;  
insert into t_order_q (order_id, cts) select gen_random_uuid(), clock_timestamp() from generate_series(1,1000000);

vi t1.sql    
begin;    
select id as vid from t_order_q order by id for update limit 1 \gset    
with tmp as     
  (delete from t_order_q where id = :vid returning order_id, cts)    
insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;     
end;

select 1000000/256.0;    
3906.2500000000000

pgbench -M extended -f ./t1.sql -n -r -P 1 -c 256 -j 8 -t 3906

transaction type: ./t1.sql  
scaling factor: 1  
query mode: extended  
number of clients: 256  
number of threads: 8  
number of transactions per client: 3906  
number of transactions actually processed: 999936/999936  
latency average = 111.243 ms  
latency stddev = 125.890 ms  
initial connection time = 234.312 ms  
tps = 2280.326174 (without initial connection time)

vi t2.sql    
begin;    
select id as vid from t_order_q order by id for update skip locked limit 1 \gset    
with tmp as     
  (delete from t_order_q where id = :vid returning order_id, cts)    
insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;     
end;

pgbench -M extended -f ./t2.sql -n -r -P 1 -c 256 -j 8 -t 3906

transaction type: ./t2.sql  
scaling factor: 1  
query mode: extended  
number of clients: 256  
number of threads: 8  
number of transactions per client: 3906  
number of transactions actually processed: 999936/999936  
latency average = 65.596 ms  
latency stddev = 104.377 ms  
initial connection time = 234.029 ms  
tps = 3795.525190 (without initial connection time)

vi t3.sql    
with tmp as     
  (delete from t_order_q where ctid = (select ctid from t_order_q where pg_try_advisory_xact_lock(id) order by id limit 1) returning order_id, cts)    
insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;  
- 或  
-   
- begin;  
- select id as v_id from t_order_q where pg_try_advisory_xact_lock(id) order by id limit 1 \gset  
- with tmp as (delete from t_order_q where id = :v_id returning order_id, cts)   
-   insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;     
- end;  
-   
- 或(sleep 模拟应用拿到需要处理的订单后的应用端操作增加的耗时.)  
-   
- begin;   
- select id as v_id from t_order_q where pg_try_advisory_xact_lock(id) order by id limit 1 \gset  
- \sleep 10ms  
- with tmp as (delete from t_order_q where id = :v_id returning order_id, cts)   
-   insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;     
- end;

pgbench -M extended -f ./t3.sql -n -r -P 1 -c 256 -j 8 -t 3906

transaction type: ./t3.sql  
scaling factor: 1  
query mode: extended  
number of clients: 256  
number of threads: 8  
number of transactions per client: 3906  
number of transactions actually processed: 999936/999936  
latency average = 20.404 ms  
latency stddev = 45.780 ms  
initial connection time = 239.823 ms  
tps = 12283.493260 (without initial connection time)

drop table if exists t_order_q;  
create unlogged table t_order_q (    
  id serial8 primary key,   -- 自增主键    
  order_id uuid ,     -- 上游传递过来的订单号    
  cts timestamp not null    -- 上游传递过来的订单创建时间     
) PARTITION BY hash (id) ;     
-- create index on t_order_q (cts); -- 如果按订单时间先后取出处理, 则需要创建时间字段索引.  本实验按自增主键顺序处理, 则不需要时间索引.     
do language plpgsql $$  
declare  
  x int := 256;  
begin  
for i in 0..x-1 loop  
  execute format ($_$create unlogged table t_order_q_%s PARTITION OF t_order_q FOR VALUES WITH (MODULUS %s, REMAINDER %s);$_$,   
    i, x, i  
  );  
end loop;  
end  
$$;

drop table if exists t_order_u;  
create unlogged table t_order_u (    
  id serial8 primary key,   -- 自增主键    
  order_id uuid unique,     -- 上游传递过来的订单号    
  cts timestamp not null,    -- 上游传递过来的订单创建时间     
  uts timestamp not null,   -- 订单处理时间    
  status int not null       -- 订单处理状态标记     
);

truncate t_order_q;  
insert into t_order_q (order_id, cts) select gen_random_uuid(), clock_timestamp() from generate_series(1,1000000);

create or replace function dyn_pre_orders(partitions int, client_id int) returns int8 as $$  
declare  
  suffix int := mod(client_id, partitions);   
begin  
  execute format('execute p%s(%s)', suffix, client_id);   
  return 0;   
  exception when others then  
    execute format('prepare p%s(int) as   
      with tmp as     
        (delete from t_order_q_%s where ctid = (select ctid from t_order_q_%s where pg_try_advisory_xact_lock(id) order by id limit 1) returning order_id, cts)    
      insert into t_order_u (order_id,cts,uts,status) select tmp.order_id, tmp.cts, now(), 1 from tmp;', suffix, suffix, suffix);   
    execute format('execute p%s(%s)', suffix, client_id);    
    return 0;  
end;  
$$ language plpgsql strict;

vi t4.sql    
\set clients 256  
select dyn_pre_orders(:clients, :client_id);

pgbench -M extended -f ./t4.sql -n -r -P 1 -c 256 -j 8 -t 3906

transaction type: ./t4.sql  
scaling factor: 1  
query mode: extended  
number of clients: 256  
number of threads: 8  
number of transactions per client: 3906  
number of transactions actually processed: 999936/999936  
latency average = 4.400 ms  
latency stddev = 12.287 ms  
initial connection time = 250.656 ms  
tps = 49892.858208 (without initial connection time)  
statement latencies in milliseconds:  
         0.000  \set clients 256  
         4.400  select dyn_pre_orders(:clients, :client_id);