RocketMQ——消息过滤机制
Tag过滤
在Producer中使⽤Tag：
Message msg = new Message("TopicTest","TagA" ,("Hello RocketMQ " ).getBytes(RemotingHelper.DEFAULT_CHARSET));
在Consumer中订阅Tag：
consumer.subscribe("TopicTest", "TagA || TagB");// * 代表订阅Topic下的所有消息
SQL过滤
默认情况不⽀持，若想使⽤需要在conf⽂件夹下修改f，添加配置 enablePropertyFilter=true，然后启动broker是使该配置⽂件⽣效mqbroker -n 192.168.150.113:9876 -fnodeselector
通过给message设置消息属性（key-value）实现过滤
（还记得我曾讲过，⼀条消息由消息头，消息属性，消息体构成）
//producer中
for(int i =0; i <100; i++){
Message msg =new Message("myTopic001",("RocketMQ 第"+ i +"条消息").getBytes());
//设置的消息属性
msg.putUserProperty("id", String.valueOf(i));
producer.send(msg);
}
///consumer中
MessageSelector ms = MessageSelector.bysql("id >= 0 and id <= 10");
consumer.subscribe("myTopic001", ms);
RocketMQ只定义了⼀些基本的语法来⽀持这个功能。
数字⽐较, >, >=, <, <=, BETWEEN, =
字符⽐较, =, <>, IN
IS NULL 或者 IS NOT NULL
逻辑运算AND, OR, NOT