rocketmq 消息过滤

r囧r小猫 2022-12-08 14:14 261阅读 0赞

--------------------

### rocketmq 消息过滤 ###

**官网：[http://rocketmq.apache.org/docs/filter-by-sql92-example/][http_rocketmq.apache.org_docs_filter-by-sql92-example]**

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**tag 过滤：****简单过滤**

**同一个应用尽可能使用同一个topic，不同的消息子类型使用不同的tag标识**

**每个message只能使用一个tag**

Message message=new Message("topic","tagA","hello world".getBytes());  //消息标识tagA
    Message message=new Message("topic","tagB","hello world".getBytes());  //消息标识tagB
    Message message=new Message("topic","tagC","hello world".getBytes());  //消息标识tagC
    
    
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
    consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");   //订阅TAGA、TAGB、TAGC标识消息
    consumer.subscribe("TOPIC", "*");                      //订阅所有标识的消息

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**sql 过滤：****添加自定义属性，进行一些复杂过滤**

**支持的数据类型：数字、字符串（使用单引号，如：'abc'）、null、true | false**

**数字比较：>、>=、<、<=、between、=**

**字符串：=、<>、in**

**is null、is not null**

**and、or、not**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

msg.putUserProperty("a", "2");  //设置消息属性
    
    
    consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");  //sql过滤，a在0、3之间

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**示例**

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**配置文件**

**application.yml**

rocketmq:
      name-server: 192.168.57.2:9876
      producer:
        group: producerGroup

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**service 层**

**ProducerService：发送消息**

@Service
    public class ProducerService {
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        public void send(){
           SendResult result=rocketMQTemplate.syncSend("topic-test:tag",
                    MessageBuilder.withPayload("hello 瓜田李下 tag").build());  //设置标签：tag
           System.out.println(result);
        }
    
        public void send2(){
            SendResult result=rocketMQTemplate.syncSend("topic-test:tag2",
                    MessageBuilder.withPayload("hello 瓜田李下 tag2").build());  //设置标签：tag2
            System.out.println(result);
        }
    
        public void send3(){
            SendResult result=rocketMQTemplate.syncSend("topic-test",
                    MessageBuilder.withPayload("hello 瓜田李下 num=2")
                            .setHeader("num","2").build());  //设置userProperty：num = 2
            System.out.println(result);
        }
    
        public void send4(){
            SendResult result=rocketMQTemplate.syncSend("topic-test",
                    MessageBuilder.withPayload("hello 瓜田李下 num=6")
                            .setHeader("num","6").build());  //设置userProperty：num = 6
            System.out.println(result);
        }
    
        public void send5(){
            SendResult result=rocketMQTemplate.syncSend("topic-test",
                    MessageBuilder.withPayload("hello 瓜田李下 name=瓜田李下")
                            .setHeader("name","瓜田李下").build());  //设置userProperty：name=瓜田李下
            System.out.println(result);
        }
    
        public void send6(){
            SendResult result=rocketMQTemplate.syncSend("topic-test",
                    MessageBuilder.withPayload("hello 瓜田李下 name=海贼王")
                            .setHeader("name","海贼王").build());    //设置userProperty：name=海贼王
            System.out.println(result);
        }
    }

**ConsumerService：消费标签为tag的消息**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup",topic = "topic-test",selectorExpression = "tag")
    public class ConsumerService implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费标签为tag的消息："+s);
        }
    }

**ConusmerService：消费标签为tag2的消息**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup2",topic = "topic-test",selectorExpression ="tag2")
    public class ConsumerService2 implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费标签为tag2的消息："+s);
        }
    }

**ConusumerService3：sql过滤，num=1**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup3",topic = "topic-test",selectorType = SelectorType.SQL92,selectorExpression = "num=2",messageModel = MessageModel.BROADCASTING)
    public class ConsumerService3 implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费userProperty（num=2）的消息："+s);
        }
    }

**ConsumerService4：sql过滤，num>4**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup4",topic = "topic-test",selectorType = SelectorType.SQL92,selectorExpression = "num>4")
    public class ConsumerService4 implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费userProperty（num>4）的消息："+s);
        }
    }

**ConsumerService5：sql过滤，name='瓜田李下'**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup5",topic = "topic-test",selectorType = SelectorType.SQL92,selectorExpression = "name='瓜田李下'")
    public class ConsumerService6 implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费userProperty（name=瓜田李下）的消息："+s);
        }
    }

**ConsumerService6：sql过滤，name<>'瓜田李下'**

@Service
    @RocketMQMessageListener(consumerGroup = "consumerGroup6",topic = "topic-test",selectorType = SelectorType.SQL92,selectorExpression = "name<>'瓜田李下'",messageModel = MessageModel.BROADCASTING)
    public class ConsumerService6 implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String s) {
            System.out.println("消费userProperty（name<>瓜田李下）的消息："+s);
        }
    }

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**controller 层**

**HelloController**

@RestController
    public class HelloController {
    
        @Resource
        private ProducerService producerService;
    
        @RequestMapping("/hello")
        public void hello(){
            producerService.send();
            producerService.send2();
            producerService.send3();
            producerService.send4();
            producerService.send5();
            producerService.send6();
    
            System.out.println("发送完成");
        }
    }

### \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ###

**使用测试**

**localhost:8080/hello**

SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C889B00000, offsetMsgId=C0A8390300002A9F0000000000000000, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=1], queueOffset=0]
    SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C88A0E0002, offsetMsgId=C0A8390300002A9F000000000000014D, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=2], queueOffset=0]
    SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C88A110004, offsetMsgId=C0A8390300002A9F000000000000029C, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=3], queueOffset=0]
    SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C88A140006, offsetMsgId=C0A8390300002A9F00000000000003E8, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=0], queueOffset=0]
    SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C88A1A0008, offsetMsgId=C0A8390300002A9F0000000000000534, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=1], queueOffset=1]
    SendResult [sendStatus=SEND_OK, msgId=240E03A1BC903BE008318B12F802000303543FEE733D69C88A1E000A, offsetMsgId=C0A8390300002A9F0000000000000698, messageQueue=MessageQueue [topic=topic-test, brokerName=broker-a, queueId=2], queueOffset=1]
    发送完成
    消费userProperty（num>4）的消息：hello 瓜田李下 num=6
    消费标签为tag2的消息：hello 瓜田李下 tag2
    消费userProperty（name<>瓜田李下）的消息：hello 瓜田李下 name=海贼王
    消费userProperty（name=瓜田李下）的消息：hello 瓜田李下 name=瓜田李下
    消费标签为tag的消息：hello 瓜田李下 tag
    消费userProperty（num=2）的消息：hello 瓜田李下 num=2

[http_rocketmq.apache.org_docs_filter-by-sql92-example]: http://rocketmq.apache.org/docs/filter-by-sql92-example/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70]: /images/20221123/8f9df72b870b409ab503e22c51d1fbaa.png