Message filtering

This topic describes how consumers filter messages on a RocketMQ broker according to tags. For more information about topics and tags, see Topic and tag best practices.

A tag is a label that classifies messages into different types under a topic. RocketMQ allows consumers to filter messages according to tags, ensuring that the consumers consume messages of types they are concerned with.

The following figure shows an example in the e-commerce transaction scenario. The process from placing an order to receiving the product by the customer will produce a series of messages, such as order message, payment message, and logistics message. These messages will be sent to the queue with the topic Trade_Topic and received by different systems, such as the payment system, logistics system, transaction success rate analysis system, and real-time computing system. Among these systems, the logistics system only receives the logistics message, and the real-time computing system receives all the messages related to the transaction (order, payment, and logistics).

Note: To classify messages, you can create multiple topics, or create multiple tags under the same topic. However, in general, there is no relationship between messages in different topics, and tags are used to distinguish related messages within the same topic, such as the relationship between the full set and the subset, and the relationship between the processes in sequence.

Sample code

Sending messages

A tag must be specified for each message before it is sent:

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

Consumption method - 1

If a consumer needs to subscribe to all types of messages under a topic, the * symbol can be used to represent the tags:

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

Consumption method - 2

If a consumer needs to subscribe a certain type of messages under a topic, the tag should be specified:

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

Consumption method - 3

If a consumer needs to subscribe to messages of multiple types under a topic, separate tags with separators (||):

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

Consumption method - 4 (error example)

If a consumer subscribes to messages of the tags under a topic for multiple times, the tags subscribed to the previous time prevail:

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