ApsaraMQ for RocketMQ:Message filtering

Scenarios

Configuration method

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

• Send messages

  Specify a tag for each message before the message is sent.

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

• Subscribe to all messages

  If you want to configure a consumer to subscribe to all 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 you want to configure a consumer 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 messages of multiple types

  If you want to configure a consumer to subscribe to messages of multiple types in a topic, specify corresponding tags and separate them 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 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 latest subscription.

      // 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;
          }
      });                

Sample code

• Send a message.

  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 this attribute is not configured for a message or the attribute value of 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 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 both the tag and custom attribute.

  Consumer consumer = ONSFactory.createConsumer(properties);
  // Subscribe to only messages with tagA and their values of the custom attribute price are 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 both 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 to only messages whose value of the custom attribute region is hangzhou and 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 set for the message in the producer is 50.

• Subscribe to all messages in the topic.

  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 custom attribute is not configured for a message when the producer sends the 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 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;
      }
  });