ApsaraMQ for RocketMQ:Message filtering

Send a message with a tag

Specify a tag for each message before sending:

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

Subscribe to all messages

Use an asterisk (*) to subscribe to all messages in a topic:

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

Subscribe to messages with a specific tag

Specify the tag to receive only matching messages:

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 with multiple 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: multiple subscriptions to the same topic

If a consumer calls consumer.subscribe() multiple times for the same topic with different tags, only the last subscription takes effect. Earlier subscriptions are overridden.

// The consumer receives only messages with TagB. Messages with TagA are NOT received.
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;
        }
    });

Attribute key rules

Each custom attribute is a key-value pair. The key must follow these rules:

• Can contain letters, digits, and underscores (_).

• Must start with a letter or an underscore (_).

Multiple attributes can be set on each message.

Producer: set custom attributes

Message msg = new Message("topic", "tagA", "Hello MQ".getBytes());
// Set custom attribute A to 1.
msg.putUserProperties("A", "1");

Consumer: define a filter expression

Use MessageSelector.bySql() to define an SQL filter expression.

// Subscribe to messages where attribute A exists and equals 1.
consumer.subscribe("topic", MessageSelector.bySql("A IS NOT NULL AND TAGS IS NOT NULL AND A = '1'"), new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        System.out.printf("Receive New Messages: %s %n", message);
        return Action.CommitMessage;
    }
});

Send a message with a tag and custom attributes

Producer producer = ONSFactory.createProducer(properties);
// Set the tag to tagA.
Message msg = new Message("topicA", "tagA", "Hello MQ".getBytes());
// Set custom attribute region to hangzhou.
msg.putUserProperties("region", "hangzhou");
// Set custom attribute price to 50.
msg.putUserProperties("price", "50");
SendResult sendResult = producer.send(msg);

Filter by a custom attribute

Consumer consumer = ONSFactory.createConsumer(properties);
// Subscribe only to messages where region is hangzhou.
// Messages without the region attribute or with a different value are not delivered.
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 result: The message is delivered. It has the region attribute set to hangzhou, which matches the filter condition.

Filter by tag and custom attribute

Consumer consumer = ONSFactory.createConsumer(properties);
// Subscribe only to messages with tagA and price 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 result: The message is delivered. It has tagA and the price attribute is 50, which is greater than 30.

Filter by multiple custom attributes

Consumer consumer = ONSFactory.createConsumer(properties);
// Subscribe only to messages where region is hangzhou and 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. The consumer requires price less than 20, but the message has price set to 50.

Subscribe to all messages with SQL

Consumer consumer = ONSFactory.createConsumer(properties);
// Set the SQL expression to TRUE to receive all messages.
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 result: All messages in the topic are delivered.

Incorrect: missing attribute and null check

If a producer does not set a custom attribute on a message, and the consumer's filter expression references that attribute without an IS NOT NULL check, the expression evaluates to NULL. The message is not delivered.

Consumer consumer = ONSFactory.createConsumer(properties);
// The product attribute was not set when the message was sent.
// The filter fails and the message is not delivered.
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;
    }
});