ApsaraMQ for RocketMQ：訊息過濾

情境樣本

配置方式

雲訊息佇列 RocketMQ 版支援通過SDK配置Tag過濾功能，分別在訊息發送和訂閱代碼中設定訊息Tag和訂閱訊息Tag。SDK詳細資料，請參見SDK參考概述。訊息發送端和消費端的代碼配置方法如下：

• 發送訊息

  發送訊息時，每條訊息必須指明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;
          }
      });                

範例程式碼

• 發送訊息

  同時設定訊息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;
      }
  });