Tag filtering and SQL filtering of rocketmq

Catalog

tag Filter

Application scenarios

principle

Sample code

Producer code

Consumer code

sql Filter  

Introduce

To configure

The sample code

producer

consumer

---

tag Filter

Application scenarios

stay RockerMQ Can be used in the topic Divide the business , For example, order 、 goods 、 Activities and other businesses are divided into different topic. To make the business logic clearer, you can also use tag Divide again , For example, the order is divided into clothing orders 、 Home appliance orders 、 Beverage order, etc .

principle

On the server side, we use tag Of hashcode Filter , When consumers tag Subscription related queue In the news tag Of hashcode When it is consistent, it will return directly , But this can only filter most tag, Because of existence hash Collision , Therefore, the client should also be based on tag Value to filter

Server use hashcode Filtering is because it reduces unnecessary network traffic , also hashcode Fast filtration , The bottom layer can directly use bit operations

Sample code

Producer code

public static void main(String[] args) {
    try {
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        //  Created a label for tag_a The news of 
        Message msg = new Message("topic_a", "tag_a", ("test").getBytes());
        producer.send(msg);
        producer.shutdown();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

After sending, it will be sent to console You can see that there are defined values in the column of the message tag

Consumer code

public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
    consumer.setNamesrvAddr("localhost:9876");
    // consumer.subscribe("topic_a", "*");
    consumer.subscribe("topic_a", "tag_a");
    consumer.setMessageModel(MessageModel.CLUSTERING);
    consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
        for (MessageExt msg : msgList) {
            System.out.println(new String(msg.getBody()));;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
}

by * Do not filter any messages , You need to subscribe to multiple tag use || separate , for example tag_a||tag_b 

sql Filter  

Introduce

In some cases , More complex filtering conditions may be required , It can be used at this time sql Filter ,sql Filtration performance ratio tag low , Only a few basic grammars are defined , as follows ：

• Numerical comparison , such as ：>,>=,<,<=,BETWEEN,=;
• Character comparison , such as ：=,<>,IN;
• IS NULL perhaps  IS NOT NULL;
• Logical symbols  AND,OR,NOT;

The constant support type is ：

• The number , such as ：123,3.1415;
• character , such as ：'abc', It must be enclosed in single quotes ;
• NULL, Special constants ;
• Boolean value ,TRUE  or  FALSE

To configure

Need to be in conf/broker.conf Add the following configuration , Otherwise, it will report a mistake

enablePropertyFilter=true

Start with the specified broker.conf file

./mqbroker -n localhost:9876 -c ../conf/broker.conf 

The sample code

producer

public static void main(String[] args) {
    try {
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        Message msg = new Message("topic_a", ("test").getBytes());
        msg.putUserProperty("age", "11");
        msg.putUserProperty("name", " Zhang San ");
        producer.send(msg);
        producer.shutdown();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

consumer

public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
    consumer.setNamesrvAddr("localhost:9876");
    //  subscribe age Greater than 10 also name It's Zhang San's message 
    consumer.subscribe("topic_a", MessageSelector.bySql("age > 10 and name = ' Zhang San '"));
    consumer.setMessageModel(MessageModel.CLUSTERING);
    consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
        for (MessageExt msg : msgList) {
            System.out.println(new String(msg.getBody()));;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
}