Message filtering

Sample scenario

Configuration method

ApsaraMQ for RocketMQ allows you to define code in client SDKs to filter messages by using tags. Before the producer sends messages, you must add tags to the messages and specify the tags of messages to which the consumer wants to subscribe. For information about SDKs, see Overview. The following section describes how to define code for the producer and consumer:

• Send messages

  Before you send messages, specify a tag for each message.

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

• Subscribe to all messages

  If a consumer wants to subscribe to all messages in a topic, use an asterisk (*) to specify 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 a specific type of message

  If a consumer wants to subscribe to a specific type of message 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 wants to subscribe to multiple types of messages in a topic, specify the corresponding tags. Separate multiple 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;
          }
      });                

• Incorrect example

  If a consumer has multiple subscriptions to a topic, and each subscription has a different tag, the consumer receives only messages whose tag matches the tag specified in the most recent subscription.

      // In the following code, a consumer can receive 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;
          }
      });                

Sample code

• Send messages

  Configure a tag and custom attributes for the message.

  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 custom attribute.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe to only messages whose value of the custom attribute region is hangzhou. If you do not configure this attribute for a message or the attribute value of the message is not hangzhou, the message is not 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 message sent in the example has the custom attribute region and the attribute value is hangzhou. The message meets the filter conditions and is delivered to the consumer.

• Subscribe to messages based on the tag and custom attribute.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe only to messages that have tagA 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 message sent in the example has tagA and the custom attribute price, and the value of the custom attribute price is greater than 30. The message meets the filter conditions and 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 the 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 message does not meet the filter conditions. The consumer subscribes to messages whose value of the custom attribute price is less than 20, but the value of the custom attribute price specified for the message in the producer is 50.

• Subscribe to all messages in the topic.

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

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

• Incorrect example

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

  Consumer consumer = ONSFactory.createConsumer(properties);
  // The message attribute product is not configured during message sending. The filtering fails and the message is not 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;
      }
  });