ApsaraMQ for RocketMQ:Message filtering

Scenarios

ApsaraMQ for RocketMQ is a middleware service that is based on the publish-subscribe (pub/sub) model and used to integrate upstream and downstream business. In actual business scenarios, messages in a topic are processed by different downstream applications that have different consumption logic. Each application needs to consume only messages that meet its consumption logic.

You can use the message filtering feature provided by ApsaraMQ for RocketMQ to efficiently filter the messages that a consumer requires. This prevents a large number of invalid messages from being delivered to the consumer and reduces the workload in the downstream systems.

The message filtering feature of ApsaraMQ for RocketMQ solves the pain point of filtering messages in a topic and is used in scenarios in which detailed categorization is required for the same line of business. If you want to manage messages for different lines of business, we recommend that you use different topics.

Overview

What is message filtering?

The message filtering feature of ApsaraMQ for RocketMQ filters messages based on consumer-configured conditions and sends the messages that meet the conditions to the consumers.

After message attributes and tags are defined for producers and consumers, messages are filtered and matched based on the filtering conditions on the ApsaraMQ for RocketMQ broker. Only messages that meet the filtering conditions are delivered to consumers.

How does message filtering work?

Classification

For more information, see Tag-based filtering and Attribute SQL-based filtering.

Subscription consistency

Filtering expressions constitute a part of a subscription. Based on the domain model of ApsaraMQ for RocketMQ, the filtering expressions of subscriptions of consumers in the same consumer group must be the same. Otherwise, specific messages cannot be consumed. For more information, see Subscriptions.

Tag-based filtering

Tag-based filtering is a basic filtering method provided by ApsaraMQ for RocketMQ. In this method, messages are matched based on the tags that are defined by producers. Consumers use the tags to specify the messages that they want to consume.

Sample scenario

Tag setting

• Before a producer sends messages, the producer can attach only one tag to each message.

• Each tag consists of a string of characters. We recommend that each tag do not exceed 128 characters in length.

• Tags are case-sensitive. For example, TAG A and tag a are different tags.

Filtering rules

Sample code

• Specify a tag before message sending

  Message message = messageBuilder.setTopic("topic")
                  // The message key. You can use the key to search for the message. 
                  .setKeys("messageKey")
                  // The message tag. Consumers can use the tag to filter messages. 
                  // In this example, the tag of the message is set to TagA. 
                  .setTag("TagA")
                  // The message body. 
                  .setBody("messageBody".getBytes())
                  .build();
                      

• Subscribe to messages that match a single tag

  String topic = "Your Topic";
  // Subscribe to messages to which TagA is attached. 
  FilterExpression filterExpression = new FilterExpression("TagA", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• Subscribe to messages that match multiple tags

  String topic = "Your Topic";
  // Subscribe to messages to which TagA, TagB, or TagC is attached. 
  FilterExpression filterExpression = new FilterExpression("TagA||TagB||TagC", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

• Subscribe to all messages in the topic

  String topic = "Your Topic";
  // Subscribe to all messages. 
  FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
  pushConsumer.subscribe(topic, filterExpression);

Attribute-based SQL filtering

Attribute-based SQL filtering is an advanced filtering method provided by ApsaraMQ for RocketMQ. In this method, messages are matched based on the key-value pair that is specified by producers. When producers send messages, the producers can specify multiple attributes for each message. Then, the consumer can specify attributes in SQL expressions to receive specific messages.

Sample scenario

Message attribute setting

Before a producer sends messages, the producer can specify custom attributes for each message. Each attribute is a custom key-value pair.

Multiple attributes can be specified for each message.

Filtering rules

You must follow the SQL-92 syntax when you write filter expressions. The following table describes the syntax.

Sample code

• Specify a tag and custom attributes before message sending

  Message message = messageBuilder.setTopic("topic")
                  // The message key. You can use the key to search for the message. 
                  .setKeys("messageKey")
                  // The message tag. Consumers can use the tag to filter messages. 
                  // In this example, the message tag is set to messageTag. 
                  .setTag("messageTag")
                  // You can also specify custom attributes for the messages, such as environment, region, and logical branch. 
                  // In this example, the custom attribute is region and the attribute value is Hangzhou. 
                  .addProperty("Region", "Hangzhou")
                  // The message body. 
                  .setBody("messageBody".getBytes())
                  .build();

• Subscribe to messages that match a single attribute

  String topic = "topic";
  // Subscribe only to messages whose region attribute value is Hangzhou. 
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND Region='Hangzhou'", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• Subscribe to messages that match multiple attributes

  String topic = "topic";
  // Subscribe to messages whose region attribute value is Hangzhou and price attribute value is greater than 30. 
  FilterExpression filterExpression = new FilterExpression("Region IS NOT NULL AND price IS NOT NULL AND Region = 'Hangzhou' AND price > 30", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

• Subscribe to all messages in the topic

  String topic = "topic";
  // Subscribe to all messages. 
  FilterExpression filterExpression = new FilterExpression("True", FilterExpressionType.SQL92);
  simpleConsumer.subscribe(topic, filterExpression);

Usage notes

Properly plan topics and specify tags

FAQ

Why are messages lost when multiple consumers subscribe to different tags in a topic?

Possible cause: If the consumers belong to the same consumer group, the tags to which the consumers subscribe must be the same. Otherwise, subscription inconsistency occurs and specific messages are lost.

How do I calculate the number of messages that are consumed by a consumer when filtering conditions are specified?

The number of messages that a consumer consumes is calculated after the filtering conditions are specified.

Why are messages accumulated in a group even though online consumers did not consume messages?

If you use SQL filtering or tag-based filtering to filter messages, messages that do not meet the filtering conditions are accumulated. The following items describe how to calculate the number of accumulated messages in the preceding scenarios:

• SQL filtering: Number of accumulated messages = Number of ready messages + Number of inflight messages - Number of messages that do not meet the filtering conditions

• Tag-based filtering: Number of accumulated messages = (Number of ready messages + Number of inflight messages) × Percentage of messages that match the tags

Note

Percentage of messages that match the tags = Number of messages that match the tags in the sample/Total number of sampled messages

References

For information about the complete sample code for messaging, see Overview.