脈絡:應用程式從 Message Channel 消費訊息,但它未必想消費該通道上的所有訊息,只想要其中一部分

預設行為與它的侷限#

預設情況下:

  • 若通道只有一個消費者,該通道上所有訊息都會被送給它
  • 若通道上有多個 Competing Consumers,任何訊息都可能給任何消費者,而每則訊息都會給某個消費者

消費者通常無法選擇自己消費哪些訊息——它永遠拿到「下一則」。

只要消費者願意接收通道上的所有訊息(多數時候如此),這就沒問題。但當消費者只想消費特定訊息時,這就成了問題。

舉例:處理貸款申請訊息的應用程式,可能想把 10 萬美元以下10 萬美元以上的貸款用不同方式處理。一種作法是有兩種消費者,一種處理小額、一種處理大額——但既然任何消費者都可能收到任何訊息,應用程式要怎麼確保正確的訊息去到正確的消費者?

幾條走不通的路#

  • 每個消費者照收,收到不對的就轉交給合適的消費者
  • 消費者發現不要這則訊息時,把它放回通道
  • 每個消費者拿到每則訊息的副本,丟掉不要的
  • 由訊息系統為每種訊息型別定義獨立通道

準則應該是接收端的屬性,而不是通道的屬性;通道上的訊息則應標明自己符合哪些準則。

解法#

過濾過程有三個部分:

  1. 指定值的生產者(Specifying Producer)——在送出訊息前指定該訊息的選擇值
  2. 選擇值(Selection Value)——訊息中的一或多個值,讓消費者據此決定是否選取該訊息
  3. 選擇性消費者(Selective Consumer)——只接收符合自身選擇準則的訊息

流程是:寄件端建立訊息時一併設定選擇值並送出 → 訊息系統遞送時,Selective Consumer 測試該值是否符合自身準則 → 符合就接收,並透過回呼交給應用程式處理。

成組使用#

Selective Consumer 常成組使用——一個過濾某組準則、另一個過濾另一組。以貸款為例,一個選擇 amount <= $100,000、另一個選擇 amount > $100,000於是每個消費者只拿到自己感興趣的那類貸款

搭配 Point-to-Point Channel#

多個 Selective Consumer 用在點對點通道上時,它們實際上成了「有選擇性的 Competing Consumers」。若兩個消費者的準則重疊、而某則訊息同時符合兩者,任一個都可能消費它

搭配 Publish-Subscribe Channel#

每則訊息都會被送給每個訂閱者,但訂閱者會直接忽略不符自身準則的那份副本

消費者一旦決定忽略某則訊息,訊息系統就能丟棄它——因為它已成功遞送且永遠不會被消費。訊息系統甚至可以最佳化這個過程:連「它知道消費者會忽略的訊息」都不遞送,藉此減少必須產生與傳輸的訊息副本數量。

這種「丟棄被忽略訊息」的行為,與所使用的 Guaranteed Delivery、Durable Subscriber 或 Message Expiration 設定無關。

讓單一通道像多個資料型別通道#

Selective Consumer 讓單一通道表現得像多個 Datatype Channel:不同型別的訊息可以有不同的選擇值,特化於某型別的消費者就只會收到該型別的訊息。

這讓用少量通道傳送大量型別成為可能,也能在「企業所需通道數超出訊息系統所能支援」時節省通道。

訊息系統能確保只有被授權的應用程式才能成功從通道接收訊息,但它通常不會對消費者的選擇準則做授權——因此一個被授權存取該通道的惡意消費者,只要改變自己的準則就能取得未授權的訊息。要安全地把應用程式擋在外面,必須使用獨立的資料型別通道。

圖 10-9:兼具競爭與選擇性的消費者

替代方案#

Message Dispatcher#

把選擇準則內建在派發器裡,由它決定每則訊息的執行者。

就像 Message Dispatcher vs. Competing Consumers 的取捨一樣,真正的問題是:你想讓訊息系統來做派發,還是想自己實作? 若訊息系統不支援 Selective Consumer,你就別無選擇,只能用 Message Dispatcher 自己實作。

關於「預設消費者」#

若通道上沒有任何 Selective Consumer 符合某訊息的選擇值,該訊息就會像通道沒有接收者一樣被忽略。這類似於程序式程式設計中「case 敘述沒有任何 case 匹配」的問題。

於是很誘人地會想建一個「預設消費者」來收那些沒人匹配的訊息,但這行不通——它需要一個能匹配所有選擇值的運算式,因而會與所有其他消費者競爭

正確作法是:用 Message Dispatcher 實作一個帶 default 選項的 case 敘述,由 default 使用預設消費者。

Message Filter#

兩者達成的目標相近,方式卻不同:

  • Selective Consumer——所有訊息都被遞送給接收端,但每個接收端忽略不想要的
  • Message Filter——坐落在「寄件端通道」與「接收端通道」之間,只把想要的訊息轉移過去,因此不想要的訊息從未被遞送到接收端的通道,接收端根本沒東西要忽略

Content-Based Router#

這種路由器和過濾器一樣,確保通道上只有接收端想要的訊息,能提高安全性與消費者的效能

舉例:需求改成要把中額貸款($50K–$150K)與小額、大額分開處理——

  • 用 Content-Based Router:得建一個中額貸款的新通道、該通道的消費者,還要調整路由器的分流方式。而且還得擔心變更生效時,那些已經被路由到原本通道、卻還沒被消費的訊息現在跑到了錯的通道上。
  • 用 Selective Consumer:只要把兩種消費者(<$100K、>$100K)換成三種(<$50K、$50K–$150K、>$150K)就好。

理想上,訊息的選擇值該放在表頭而非內文

範例:JMS 訊息選擇器與 .NET 的 Peek / ReceiveById

JMS Message Selector#

String selector = "req_type = 'quote'";
MessageConsumer consumer = session.createConsumer(destination, selector);

這個接收端會忽略所有 request type 屬性不是 quote 的訊息,就好像那些訊息從未被遞送到該目的地一樣

.NET 的 Peek、ReceiveById 與 ReceiveByCorrelationId#

在 .NET 中,MessageQueue.Receive 本身不支援 JMS 風格的訊息選擇器。接收端能做的是用 MessageQueue.Peek 先看一眼訊息,符合準則再用 MessageQueue.Receive 從佇列讀出。

因此應改用 ReceiveById:由消費者指定它想接收之訊息的 Id 屬性值,確保拿到的就是剛才 Peek 到的那一則

.NET 的另一個選項是 ReceiveByCorrelationId——消費者指定它想接收之訊息的 CorrelationId 屬性值。