Listen 和 Notify 是PG很有意思的一個功能，可以用來進行多應用間的通信。它們可以在SQL中使用，也可以用C、JDBC里面的API調用。下面介紹一下其使用方法和內核實現。 ## 使用方法 用一個簡單的例子，來看一下Listen/Notify如何使用。假設我們有兩個應用A和B，部署在不同的機器上：A機器處理前端用戶請求，同時需要將一些可以異步執行的任務，分配給后臺服務器B。B接收到任務并處理完成后反饋給A結果。 ~~~ --分別在A和B兩臺機器上初始化PG連接，它們互相監聽對方消息，A負責派發任務給B --機器A初始化與PG的連接 session A: listen workA; commit; --機器B初始化與PG的連接 session B: listen workerB; commit; --A派發任務給B session A: begin; notify workerB, 'do job 1001'; commit; --B接受消息 session B: begin; Asynchronous notification "worker1" with payload "1001" received from server process with PID 29826. commit; --B解析消息(可用腳本或應用實現），然后完成任務，發送反饋給A session B: .... begin; notify workA 'job 1001 done'; commit; ~~~ 利用上面的步驟，A和B兩個機器通過PG完成了通信。在上面的過程中，需要注意的是： 1. B要想接受到消息，必須在A Notify之前運行了Listen命令； 2. A需要使用事務commit操作來觸發消息發送； 3. 消息是異步發送到B的，即無論B的狀態如何，消息都會先到達PG的消息隊列（每個PG實例只有一個唯一的存放所有消息的隊列）；B要查看消息，如果使用的是psql客戶端，則需要先發送帶有事務操作的命令（如begin、commit或rollback）給PG； 4. A 如果連續發送多個消息，B會一次性收到這些消息； 5. 在C代碼里面，你可以使用如下的調用來獲取所有已到達的消息，如果沒有消息到達，則進入睡眠狀態。 ~~~ while (1) { sock = PQsocket(conn); /* Monitor socket. Sleep if there is no message */ select(sock + 1, &input_mask, NULL, NULL, NULL) /* Now check for input */ PQconsumeInput(conn); /* Loop until all notifications currently received have been handled */ while ((notify = PQnotifies(conn)) != NULL) { /* Received some message, print it out */ fprintf(stderr, "ASYNC NOTIFY of '%s' received from backend PID %d\n", notify->relname, notify->be_pid); PQfreemem(notify); } } ~~~ ## 內核實現 Listen/Notify的實現其實比較簡單。主要的數據結構是一個消息隊列（`asyncQueueControl->tail`和`asyncQueueControl->head`分別指向隊列尾和隊列頭）和一個進程狀態數組（`asyncQueueControl->backend`），如下圖所示：  消息隊列里面存放了所有進程的所有通知消息，而狀態數組存放了所有執行了Listen命令、準備接收異步消息的進程的狀態信息。狀態數組中含有每個進程已經讀取到的消息在隊列里面的位置指針。如果有了新消息，進程就從此指針往后取，直到讀取全部消息。 當一個連接的后臺進程接收到Listen命令時，先將Listen的信息記錄下來，然后在事務提交時，執行Listen操作，即把本進程放入狀態數組（參見`Exec_ListenPreCommit`函數）。 執行Notify命令時，Async_Notify函數負責把通知放入pendingNotifies鏈表。在事務Commit操作前后，執行下面的邏輯： 1. 調用PreCommit_Notify函數，將pendingNotifies鏈表中的消息，放入全局消息隊列; 2. 執行Commit操作; 3. 利用調用鏈`ProcessCompletedNotifies->SignalBackends->SendProcSignal->kill`，向其他所有狀態數組中的進程，發出通知信號。 另一方面，每個進程在接收到信號后，利用函數`HandleNotifyInterrupt`處理信號。如果當前進程處于事務中，則不立即處理消息，等到事務提交完畢，調用`prepare_for_client_read`讀取下一個用戶命令時，利用`ProcessIncomingNotify`處理消息；否則，立即調用`ProcessIncomingNotify`處理消息。`ProcessIncomingNotify`最終調用`NotifyMyFrontEnd`發送消息到客戶端： ~~~ ProcessIncomingNotify --> asyncQueueReadAllNotifications() --> NotifyMyFrontEnd ~~~ 注意，客戶端收到消息后，并不立即顯示出來，而是需要用API進行獲取。例如，psql就是在執行下一個命令時（如begin、commit），會順便把收到的消息顯示出來的。 ## 總結 Listen/Notify是一個輕量級的應用間通信機制，有了它，具有訪問數據庫能力的應用可以輕易的利用PG實現互操作。當然，由于消息隊列是存放在內存里面的，在發生實例宕機等問題時，消息將丟失，對可靠性要求高的應用，需要自己進行消息持久化（如利用PG存儲消息，進行持久化）。