ApsaraMQ for RocketMQ：訊息過濾

應用情境

雲訊息佇列 RocketMQ 版作為發布訂閱模型的訊息中介軟體廣泛應用於上下遊業務整合情境。在實際業務情境中，同一個主題下的訊息往往會被多個不同的下遊業務方處理，各下遊的處理邏輯不同，只關注自身邏輯需要的訊息子集。

使用雲訊息佇列 RocketMQ 版的訊息過濾功能，可以協助消費者更高效地過濾自己需要的訊息集合，避免大量無效訊息投遞給消費者，降低下遊系統處理壓力。

雲訊息佇列 RocketMQ 版主要解決的單個業務域即同一個主題內不同訊息子集的過濾問題，一般是基於同一業務下更具體的分類進行過濾匹配。如果是需要對不同業務域的訊息進行拆分，建議使用不同主題處理不同業務域的訊息。

功能概述

訊息過濾定義

過濾的含義指的是將合格訊息投遞給消費者，而不是將匹配到的訊息過濾掉。

雲訊息佇列 RocketMQ 版的訊息過濾功能通過生產者和消費者對訊息的屬性、標籤進行定義，並在雲訊息佇列 RocketMQ 版服務端根據過濾條件進行篩選匹配，將合格訊息投遞給消費者進行消費。

訊息過濾原理

訊息過濾分類

具體的使用方式及樣本，請參見下文的Tag標籤過濾和SQL屬性過濾。

訂閱關係一致性

過濾運算式屬於訂閱關係的一部分，雲訊息佇列 RocketMQ 版的領域模型規定，同一消費者分組內的多個消費者的訂閱關係包括過濾運算式，必須保持一致，否則可能會導致部分訊息消費不到。更多資訊，請參見訂閱關係（Subscription）。

Tag標籤過濾

Tag標籤過濾方式是雲訊息佇列 RocketMQ 版提供的基礎訊息過濾能力，基於生產者為訊息設定的Tag標籤進行匹配。生產者在發送訊息時，設定訊息的Tag標籤，消費者需指定已有的Tag標籤來進行匹配訂閱。

情境樣本

Tag標籤設定

• Tag由生產者發送訊息時設定，每條訊息允許設定一個Tag標籤。

• Tag使用可見字元，建議長度不超過128字元。

• Tag匹配時大小寫敏感，例如，TAG A和tag a屬於兩個標籤。

Tag標籤過濾規則

使用樣本

• 發送訊息，設定Tag標籤。

  Message message = messageBuilder.setTopic("topic")
                  //設定訊息索引鍵，可根據關鍵字精確尋找某條訊息。
                  .setKeys("messageKey")
                  //設定訊息Tag，用於消費端根據指定Tag過濾訊息。
                  //該樣本表示訊息的Tag設定為“TagA”。
                  .setTag("TagA")
                  //訊息體。
                  .setBody("messageBody".getBytes())
                  .build();
                      

• 訂閱訊息，匹配單個Tag標籤。

  String topic = "Your Topic";
  //只訂閱訊息標籤為“TagA”的訊息。
  FilterExpression filterExpression = new FilterExpression("TagA", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• 訂閱訊息，匹配多個Tag標籤。

  String topic = "Your Topic";
  //只訂閱訊息標籤為“TagA”、“TagB”或“TagC”的訊息。
  FilterExpression filterExpression = new FilterExpression("TagA||TagB||TagC", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• 訂閱訊息，匹配Topic中的所有訊息，不進行過濾。

  String topic = "Your Topic";
  //使用Tag標籤過濾訊息，訂閱所有訊息。
  FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

SQL屬性過濾

SQL屬性過濾是雲訊息佇列 RocketMQ 版提供的進階訊息過濾方式，通過生產者為訊息設定的屬性（Key）及屬性值（Value）進行匹配。生產者在發送訊息時可設定多個屬性，消費者訂閱時可設定SQL文法的過濾運算式過濾多個屬性。

情境樣本

訊息屬性設定

• 生產者發送訊息時可以設定訊息屬性，每個屬性都是一個自訂的索引值對（Key-Value）。

  • 屬性Key可以由字母（a-z, A-Z）、數字（0-9）、底線（_）組成。

  • 屬性Key的第一個字元必須是字母或底線，不能是數字。

• 每條訊息支援設定多個屬性。

SQL屬性過濾規則

SQL屬性過濾使用SQL92文法作為過濾規則運算式，文法規範如下：

使用樣本

• 發送訊息，同時設定訊息Tag標籤和自訂屬性。

  Message message = messageBuilder.setTopic("topic")
                  //設定訊息索引鍵，可根據關鍵字精確尋找某條訊息。
                  .setKeys("messageKey")
                  //設定訊息Tag，用於消費端根據指定Tag過濾訊息。
                  //該樣本表示訊息的Tag設定為“messageTag”。
                  .setTag("messageTag")
                  //訊息也可以設定自訂的分類屬性，例如環境標籤、地區、邏輯分支。
                  //該樣本表示為訊息自訂一個屬性，該屬性為地區，屬性值為杭州。
                  .addProperty("Region", "Hangzhou")
                  //訊息體。
                  .setBody("messageBody".getBytes())
                  .build();

• 訂閱訊息，根據單個自訂屬性匹配訊息。

  String topic = "topic";
  //只訂閱地區屬性為杭州的訊息。
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND Region='Hangzhou'", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• 訂閱訊息，同時根據多個自訂屬性匹配訊息。

  String topic = "topic";
  //只訂閱地區屬性為杭州且價格屬性大於30的訊息。
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND price IS NOT NULL AND Region = 'Hangzhou' AND price > 30", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• 訂閱訊息，匹配Topic中的所有訊息，不進行過濾。

  String topic = "topic";
  //訂閱所有訊息。
  FilterExpression filterExpression = new FilterExpression("True", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

使用建議

合理劃分主題和Tag標籤

訊息過濾常見問題

多個消費者訂閱同一個Topic下的不同Tag，出現訊息丟失情況。

可能原因：若多個消費者是通過同一個消費者分組（Group ID）訂閱的指定Topic，則所有消費者的過濾條件即訂閱的Tag要一致，否則會出現訂閱關係不一致，導致部分訊息丟失。

消費者設定了過濾條件，訊息消費數量是按照過濾前還是過濾後計算呢？

按照過濾後的訊息數量計算。

消費者線上無消費訊息，但Group有堆積。

採用SQL/TAG消費過濾的方式，未被過濾條件命中的訊息會計算為訊息堆積，堆積量的計算如下所示。

相關文檔

完整的訊息收發範例程式碼，請參見SDK參考概述。