【重新发现PostgreSQL之美】- 7 垂帘听政异步消息

2021-07-07 03:46:04

背景

场景:

重要数据在写入、更新、删除时实时告警或转存
流式数据(公务车电子围栏、刑侦数据探针、股票数据规则探针、服务器运行情况) 实时预警或事件触发
危险操作(DDL) 异步监控

规则+ 异步消息的优势:
1、通过规则过滤掉不需要写入的正常数据, 由于业务正常数据通常占比在99%以上, 从而大幅减轻写入量.
2、传统的利用定时器查询所有数据去发现问题, 还需要在时间、VAL、SID等层面去建立索引, 消耗大量存储, 同时索引增加写入RT,性能下降. 规则+异步完全规避这个问题.
3、可以实时发现并预警或触发其他动作

语法

postgres=# \h create rule

Command: CREATE RULE

Description: define a new rewrite rule

Syntax:

CREATE [ OR REPLACE ] RULE name AS ON event

TO table_name [ WHERE condition ]

DO [ ALSO | INSTEAD ] { NOTHING | command | ( command ; command ... ) }

where event can be one of:

SELECT | INSERT | UPDATE | DELETE

URL: https://www.postgresql.org/docs/14/sql-createrule.html

postgres=# \h listen

Command: LISTEN

Description: listen for a notification

Syntax:

LISTEN channel

URL: https://www.postgresql.org/docs/14/sql-listen.html

postgres=# \h notify

Command: NOTIFY

Description: generate a notification

Syntax:

NOTIFY channel [ , payload ]

URL: https://www.postgresql.org/docs/14/sql-notify.html

postgres=# \df *.*channel*

List of functions

Schema | Name | Result data type | Argument data types | Type

------------+-----------------------+------------------+---------------------+------

pg_catalog | pg_listening_channels | SETOF text | | func

(1 row)

postgres=# \df *.*notify*

List of functions

Schema | Name | Result data type | Argument data types | Type

------------+-----------+------------------+---------------------+------

pg_catalog | pg_notify | void | text, text | func

(1 row)

例子

机房传感器

create table tbl_sensor_log (

id serial8 primary key,

sid int,

val jsonb,

crt_time timestamp

);

定义规则, 发现异常数据向alert通道发送消息

create or replace rule r1 as on insert

to tbl_sensor_log

where coalesce(val['temp']::float4,0) >= 60

or coalesce(val['cpu_perct']::float4,0) >= 80

or coalesce(val['mem_perct']::float4,0) >= 80

or coalesce(val['io_perct']::float4,0) >= 80

do also

select pg_notify('alert', format('sensor: %s, ts:%s, val:%s', NEW.sid, NEW.crt_time, NEW.val));

定义规则(可选), 正常数据不写入

create or replace rule r2 as on insert

to tbl_sensor_log

where not (coalesce(val['temp']::float4,0) >= 60

or coalesce(val['cpu_perct']::float4,0) >= 80

or coalesce(val['mem_perct']::float4,0) >= 80

or coalesce(val['io_perct']::float4,0) >= 80)

do instead NOTHING;

postgres=# \d+ tbl_sensor_log;

Table "public.tbl_sensor_log"

----------+-----------------------------+-----------+----------+--------------------------------------------+----------+-------------+--------------+-------------

Indexes:

"tbl_sensor_log_pkey" PRIMARY KEY, btree (id)

Rules:

r1 AS

ON INSERT TO tbl_sensor_log

WHERE COALESCE(new.val['temp'::text]::real, 0::real) >= 60::double precision OR COALESCE(new.val['cpu_perct'::text]::real, 0::real) >= 80::double precision OR COALESCE(new.val['mem_perct'::text]::real, 0::real) >= 80::double precision OR COALESCE(new.val['io_perct'::text]::real, 0::real) >= 80::double precision DO SELECT pg_notify('alert'::text, format('sensor: %s, val:%s'::text, new.sid, new.val)) AS pg_notify

Access method: heap

压测

CREATE TYPE sensor_js AS (temp float4, cpu_perct float4, mem_perct float4, io_perct float4);

insert into tbl_sensor_log (sid,val,crt_time)

values (

row_to_json(row(1,80.1,2,99.11)::sensor_js)::jsonb,

now()

);

vi test.sql

\set sid random(1,1000000)

\set v1 random(1,61)

\set v2 random(1,81)

\set v3 random(1,81)

\set v4 random(1,81)

insert into tbl_sensor_log (sid,val,crt_time)

values (:sid, row_to_json(row(:v1,:v2,:v3,:v4)::sensor_js)::jsonb,now());

pgbench -M prepared -n -r -P 1 -f ./test.sql -c 5 -j 5 -T 120

开启其他会话, 监听alert这个通道的异步消息.

PG 的异步消息为广播模式. 可以在多个会话监听同一个通道, 如果有多个业务希望接收同一类异步消息, 则可以这么做.

listen alter;

Asynchronous notification "alert" with payload "sensor: 459294, val:{"temp": 32, "io_perct": 81, "cpu_perct": 76, "mem_perct": 39}" received from server process with PID 1715.

Asynchronous notification "alert" with payload "sensor: 788337, val:{"temp": 60, "io_perct": 34, "cpu_perct": 12, "mem_perct": 53}" received from server process with PID 1714.

Asynchronous notification "alert" with payload "sensor: 421071, val:{"temp": 7, "io_perct": 81, "cpu_perct": 12, "mem_perct": 14}" received from server process with PID 1716.

Asynchronous notification "alert" with payload "sensor: 523366, val:{"temp": 13, "io_perct": 45, "cpu_perct": 70, "mem_perct": 80}" received from server process with PID 1713.

Asynchronous notification "alert" with payload "sensor: 94909, val:{"temp": 57, "io_perct": 1, "cpu_perct": 32, "mem_perct": 81}" received from server process with PID 1713.

Asynchronous notification "alert" with payload "sensor: 13910, val:{"temp": 61, "io_perct": 39, "cpu_perct": 39, "mem_perct": 2}" received from server process with PID 1714.

Asynchronous notification "alert" with payload "sensor: 252342, val:{"temp": 7, "io_perct": 31, "cpu_perct": 80, "mem_perct": 13}" received from server process with PID 1714.

Asynchronous notification "alert" with payload "sensor: 222983, val:{"temp": 56, "io_perct": 76, "cpu_perct": 80, "mem_perct": 25}" received from server process with PID 1715.

Asynchronous notification "alert" with payload "sensor: 913661, val:{"temp": 60, "io_perct": 23, "cpu_perct": 80, "mem_perct": 9}" received from server process with PID 1716.

压测数据分析:

1、在不开启rule时, 写入速度比开启rule快, 因为rule里面有CPU运算. 增加了RT.

但是这是纯计算, 没有IO, 内存等开销. 总体效率绝对比定时器后查询快很多.

progress: 1.0 s, 63373.9 tps, lat 0.078 ms stddev 0.066

progress: 2.0 s, 67591.2 tps, lat 0.074 ms stddev 0.044

progress: 3.0 s, 66330.3 tps, lat 0.075 ms stddev 0.039

progress: 4.0 s, 65786.8 tps, lat 0.076 ms stddev 0.038

progress: 5.0 s, 65436.3 tps, lat 0.076 ms stddev 0.043

progress: 6.0 s, 64276.1 tps, lat 0.077 ms stddev 0.042

progress: 7.0 s, 59162.6 tps, lat 0.084 ms stddev 0.045

progress: 8.0 s, 53887.5 tps, lat 0.092 ms stddev 0.048

progress: 1.0 s, 43413.8 tps, lat 0.114 ms stddev 0.084