消息过滤

场景示例

配置方式

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