这里是weihubeats,觉得文章不错可以关注公众号小奏技术，文章首发。拒绝营销号，拒绝标题党
RocketMQ 版本
- 5.1.0
RokcetMQ消息过滤
目前官方支持的消息过滤方式主要有两种
- tag
- sql92
我们可以通过查看ExpressionType的源码证明
tag过滤方式是现在最为常用的过滤方式，但是一个消息只能包含一个tag。
对于相对复杂的消息过滤场景tag过滤方式可能就不够用了，但是绝大多数业务场景tag过滤方式已经够用了。
sql92过滤方式可以有助于我们实现一些高级功能，比如RocketMQ的多测试环境消息隔离等。
这里就暂时不过多讨论sql92的具体使用场景，我们还是先来学习怎么使用sql92
sql92 语法规则
| 语法 | 说明 | 示例 |
|---|---|---|
| IS NULL | 判断属性不存在。 | a IS NULL ：属性a不存在。 |
| IS NOT NULL | 判断属性存在。 | a IS NOT NULL：属性a存在。 |
| > >= < <= | 用于比较数字，不能用于比较字符串，否则消费者客户端启动时会报错。 说明 可转化为数字的字符串也被认为是数字。 | a IS NOT NULL AND a > 100：属性a存在且属性a的值大于100。 a IS NOT NULL AND a > ‘abc’：错误示例，abc为字符串，不能用于比较大小。 |
| BETWEEN xxx AND xxx | 用于比较数字，不能用于比较字符串，否则消费者客户端启动时会报错。等价于>= xxx AND <= xxx。表示属性值在两个数字之间。 | a IS NOT NULL AND (a BETWEEN 10 AND 100)：属性a存在且属性a的值大于等于10且小于等于100。 |
| NOT BETWEEN xxx AND xxx | 用于比较数字，不能用于比较字符串，否则消费者客户端启动会报错。等价于< xxx OR > xxx，表示属性值在两个值的区间之外。 | a IS NOT NULL AND (a NOT BETWEEN 10 AND 100)：属性a存在且属性a的值小于10或大于100。 |
| IN (xxx, xxx) | 表示属性的值在某个集合内。集合的元素只能是字符串。 | a IS NOT NULL AND (a IN (‘abc’, ‘def’))：属性a存在且属性a的值为abc或def。 |
| = <> | 等于和不等于。可用于比较数字和字符串。 | a IS NOT NULL AND (a = ‘abc’ OR a<>‘def’)：属性a存在且属性a的值为abc或a的值不为def。 |
| AND OR | 逻辑与、逻辑或。可用于组合任意简单的逻辑判断，需要将每个逻辑判断内容放入括号内。 | a IS NOT NULL AND (a > 100) OR (b IS NULL)：属性a存在且属性a的值大于100或属性b不存在。 |
由于SQL属性过滤是生产者定义消息属性，消费者设置SQL过滤条件，因此过滤条件的计算结果具有不确定性，服务端的处理方式如下：
-
异常情况处理：如果过滤条件的表达式计算抛异常，消息默认被过滤，不会被投递给消费者。例如比较数字和非数字类型的值。
-
空值情况处理：如果过滤条件的表达式计算值为null或不是布尔类型（true和false），则消息默认被过滤，不会被投递给消费者。例如发送消息时未定义某个属性，在订阅时过滤条件中直接使用该属性，则过滤条件的表达式计算结果为null。
-
数值类型不符处理：如果消息自定义属性为浮点型，但过滤条件中使用整数进行判断，则消息默认被过滤，不会被投递给消费者。
sql92使用
源码
所有源码已上传至github
- 地址:https://github.com/weihubeats/weihubeats_demos/tree/master/java-demos/rocketmq-demo/src/main/java/com/weihubeats/rocketmq/demo/sql92
消息发送
public class SQLProducer {
public static int count = 10;
public static String topic = "xiao-zou-topic";
public static void main(String[] args) {
DefaultMQProducer producer = MQUtils.createLocalProducer();
IntStream.range(0, count).forEach(i -> {
Message message = new Message(topic, ("sql92 test" + i).getBytes(StandardCharsets.UTF_8));
try {
if (i % 2 == 0) {
message.putUserProperty("gray", "dev1");
}
SendResult sendResult = producer.send(message);
DateTimeFormatter dtf2 = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
System.out.printf("%s %s%n", sendResult, dtf2.format(LocalDateTime.now()));
}
catch (Exception e) {
throw new RuntimeException(e);
}
});
producer.shutdown();
}
}
这里我们假装消息是发送个多个测试的消息，所以每条消息都在UserProperty添加了一个dev1标签。
我们要实现的就是比如只有dev1环境的消费者才会消费带有dev1标签的消息，其他消息则丢弃掉
消息消费
public class SQLConsumer {
public static String GID = "xiao-zou-gid";
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = MQUtils.createLocalConsumer(GID);
String sql = "gray is not null and gray = 'dev1'";
consumer.subscribe(MQUtils.TOPIC, MessageSelector.bySql(sql));
consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
/*
* Launch the consumer instance.
*/
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
这里的消息消费方式唯一不同的是我们订阅消息的方式发生了变化
普通方法我们调用的是这个方法进行消息订阅的，传入tag就行
比如像这样
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
但是这里我们使用的是sql92方式
传入的是一个MessageSelector，订阅的规则是
String sql = "gray is not null and gray = 'dev1'";
运行效果
- 消息发送
这里我们发送了十条消息，只有5条是带有gray标签的
- 消息消费
可以看到消息消费只有消费了带有gray标签的5条消息，符合我们的预期
sql92是在客户端还是在服务端过滤的？
sql92和tag都是在服务端过滤的，我们可以查看源码得知
不过tag的过滤方式会在客户端再次过滤。因为在服务端是通过hashcode进行过滤的，为了提高性能，没有对原始的tag进行过滤，在通过hashcode过滤掉绝大多少的消息后，在客户端进行最后的tag完全过滤。
org.apache.rocketmq.client.impl.consumer.PullAPIWrapper#processPullResult
如果统一都在客户端过滤会导致传输大量的消息到客户端，影响性能
总结
本次我们对RocketMQ sql92过滤消息进行了简单的使用以及少量的源码分析，并没有完整的从整个流程进行分析，因为本篇并不是源码分析偏。sql92在实际的项目中的相对来说较少，偶尔如果做RocketMQ消息的多册环境或者灰度，可能是一个方案，但不是最佳的
参考
- 官方文档
