ApsaraMQ for RocketMQ：訊息過濾

功能描述

訊息過濾功能指訊息生產者向Topic中發送訊息時，設定訊息屬性對訊息進行分類，消費者訂閱Topic時，根據訊息屬性設定過濾條件對訊息進行過濾，只有符合過濾條件的訊息才會被投遞到消費端進行消費。

消費者訂閱Topic時若未設定過濾條件，無論訊息發送時是否有設定過濾屬性，Topic中的所有訊息都將被投遞到消費端進行消費。

Tag過濾

Tag，即訊息標籤，用於對某個Topic下的訊息進行分類。雲訊息佇列 RocketMQ 版的生產者在發送訊息時，指定訊息的Tag，消費者需根據已經指定的Tag來進行訂閱。

情境樣本

過濾示意圖如下所示。

配置方式

• 發送訊息

  發送訊息時，每條訊息必須指明Tag。

      Message msg = new Message("MQ_TOPIC","TagA","Hello MQ".getBytes());                

• 訂閱所有Tag

  消費者如需訂閱某Topic下所有類型的訊息，Tag用星號（*）表示。

      consumer.subscribe("MQ_TOPIC", "*", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• 訂閱單個Tag

  消費者如需訂閱某Topic下某一種類型的訊息，請明確標明Tag。

      consumer.subscribe("MQ_TOPIC", "TagA", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• 訂閱多個Tag

  消費者如需訂閱某Topic下多種類型的訊息，請在多個Tag之間用兩個豎線（||）分隔。

      consumer.subscribe("MQ_TOPIC", "TagA||TagB", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

• 錯誤樣本

  同一個消費者多次訂閱某個Topic下的Tag，以最後一次訂閱的Tag為準。

      //如下錯誤碼中，Consumer只能訂閱到MQ_TOPIC下TagB的訊息，而不能訂閱TagA的訊息。
      consumer.subscribe("MQ_TOPIC", "TagA", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });
      consumer.subscribe("MQ_TOPIC", "TagB", new MessageListener() {
          public Action consume(Message message, ConsumeContext context) {
              System.out.println(message.getMsgID());
              return Action.CommitMessage;
          }
      });                

SQL屬性過濾

SQL屬性過濾是在訊息發送時設定訊息的自訂屬性，消費者訂閱時使用SQL文法設定過濾運算式，根據自訂屬性過濾訊息，雲訊息佇列 RocketMQ 版根據運算式的邏輯進行計算，將合格訊息投遞到消費端。

使用限制

情境樣本

過濾示意圖如下所示。

配置方式

範例程式碼

• 發送訊息

  同時設定訊息Tag和自訂屬性。

  Producer producer = ONSFactory.createProducer(properties);
  // 設定Tag的值為tagA。
  Message msg = new Message("topicA", "tagA", "Hello MQ".getBytes());
  // 設定自訂屬性region為hangzhou。
  msg.putUserProperties("region", "hangzhou");
  // 設定自訂屬性price為50。
  msg.putUserProperties("price", "50");
  SendResult sendResult = producer.send(msg);

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

  Consumer consumer = ONSFactory.createConsumer(properties);
  // 只訂閱屬性region為hangzhou的訊息，若訊息中未定義屬性region或屬性值不是hangzhou，則訊息不會被投遞到消費端。
  consumer.subscribe("topicA", MessageSelector.bySql("region IS NOT NULL AND region = 'hangzhou'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  預期結果：樣本中發送的訊息屬性符合訂閱的過濾條件，訊息被投遞到消費端。

• 同時根據Tag和自訂屬性訂閱訊息。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // 只訂閱Tag的值為tagA且屬性price大於30的訊息。
  consumer.subscribe("topicA", MessageSelector.bySql("TAGS IS NOT NULL AND price IS NOT NULL AND TAGS = 'tagA' AND price > 30 "), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  預期結果：樣本中發送的訊息Tag和自訂屬性符合訂閱的過濾條件，訊息被投遞到消費端。

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

  Consumer consumer = ONSFactory.createConsumer(properties);
  // 只訂閱region為hangzhou且屬性price小於20的訊息。
  consumer.subscribe("topicA", MessageSelector.bySql("region IS NOT NULL AND price IS NOT NULL AND region = 'hangzhou' AND price < 20"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  預期結果：訊息不會被投遞到消費端。訂閱的過濾條件中price小於20，發送的訊息中price屬性值為50，不符合訂閱過濾條件。

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

  Consumer consumer = ONSFactory.createConsumer(properties);
  // 若需要訂閱Topic中的所有訊息，需要將SQL運算式的值設定為“TRUE”。
  consumer.subscribe("topicA", MessageSelector.bySql("TRUE"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });

  預期結果：Topic中的所有訊息都將被投遞到消費端進行消費。

• 錯誤樣本

  訊息發送時未自訂某屬性，消費端在訂閱時未判斷該屬性是否存在直接使用，則訊息不會被投遞給消費端。

  Consumer consumer = ONSFactory.createConsumer(properties);
  // 屬性product在發送訊息時未定義，過濾失敗，訊息不會被投遞至消費端。
  consumer.subscribe("topicA", MessageSelector.bySql("product = 'MQ'"), new MessageListener() {
      public Action consume(Message message, ConsumeContext context) {
          System.out.printf("Receive New Messages: %s %n", message);
          return Action.CommitMessage;
      }
  });               

更多資訊

• 同一個Group ID下的消費者執行個體與Topic的訂閱關係需保持一致，更多資訊，請參見訂閱關係一致。
• 合理使用Topic和Tag來過濾訊息可以讓業務更清晰，更多資訊，請參見Topic與Tag最佳實務。