場景示例

配置方式

云消息隊列 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;
      }
  });