Message Filtering Based on Topics in RocketMQ

In RocketMQ's high-performance distributed messaging system, message filtering is a key technology for improving consumption efficiency, reducing storage costs, and implementing fine-grained business routing. When producers send messages to a specific Topic, if multiple consumer groups exist under that topic and different groups handle distinct business logic, not all subscribers need to receive the same message. In such scenarios, the topic-based message filtering mechanism plays a central role by allowing consumers to pre-filter message bodies before pulling them from the broker, delivering only those matching specific conditions to corresponding consumption instances.

RocketMQ's message filtering primarily relies on the "Tag" (标签) mechanism for implementation. Producers can specify one or more tags when sending messages, while consumers define which messages to receive by specifying tag expressions during Topic subscription. This design not only avoids unnecessary transmission of invalid messages over networks and disks but also significantly reduces the load on downstream application systems.
Tags are string collections used in RocketMQ to identify message attributes. They are typically dynamically added by producers or predefined at the Topic level. On the producer side, tags can be set using DefaultTags or custom Tag policies. For example, in an e-commerce system, order messages can be tagged with labels such as "CREATED", "PAID_SUCCESSFULLY", or "SHIPPED" to ensure downstream services (such as inventory deduction, point issuance, and logistics notifications) precisely retrieve the required information.

On the consumer side, subscription relationships are established via Subscription objects, which contain both the Topic name and a Tag expression. Tag expressions support various comparison operators:

• Exact Match: Uses the = operator, e.g., ORDER_STATUS=CREATED.
• Fuzzy Match: Uses the LIKE operator, supporting wildcards %, e.g., USER_TYPE% matches all tags starting with USER_TYPE.
• Regular Expression Match: Uses the REGEX operator, e.g., STATUS=(SUCCESS|FAIL).

At least one subscriber within a consumer group must contain a valid Tag expression; otherwise, that consumer will be unable to pull any messages from the Broker. If a consumption instance has no Tag conditions configured, it defaults to receiving all messages under that Topic, which is typically used for scenarios requiring full data synchronization.

Application Strategies and Best Practices for Filters

In practical architecture design, properly utilizing message filters can significantly optimize resource utilization. First, for non-core business or low-frequency update data, broad filtering conditions (such as LIKE) can be set to push only partial data to specific services, avoiding delays and bandwidth waste caused by full synchronization. Second, in microservice governance scenarios, different services often have varying requirements for messages on the same topic; utilizing Tag filters enables message decoupling and independent scaling.

It is important to note that RocketMQ's filtering logic executes after consumers pull messages but before delivering them to specific consumption instances. This means the Broker does not actually store duplicate copies of filtered-out messages, thereby saving storage space. However, this also requires producers to accurately configure Tags; any misconfiguration can lead to data loss or incomplete processing. Additionally, there is a limit on Tag length, and it is generally recommended to keep them within reasonable bounds to balance readability with performance.

Code Implementation Examples and Verification Methods

To demonstrate how topic-based message filtering works practically, the following simplified Java code example shows how producers add tags and consumers subscribe to specific tags using expressions.

// Producer side: Add Tag to order messages
public class OrderProducer {
    public void sendOrderMessage(String topic, String orderData) {
        DefaultTags tags = new DefaultTags();
        // Dynamically set Tags based on business logic, e.g., by order type or status
        tags.set("ORDER_TYPE", "SALE"); 
        tags.set("STATUS", "PENDING");
        
        MQMessage msg = new MQMessage(topic, orderData.getBytes());
        msg.setTag(tags.toString()); // RocketMQ internally converts DefaultTags to a string
        
        // Send message
        producer.send(msg);
    }
}

// Consumer side: Subscribe to messages containing specific Tags
public class OrderConsumer {
    public void subscribeAndConsume(String topic) {
        Subscription sub = new Subscription();
        sub.setTopic(topic);
        // Use LIKE expression to match all order types starting with SALE
        sub.setExpression("ORDER_TYPE=SALE"); 
        
        Consumer consumer = new DefaultMQPushConsumer("my_group");
        consumer.subscribe(sub);
        
        // Start consumption
        consumer.start();
    }
}

In actual deployment environments, the effectiveness of filters can be verified using RocketMQ Dashboard or command-line tools. Administrators can query Broker logs to observe whether messages marked with ORDER_TYPE=SALE are successfully delivered to the corresponding Consumer Group and whether unmatching conditions are correctly intercepted. Additionally, it is recommended to conduct low-volume tests before production deployment to ensure Tag parsing logic aligns with expectations, preventing data anomalies caused by regex misconfiguration.

In summary, topic-based message filtering is an indispensable component of RocketMQ architecture. Through its flexible tagging mechanism, it achieves efficient distribution and management of message streams, laying a solid foundation for building high-availability, scalable enterprise-grade messaging systems. Developers should deeply understand its working principles and plan Tag strategies during the system design phase to ensure consistency and accuracy in message processing.