Filtering RocketMQ Messages in Spring Boot: Tag vs SQL‑92 Techniques

Dependencies and Configuration

Use Spring Boot 2.3.9.RELEASE with RocketMQ 4.8.0. Add the following Maven dependencies:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <version>2.2.0</version>
</dependency>

Application.yml snippet:

server:
  port: 8082
rocketmq:
  nameServer: localhost:9876
  producer:
    group: demo-mq

Method 1: Tag‑Based Filtering

Each message can carry a single tag; the consumer can subscribe to multiple tags using the || operator.

@RocketMQMessageListener(topic = "filter-topic",
    consumerGroup = "consumer06-group",
    selectorExpression = "tag11 || tag12 || tag13",
    messageModel = MessageModel.CLUSTERING,
    selectorType = SelectorType.TAG)
@Component
public class ConsumerFilterMessageListener implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        System.out.println(Thread.currentThread().getName());
        System.out.println(new String(message.getBody()));
        System.out.println(message.getProperties());
    }
}

Producer sends a message with a tag:

public void sendFilterMessage(String topic, String message, String tags) {
    Message<String> msg = MessageBuilder.withPayload(message).build();
    rocketMQTemplate.convertAndSend(topic + ":" + tags, msg);
}

Test endpoint:

@GetMapping("/filter")
public Object sendFilterMessage(String content) {
    ps.sendFilterMessage("filter-topic", content, "tag12");
    return "send filter message success";
}

Method 2: SQL‑92 Expression Filtering

RocketMQ supports a subset of SQL‑92 syntax to filter messages based on header properties.

Numeric comparison: >, >=, <, <=, BETWEEN, = String comparison: =, <>, IN Null checks: IS NULL, IS NOT NULL Logical operators: AND, OR, NOT Constants: numbers, quoted strings, NULL, TRUE/FALSE

Consumer example using an SQL selector:

@RocketMQMessageListener(topic = "filter-topic",
    consumerGroup = "consumer06-group",
    selectorExpression = "pack = 'abc' || a = 1",
    messageModel = MessageModel.CLUSTERING,
    selectorType = SelectorType.SQL92)
@Component
public class ConsumerFilterMessageListener implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        System.out.println(Thread.currentThread().getName());
        System.out.println(new String(message.getBody()));
        System.out.println(message.getProperties());
    }
}

Producer sets header properties and sends the message:

public void sendFilterMessage(String topic, String message, String tags) {
    Message<String> msg = MessageBuilder.withPayload(message).build();
    Map<String, Object> headers = new HashMap<>();
    headers.put("pack", "abc");
    headers.put("a", 10);
    rocketMQTemplate.convertAndSend(topic, msg, headers);
}

Test endpoint (no tag needed):

@GetMapping("/filter")
public Object sendFilterMessage(String content) {
    ps.sendFilterMessage("filter-topic", content, null);
    return "send filter message success";
}

Changing the selector expression from OR to AND makes the consumer stop receiving messages, demonstrating the effect of the logical operator.

These two approaches—tag‑based and SQL‑92 expression filtering—cover the common ways to filter RocketMQ messages in a Spring Boot application.