脈絡:訊息式系統的關鍵好處之一是參與者之間的鬆散耦合——收發雙方對彼此的身分不做(或幾乎不做)假設。接收端從通道取回訊息時,通常既不知道也不在意是哪個應用程式把它放上來的;訊息依定義就是自我完備的,不與特定寄件端關聯。這是訊息式系統的架構強項之一。

  • 若我們不確定訊息會去哪,要怎麼評估「改動訊息格式」的衝擊?
  • 若我們不知道是哪個應用程式發布了某則訊息,就很難修正該訊息的問題。

幾條走不通的路#

靠 Control Bus 收集#

Control Bus 監控每個處理訊息之元件的狀態,但它不關心個別訊息走了什麼路徑。我們可以修改每個元件,把流經它的每則訊息的唯一識別碼發布到控制匯流排,再收集到一個共同資料庫(Message Store)。

靠訊息 ID 追蹤#

因此我們得另外指定一個「從進站訊息複製到出站訊息」的新鍵,兩則訊息才能事後被關聯起來。

解法#

與其「替訊息貼標籤來標識其路徑」,不如讓訊息自己收集一份它走過的元件清單:只要訊息系統中每個元件都帶有唯一識別碼,每個元件就能把自己的識別碼加進它所發布的每則訊息中

Message History 維護訊息所經過的所有元件清單,每個處理該訊息的元件(包含原始產生者)都往清單加一筆

圖 11-6:Message History 解法示意

一則訊息可能來自多則訊息#

若想在 Message History 中表現這種情境,有兩個選擇:

  • 完整追蹤——把 Message History 強化成階層式樹狀結構。由於樹狀結構的遞迴本質,單一節點底下可以存放多份訊息歷程。
  • 只保留一份——維持簡單清單,只保留其中一則進站訊息的歷程。若某則進站訊息對結果比其他輔助訊息重要得多,這個作法效果不錯。

什麼時候最有用#

若「管理訊息所走的路徑」本身很重要,Process Manager 會很有幫助——它為每則進站觸發訊息建立一個流程實例,訊息流經各元件的過程因而被集中管理,也就不必替每則訊息貼上歷程了。

另一個重要好處:避免無窮迴圈#

替訊息配上歷程,在用 Publish-Subscribe Channel 傳播事件時還有另一個重要好處。

考慮一個透過發布訂閱通道把地址變更傳播到多個系統的系統:每次地址變更都廣播給所有有興趣的系統,這對新增系統極有彈性——新系統會自動收到廣播訊息,完全不必改動既有的訊息系統。

但假設客服系統也是把地址存在應用程式資料庫中的系統之一:

  1. 資料庫欄位的每次變更都觸發一則訊息,通知所有系統
  2. 依發布訂閱的本質,所有訂閱「地址已變更」通道的系統都會收到該事件
  3. 客服系統自己也必須訂閱這個通道(才能收到其他系統,例如自助網站,所做的更新)
  4. 於是客服系統會收到自己剛剛發布的那則訊息
  5. 這則收到的訊息造成一次資料庫更新,而更新又觸發另一則「地址已變更」訊息

我們就掉進了「地址已變更」訊息的無窮迴圈。

要避免這種迴圈,訂閱的應用程式可以檢視 Message History 判斷該訊息是否源自自己,若是就忽略它。

範例:TIBCO ActiveEnterprise

許多 EAI 整合套件支援 Message History。例如 ActiveEnterprise 每則訊息的表頭都含有一個 tracking 欄位,維護該訊息所經過的所有元件清單。

這裡要注意:TIBCO ActiveEnterprise 元件會把「所消費之訊息的 message ID」指派給出站訊息。這讓跨元件追蹤訊息更容易,但也意味著 message ID 不是全系統唯一的屬性——多則個別訊息會共用同一個 ID。例如實作 Recipient List 時,ActiveEnterprise 會把所消費訊息的 ID 轉給每則出站訊息。

以下是一則穿過多個元件(包含兩個 Integration Manager 流程 OrderProcessVerifyCustomerStub)的訊息傾印:

tw.training.customer.verify.response
{
    RVMSG_INT          2 ^pfmt^        10
    RVMSG_INT          2 ^ver^         30
    RVMSG_INT          2 ^type^        1
    RVMSG_RVMSG 108 ^data^
    {
        RVMSG_STRING 23 ^class^             "VerifyCustomerResponse"
        RVMSG_INT       4 ^idx^         1
        RVMSG_STRING     6 CUSTOMER_ID "12345"
        RVMSG_STRING     6 ORDER_ID         "22222"
        RVMSG_INT       4 RESULT        0
    }
    RVMSG_RVMSG 150 ^tracking^
    {
        RVMSG_STRING 28 ^id^ "4OEaDEoiBIpcYk6qihzzwB5Uzzw"
        RVMSG_STRING 41 ^1^        "imed_debug_engine1-OrderProcess-Job-4300"
        RVMSG_STRING 47 ^2^        "imed_debug_engine1-VerifyCustomerStub-Job-4301"
    }
}