381
技術社區[雲棲]
PostgreSQL 異步消息實踐 - 億級/分鍾 FEED係統實時監測
標簽
PostgreSQL , 異步消息 , 觸發器 , 規則 , insert on conflict , 實時分析
背景
在很多業務係統中，為了定位問題、運營需要、分析需要或者其他需求，會在業務中設置埋點，記錄用戶的行為在業務係統中產生的日誌，也叫FEED日誌。
比如訂單係統、在業務係統中環環相扣，從購物車、下單、付款、發貨，收貨（還有糾紛、退款等等），一筆訂單通常會產生若幹相關聯的記錄。
每個環節產生的屬性可能是不一樣的，有可能有新的屬性產生，也有可能變更已有的屬性值。
為了便於分析，通常有必要將訂單在整個過程中產生的若幹記錄（若幹屬性），合並成一條記錄（訂單大寬表）。
通常業務係統會將實時產生的訂單FEED數據寫入消息隊列，消息隊列使得數據變成了流動的數據：
RDS PG + OSS + HDB PG 分鍾清洗和主動檢測
數據通過消息隊列消費後，實時寫入RDS PG，在RDS PG進行訂單FEED的合並，寫入OSS外部表。(支持壓縮格式，換算成裸數據的寫入OSS的速度約100MB/s/會話)
HDB PG從OSS外部表讀取(支持壓縮格式，換算成裸數據的讀取OSS的速度約100MB/s/數據節點)，並將訂單FEED數據合並到全量訂單表。
《打造雲端流計算、在線業務、數據分析的業務數據閉環 - 阿裏雲RDS、HybridDB for PostgreSQL最佳實踐》
數據進入HDB PG後，通過規則SQL，從全量訂單表中，挖掘異常數據（或者分析）。
通過這種方案，實現了海量訂單FEED數據的分鍾級準實時分析。
這個方案已支撐了雙十一業務，高吞吐、低延遲，絲般柔滑。
毫秒級FEED監測與反饋方案
技術永遠是為業務服務的，分鍾級延遲雖然說已經很高了，但是在一些極端情況下，可能需要更低的延遲。
實際上RDS PostgreSQL還有更強的殺手鐧，可以實現毫秒級的異常FEED數據發現和反饋。
流式處理+異步消息，方法如下：
1、通過觸發機製結合異步消息通道實現。
2、通過pipeline，流式SQL結合異步消息通道實現。
應用程序監聽消息通道(listen channel)，數據庫則將異常數據寫入到消息通道(notify channel, message)。實現異常數據的主動異步推送。
毫秒級FEED監測與反饋架構設計
RDS PG設計
1、分實例，提高係統級吞吐。（例如單實例處理能力是15萬行/s，那麼100個實例，可以支撐1500萬行/s的實時處理。）
例如：
DB0, DB1, DB2, DB3, ..., DB255
映射關係：
db0, host?, port?
db1, host?, port?
...
2、實例內使用分表，提高單實例並行處理吞吐。當規則眾多時，分表可以提高單實例的規則處理吞吐。
例如
tbl0, tbl1, tbl2, ..., tbl127
tbl128, tbl129, tbl130, ..., tbl255
映射關係：
tbl0, db?
tbl1, db?
...
HDB PG設計
HDB PG依舊保留，用於PB級數據量的海量數據實時分析。
數據通路依舊采用OSS，批量導入的方式。
DEMO
1、創建訂單feed全寬表（當然，我們也可以使用jsonb字段來存儲所有屬性。因為PostgreSQL支持JSONB類型哦。PostgreSQL支持的多值類型還有hstore, xml等。）
create table feed(id int8 primary key, c1 int, c2 int, c3 int, c4 int, c5 int, c6 int, c7 int, c8 int, c9 int, c10 int, c11 int, c12 int);
2、訂單FEED數據的寫入，例如A業務係統，寫入訂單的c1,c2字段。B業務係統，寫入訂單的c3,c4字段。......
使用on conflict do something語法，進行訂單屬性的合並。
insert into feed (id, c1, c2) values (2,2,30001) on conflict (id) do update set c1=excluded.c1, c2=excluded.c2 ;
insert into feed (id, c3, c4) values (2,99,290001) on conflict (id) do update set c3=excluded.c3, c4=excluded.c4 ;
3、建立訂單FEED的實時監測規則，當滿足條件時，向PostgreSQL的異步消息中發送消息。監聽該通道的APP，循環從異步消息獲取數據，即可滿足消息的實時消費。
規則可以保留在TABLE中，也可以寫在觸發器代碼中，也可以寫在UDF代碼中。
3.1、如果數據是批量寫入的，可以使用語句級觸發器，降低觸發器函數被調用的次數，提高寫入吞吐。
create or replace function tg1() returns trigger as $$
declare
begin
-- 規則定義，實際使用時，可以聯合規則定義表
-- c2大於1000時，發送異步消息
perform pg_notify('channel_1', 'Resone:c2 overflow::'||row_to_json(inserted)) from inserted where c2>1000;
-- 多個規則，寫單個notify的方法。
-- perform pg_notify(
-- 'channel_1',
-- case
-- when c2>1000 then 'Resone:c2 overflow::'||row_to_json(inserted)
-- when c1>200 then 'Resone:c1 overflow::'||row_to_json(inserted)
-- end
-- )
-- from inserted
-- where
-- c2 > 1000
-- or c1 > 200;
-- 多個規則，可以寫多個notify，或者合並成一個NOTIFY。
return null;
end;
$$ language plpgsql strict;
3.2、如果數據是單條寫入的，可以使用行級觸發器。（本例後麵的壓測使用這個）
create or replace function tg2() returns trigger as $$
declare
begin
-- 規則定義，實際使用時，可以聯合規則定義表
-- c2大於9999時，發送異步消息
perform pg_notify('channel_1', 'Resone:c2 overflow::'||row_to_json(NEW)) where NEW.c2>9999;
-- 多個規則，調用單個notify，寫一個CHANNEL的方法。
-- perform pg_notify(
-- 'channel_1',
-- case
-- when c2>1000 then 'Resone:c2 overflow::'||row_to_json(NEW)
-- when c1>200 then 'Resone:c1 overflow::'||row_to_json(NEW)
-- end
-- )
-- where
-- NEW.c2 > 10000
-- or NEW.c1 > 200;
-- 多個規則，調用單個notify，寫多個CHANNEL的方法。
-- perform pg_notify(
-- case
-- when c2>1000 then 'channel_1'
-- when c1>200 then 'channel_2'
-- end,
-- case
-- when c2>1000 then 'Resone:c2 overflow::'||row_to_json(NEW)
-- when c1>200 then 'Resone:c1 overflow::'||row_to_json(NEW)
-- end
-- )
-- where
-- NEW.c2 > 1000
-- or NEW.c1 > 200;
-- 多個規則，可以寫多個notify，或者合並成一個NOTIFY。
-- 例如
-- perform pg_notify('channel_1', 'Resone:c2 overflow::'||row_to_json(NEW)) where NEW.c2 > 1000;
-- perform pg_notify('channel_2', 'Resone:c1 overflow::'||row_to_json(NEW)) where NEW.c1 > 200;
-- 也可以把規則定義在TABLE裏麵，實現動態的規則
-- 規則不要過於冗長，否則會降低寫入的吞吐，因為是串行處理規則。
-- udf的輸入為feed類型以及rule_table類型，輸出為boolean。判斷邏輯定義在UDF中。
-- perfrom pg_notify(channel_column, resone_column||'::'||row_to_json(NEW)) from rule_table where udf(NEW::feed, rule_table);
return null;
end;
$$ language plpgsql strict;
3.3、如上代碼中所述，規則可以定義在很多地方。
4、創建觸發器。
4.1、語句級觸發器(批量寫入，建議采用)
create trigger tg1 after insert on feed REFERENCING NEW TABLE AS inserted for each statement execute procedure tg1();
create trigger tg2 after update on feed REFERENCING NEW TABLE AS inserted for each statement execute procedure tg1();
4.2、行級觸發器(單步寫入建議采用)，（本例後麵的壓測使用這個）
create trigger tg1 after insert on feed for each row execute procedure tg2();
create trigger tg2 after update on feed for each row execute procedure tg2();
5、協商好通道名稱。
6、應用端監聽消息通道。
listen channel_1;
接收消息：
loop
sleep ?;
get 消息;
end loop
7、寫入訂單數據，每行數據都會實時過觸發器，在觸發器中寫好了邏輯，當滿足一些規則時，向協商好的消息通道發送消息。
postgres=# insert into feed (id, c1, c2) values (2,2,30001) on conflict (id) do update set c1=excluded.c1, c2=excluded.c2 ;
INSERT 0 1
8、接收到的消息樣本如下：
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":2,"c1":2,"c2":30001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
9、批量插入
postgres=# insert into feed (id, c1, c2) select id,random()*100, random()*1001 from generate_series(1,10000) t(id) on conflict (id) do update set c1=excluded.c1, c2=excluded.c2 ;
INSERT 0 10000
Time: 59.528 ms
一次接收到的樣本如下：
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":362,"c1":92,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":4061,"c1":90,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":4396,"c1":89,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":5485,"c1":72,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":6027,"c1":56,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":6052,"c1":91,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":7893,"c1":84,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":8158,"c1":73,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
10、更新數據
postgres=# update feed set c1=1;
UPDATE 10000
Time: 33.444 ms
接收到的異步消息樣本如下：
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":1928,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":2492,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":2940,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":2981,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":4271,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":4539,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":7089,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":7619,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":8001,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":8511,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":8774,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":9394,"c1":1,"c2":1001,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 38445.
壓測
1、假設每1萬條記錄中，有一條異常記錄需要推送，這樣的頻率算是比較現實的。
vi test.sql
\set id random(1,10000000)
\set c1 random(1,1001)
\set c2 random(1,10000)
insert into feed (id, c1, c2) values (:id, :c1, :c2) on conflict (id) do update set c1=excluded.c1, c2=excluded.c2 ;
2、壓測結果，167190 行/s處理吞吐。
transaction type: ./test.sql
scaling factor: 1
query mode: prepared
number of clients: 56
number of threads: 56
duration: 120 s
number of transactions actually processed: 20060111
latency average = 0.335 ms
latency stddev = 0.173 ms
tps = 167148.009836 (including connections establishing)
tps = 167190.475312 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.002 \set id random(1,10000000)
0.001 \set c1 random(1,1001)
0.000 \set c2 random(1,10000)
0.332 insert into feed (id, c1, c2) values (:id, :c1, :c2) on conflict (id) do update set c1=excluded.c1, c2=excluded.c2 ;
3、監聽到的異步消息采樣
postgres=# listen channel_1;
LISTEN
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":3027121,"c1":393,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 738.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":5623104,"c1":177,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 758.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":3850742,"c1":365,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 695.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":5244809,"c1":55,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 716.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":4062585,"c1":380,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 722.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":8536437,"c1":560,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 695.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":7327211,"c1":365,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 728.
Asynchronous notification "channel_1" with payload "Resone:c2 overflow::{"id":431739,"c1":824,"c2":10000,"c3":null,"c4":null,"c5":null,"c6":null,"c7":null,"c8":null,"c9":null,"c10":null,"c11":null,"c12":null}" received from server process with PID 731.
單實例分表的schemaless設計
請參考如下用法或案例，目的是自動建表，自動分片。
《PostgreSQL 在鐵老大訂單係統中的schemaless設計和性能壓測》
《PostgreSQL 按需切片的實現(TimescaleDB插件自動切片功能的plpgsql schemaless實現)》
《PostgreSQL 時序最佳實踐 - 證券交易係統數據庫設計 - 阿裏雲RDS PostgreSQL最佳實踐》
jdbc 異步消息使用例子
https://jdbc.postgresql.org/documentation/81/listennotify.html
import java.sql.*;
public class NotificationTest {
public static void main(String args[]) throws Exception {
Class.forName("org.postgresql.Driver");
String url = "jdbc:postgresql://localhost:5432/test";
// Create two distinct connections, one for the notifier
// and another for the listener to show the communication
// works across connections although this example would
// work fine with just one connection.
Connection lConn = DriverManager.getConnection(url,"test","");
Connection nConn = DriverManager.getConnection(url,"test","");
// Create two threads, one to issue notifications and
// the other to receive them.
Listener listener = new Listener(lConn);
Notifier notifier = new Notifier(nConn);
listener.start();
notifier.start();
}
}
class Listener extends Thread {
private Connection conn;
private org.postgresql.PGConnection pgconn;
Listener(Connection conn) throws SQLException {
this.conn = conn;
this.pgconn = (org.postgresql.PGConnection)conn;
Statement stmt = conn.createStatement();
stmt.execute("LISTEN mymessage");
stmt.close();
}
public void run() {
while (true) {
try {
// issue a dummy query to contact the backend
// and receive any pending notifications.
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery("SELECT 1");
rs.close();
stmt.close();
org.postgresql.PGNotification notifications[] = pgconn.getNotifications();
if (notifications != null) {
for (int i=0; i<notifications.length; i++) {
System.out.println("Got notification: " + notifications[i].getName());
}
}
// wait a while before checking again for new
// notifications
Thread.sleep(500);
} catch (SQLException sqle) {
sqle.printStackTrace();
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
}
}
class Notifier extends Thread {
private Connection conn;
public Notifier(Connection conn) {
this.conn = conn;
}
public void run() {
while (true) {
try {
Statement stmt = conn.createStatement();
stmt.execute("NOTIFY mymessage");
stmt.close();
Thread.sleep(2000);
} catch (SQLException sqle) {
sqle.printStackTrace();
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
}
}
libpq 異步消息的使用方法
https://www.postgresql.org/docs/10/static/libpq-notify.html
觸發器的用法
https://www.postgresql.org/docs/10/static/sql-createtrigger.html
注意事項
1、異步消息快速接收，否則會占用實例 $PGDATA/pg_notify 的目錄空間。
2、異步消息上限，沒有上限，和存儲有個。
buffer大小：
/*
* The number of SLRU page buffers we use for the notification queue.
*/
#define NUM_ASYNC_BUFFERS 8
3、異步消息可靠性，每個異步消息通道，PG都會跟蹤監聽這個通道的會話已接收到的消息的位置偏移。
新發起的監聽，隻從監聽時該通道的最後偏移開始發送，該偏移之前的消息不會被發送。
消息接收後，如果沒有任何監聽需要，則會被清除。
監聽消息通道的會話，需要持久化，也就是說會話斷開的話，（未接收的消息，以及到會話重新監聽這段時間，新產生的消息，都收不到）
4、如果需要強可靠性（替換掉異步消息，使用持久化的模式）
方法：觸發器內pg_notify改成insert into feedback_table ....;
持久化消息的消費方法，改成如下（閱後即焚模式）：
with t1 as (select ctid from feedback_table order by crt_time limit 100)
delete from feedback_table where
ctid = any (array(select ctid from t1))
returning *;
持久化消息，一樣能滿足10萬行以上的消費能力（通常異常消息不會那麼多，所以這裏可以考慮使用單個異常表，多個訂單表）。
隻不過會消耗更多的RDS PG的IOPS，（產生寫 WAL，VACUUM WAL。）
其他
1、已推送的異常，當數據更新後，可能會被再次觸發，通過在邏輯中對比OLD value和NEW value可以來規避這個問題。本文未涉及。實際使用是可以改寫觸發器代碼。
參考
《在PostgreSQL中實現update | delete limit - CTID掃描實踐 (高效閱後即焚)》
《(流式、lambda、觸發器)實時處理大比拚 - 物聯網(IoT)\金融,時序處理最佳實踐》
《PostgreSQL 10.0 preview 功能增強 - 觸發器函數內置中間表》
https://www.postgresql.org/docs/10/static/sql-createtrigger.html
https://jdbc.postgresql.org/documentation/81/listennotify.html
https://www.postgresql.org/docs/10/static/libpq-notify.html
《(流式、lambda、觸發器)實時處理大比拚 - 物聯網(IoT)\金融,時序處理最佳實踐》
最後更新：2017-11-12 02:06:33