消息过滤

功能描述

消息过滤功能指消息生产者向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最佳实践。