Filter messages

Feature description

The message filtering feature is implemented by configuring message attributes. Message attributes are used to classify messages. You can configure message attributes to classify the messages that producers send to topics on the Message Queue for Apache RocketMQ broker. You can configure filter conditions in consumers so that the consumers can subscribe to messages with specified attributes in the topics. Then, the broker filters the messages sent from the producers and delivers only the messages that meet the specified conditions to the consumers.

If a consumer subscribes to a topic but no filter condition is configured in the consumer, all messages in the topic are delivered to the consumer no matter whether attributes are configured for the messages when they are sent from the producers.

Filter messages by using tags

A tag is a label that classifies messages in a topic. You can add a tag to a message in aMessage Queue for Apache RocketMQ producer. The consumer subscribes to the message based on the specified tag.

Scenarios

The following figure shows the filtering process.

Configure message filtering in code

• Send messages

  A tag must be specified for each message before the message is sent.

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

• Subscribe to all types of messages

  If a consumer needs to subscribe to all types of messages in a topic, use an asterisk (*) to indicate all tags.

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

• Subscribe to messages of a specific type

  If a consumer needs to subscribe to messages of a specific type in a topic, specify the corresponding tag.

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

• Subscribe to multiple types of messages

  If a consumer needs to subscribe to multiple types of messages in a topic, separate tags with two vertical bars (||).

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

• Error example

  If a consumer subscribes to messages with specific tags in a topic multiple times, the tags in the last subscription prevail:

      // In the following code, a consumer can receive only messages with TagB in MQ_TOPIC but cannot receive messages with 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;
          }
      });                

Filter messages by using SQL attributes

Messages are filtered by using SQL attributes in the following process: Configure custom attributes for messages in the producer and define a filter expression by using SQL syntax in the consumer. Message Queue for Apache RocketMQ calculates the filter expression to filter messages that are sent from the producer based on the custom attributes and delivers messages that meet the filter conditions to the consumer.

Limits

Scenarios

The following figure shows the filtering process.

Configure message filtering in code

Sample code

• Send messages

  Configure tags and custom attributes for messages.

  Producer producer = ONSFactory.createProducer(properties);
  // Set the value of Tag to tagA. 
  Message msg = new Message("topicA", "tagA", "Hello MQ".getBytes());
  // Set the value of the custom attribute region to hangzhou. 
  msg.putUserProperties("region", "hangzhou");
  // Set the value of the custom attribute price to 50. 
  msg.putUserProperties("price", "50");
  SendResult sendResult = producer.send(msg);

• Subscribe to messages based on a single custom attribute.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe only to messages whose value of the custom attribute region is hangzhou. If this attribute is not configured for a message or the value of this attribute for the message is not hangzhou, the message will not be delivered to the consumer. 
  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;
      }
  });

  Expected results: The value of the custom attribute region of the message sent in the example meets the filter conditions defined in the consumer, and the message is delivered to the consumer.

• Subscribe to messages based on both tags and custom attributes.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe only to messages where tagA is defined and whose value of the custom attribute price is greater than 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;
      }
  });

  Expected results: The tag of the message sent in the example and the value of the custom attribute price of the message meet the filter conditions defined in the consumer, and the message is delivered to the consumer.

• Subscribe to messages based on multiple custom attributes.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe only to messages whose value of the custom attribute region is hangzhou and whose value of the custom attribute price is less than 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;
      }
  });

  Expected result: The message is not delivered to the consumer because the filter conditions defined in the consumer are not met. The consumer subscribes to messages whose value of the custom attribute price is less than 20, but the value of the custom attribute price set for the message in the producer is 50.

• Subscribe to all messages in the topic without filtering.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Set the value of the SQL expression to TRUE to subscribe to all messages in the topic. 
  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;
      }
  });

  Expected results: All messages in the topic are delivered to the consumer.

• Error example

  A specified custom attribute is not configured for a message in the producer and no logic is defined in the filter expression in the consumer to check whether the custom attribute exists. Instead, the custom attribute is directly used as a filter condition in the filter expression in the consumer. In this case, the message is not delivered to the consumer.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // The attribute product is not defined in the producer. The filtering fails and the message will not be delivered to the consumer. 
  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;
      }
  });               

References

• Consumer instances that use the same group ID must subscribe to the same topics. For more information, see Subscription consistency.
• You can use topics and tags to filter messages. If you properly use the topics and tags, the structure of your business can become clearer. For more information, see Best practices of topics and tags.