PostgreSQL , 異步消息 , 觸發器 , 規則 , insert on conflict , 實時分析
在很多業務系統中，為了定位問題、營運需要、分析需要或者其他需求，會在業務中設定埋點，記錄使用者的行為在業務系統中産生的日志，也叫FEED日志。
比如訂單系統、在業務系統中環環相扣，從購物車、下單、付款、發貨，收貨（還有糾紛、退款等等），一筆訂單通常會産生若幹相關聯的記錄。
每個環節産生的屬性可能是不一樣的，有可能有新的屬性産生，也有可能變更已有的屬性值。
為了便于分析，通常有必要将訂單在整個過程中産生的若幹記錄（若幹屬性），合并成一條記錄（訂單大寬表）。
通常業務系統會将實時産生的訂單FEED資料寫入消息隊列，消息隊列使得資料變成了流動的資料：
<a href="https://github.com/digoal/blog/blob/master/201707/20170706_01.md">《從人類河流文明 洞察 資料流動的重要性》</a>
資料通過消息隊列消費後，實時寫入RDS PG，在RDS PG進行訂單FEED的合并，寫入OSS外部表。(支援壓縮格式，換算成裸資料的寫入OSS的速度約100MB/s/會話)
HDB PG從OSS外部表讀取(支援壓縮格式，換算成裸資料的讀取OSS的速度約100MB/s/資料節點)，并将訂單FEED資料合并到全量訂單表。
<a href="https://github.com/digoal/blog/blob/master/201707/20170728_01.md">《打造雲端流計算、線上業務、資料分析的業務資料閉環 - 阿裡雲RDS、HybridDB for PostgreSQL最佳實踐》</a>
資料進入HDB PG後，通過規則SQL，從全量訂單表中，挖掘異常資料（或者分析）。
通過這種方案，實作了海量訂單FEED資料的分鐘級準實時分析。
這個方案已支撐了雙十一業務，高吞吐、低延遲，絲般柔滑。
技術永遠是為業務服務的，分鐘級延遲雖然說已經很高了，但是在一些極端情況下，可能需要更低的延遲。
實際上RDS PostgreSQL還有更強的殺手锏，可以實作毫秒級的異常FEED資料發現和回報。
流式處理+異步消息，方法如下：
1、通過觸發機制結合異步消息通道實作。
2、通過pipeline，流式SQL結合異步消息通道實作。
應用程式監聽消息通道(listen channel)，資料庫則将異常資料寫入到消息通道(notify channel, message)。實作異常資料的主動異步推送。
1、分執行個體，提高系統級吞吐。（例如單執行個體處理能力是15萬行/s，那麼100個執行個體，可以支撐1500萬行/s的實時處理。）
例如：
映射關系：
2、執行個體内使用分表，提高單執行個體并行處理吞吐。當規則衆多時，分表可以提高單執行個體的規則處理吞吐。
例如
HDB PG依舊保留，用于PB級資料量的海量資料實時分析。
資料通路依舊采用OSS，批量導入的方式。
1、建立訂單feed全寬表（當然，我們也可以使用jsonb字段來存儲所有屬性。因為PostgreSQL支援JSONB類型哦。PostgreSQL支援的多值類型還有hstore, xml等。）
2、訂單FEED資料的寫入，例如A業務系統，寫入訂單的c1,c2字段。B業務系統，寫入訂單的c3,c4字段。......
使用on conflict do something文法，進行訂單屬性的合并。
3、建立訂單FEED的實時監測規則，當滿足條件時，向PostgreSQL的異步消息中發送消息。監聽該通道的APP，循環從異步消息擷取資料，即可滿足消息的實時消費。
規則可以保留在TABLE中，也可以寫在觸發器代碼中，也可以寫在UDF代碼中。
3.1、如果資料是批量寫入的，可以使用語句級觸發器，降低觸發器函數被調用的次數，提高寫入吞吐。
3.2、如果資料是單條寫入的，可以使用行級觸發器。（本例後面的壓測使用這個）
3.3、如上代碼中所述，規則可以定義在很多地方。
4、建立觸發器。
4.1、語句級觸發器(批量寫入，建議采用)
4.2、行級觸發器(單步寫入建議采用)，（本例後面的壓測使用這個）
5、協商好通道名稱。
6、應用端監聽消息通道。
7、寫入訂單資料，每行資料都會實時過觸發器，在觸發器中寫好了邏輯，當滿足一些規則時，向協商好的消息通道發送消息。
8、接收到的消息樣本如下：
9、批量插入
一次接收到的樣本如下：
10、更新資料
接收到的異步消息樣本如下：
1、假設每1萬條記錄中，有一條異常記錄需要推送，這樣的頻率算是比較現實的。
2、壓測結果，167190 行/s處理吞吐。
3、監聽到的異步消息采樣
請參考如下用法或案例，目的是自動建表，自動分片。
<a href="https://github.com/digoal/blog/blob/master/201709/20170927_03.md">《PostgreSQL 在鐵老大訂單系統中的schemaless設計和性能壓測》</a>
<a href="https://github.com/digoal/blog/blob/master/201711/20171102_02.md">《PostgreSQL 按需切片的實作(TimescaleDB插件自動切片功能的plpgsql schemaless實作)》</a>
<a href="https://github.com/digoal/blog/blob/master/201705/20170511_01.md">《PostgreSQL schemaless 的實作》</a>
<a href="https://github.com/digoal/blog/blob/master/201704/20170417_01.md">《PostgreSQL 時序最佳實踐 - 證券交易系統資料庫設計 - 阿裡雲RDS PostgreSQL最佳實踐》</a>
<a href="https://jdbc.postgresql.org/documentation/81/listennotify.html">https://jdbc.postgresql.org/documentation/81/listennotify.html</a>
<a href="https://www.postgresql.org/docs/10/static/libpq-notify.html">https://www.postgresql.org/docs/10/static/libpq-notify.html</a>
<a href="https://www.postgresql.org/docs/10/static/sql-createtrigger.html">https://www.postgresql.org/docs/10/static/sql-createtrigger.html</a>
<a href="https://github.com/digoal/blog/blob/master/201303/20130311_01.md">《PostgreSQL 觸發器 用法詳解 1》</a>
<a href="https://github.com/digoal/blog/blob/master/201303/20130311_02.md">《PostgreSQL 觸發器 用法詳解 2》</a>
1、異步消息快速接收，否則會占用執行個體 <code>$PGDATA/pg_notify</code> 的目錄空間。
2、異步消息上限，沒有上限，和存儲有個。
buffer大小：
3、異步消息可靠性，每個異步消息通道，PG都會跟蹤監聽這個通道的會話已接收到的消息的位置偏移。
新發起的監聽，隻從監聽時該通道的最後偏移開始發送，該偏移之前的消息不會被發送。
消息接收後，如果沒有任何監聽需要，則會被清除。
監聽消息通道的會話，需要持久化，也就是說會話斷開的話，（未接收的消息，以及到會話重新監聽這段時間，新産生的消息，都收不到）
4、如果需要強可靠性（替換掉異步消息，使用持久化的模式）
方法：觸發器内pg_notify改成<code>insert into feedback_table ....;</code>
持久化消息的消費方法，改成如下（閱後即焚模式）：
持久化消息，一樣能滿足10萬行以上的消費能力（通常異常消息不會那麼多，是以這裡可以考慮使用單個異常表，多個訂單表）。
隻不過會消耗更多的RDS PG的IOPS，（産生寫 WAL，VACUUM WAL。）
1、已推送的異常，當資料更新後，可能會被再次觸發，通過在邏輯中對比OLD value和NEW value可以來規避這個問題。本文未涉及。實際使用是可以改寫觸發器代碼。
<a href="https://github.com/digoal/blog/blob/master/201608/20160827_01.md">《在PostgreSQL中實作update | delete limit - CTID掃描實踐 (高效閱後即焚)》</a>
<a href="https://github.com/digoal/blog/blob/master/201705/20170518_01.md">《(流式、lambda、觸發器)實時處理大比拼 - 物聯網(IoT)\金融,時序處理最佳實踐》</a>
<a href="https://github.com/digoal/blog/blob/master/201704/20170401_01.md">《PostgreSQL 10.0 preview 功能增強 - 觸發器函數内置中間表》</a>