java操作rocketmq实现消息过滤的核心方式是tag和sql表达式。1. tag过滤适用于简单分类，通过设置tag并使用||订阅多个tag提高效率；2. sql表达式过滤支持and、or、not及比较运算符，需在broker中开启enablepropertyfilter并设置用户属性；3. 选择时根据需求复杂度决定，tag适合简单场景，sql适合复杂条件；4. 性能优化包括简化表达式、控制tag数量、启用缓存、优化属性及监控性能；5. 排查sql失效需检查broker配置、语法、属性设置及日志；6. 还可自定义messagefilter实现灵活过滤。合理选择与优化过滤方式有助于提升消费效率并降低负载。
Java操作RocketMQ，核心在于利用Tag和SQL表达式实现消息过滤，提高消费效率。
解决方案
RocketMQ提供了两种主要的消息过滤方式：基于Tag的过滤和基于SQL表达式的过滤。选择哪种取决于你的具体需求和消息属性的复杂程度。
立即学习“Java免费学习笔记（深入）”；
-
基于Tag的过滤
Tag过滤是最简单的一种方式。发送消息时，为每条消息设置一个Tag。消费者在订阅时，可以指定要消费的Tag。
- 发送消息:
DefaultMQProducer producer = new DefaultMQProducer("group_name"); producer.setNamesrvAddr("your_namesrv_address"); producer.start(); Message msg = new Message("TopicTest", "TagA", "OrderID001", "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); producer.shutdown();- 消费消息:
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4"); consumer.setNamesrvAddr("your_namesrv_address"); consumer.subscribe("TopicTest", "TagA || TagB || TagC"); // 订阅TagA、TagB或TagC的消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(Listmsgs, ConsumeConcurrentlyContext context) { System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); 注意点:
- Tag过滤效率高，适用于简单的消息分类。
- Tag的数量不宜过多，避免影响性能。
- 消费者可以使用
||运算符订阅多个Tag。
-
基于SQL表达式的过滤
SQL表达式过滤允许你使用更复杂的条件来过滤消息。你需要先开启Broker的SQL过滤功能，然后在发送消息时设置用户属性，消费者使用SQL表达式进行过滤。
-
开启Broker SQL过滤 (重要)
在
broker.conf文件中添加enablePropertyFilter=true，重启Broker。 如果不开启，SQL过滤会失效。
蚂蚁分类信息系统多城市分站版6.1S下载MayiCMS·蚂蚁分类信息系统是一款基于PHP+MYSQL（PC+手机+小程序+APP，跨平台、跨终端）的建站软件，拥有专业且完善的信息审核机制，信息刷新/置顶消费机制，信息分享/发布奖励机制，信息查看/付费授权机制，会员等级自助续费机制，为在各种类型操作系统服务器上架设信息发布平台提供完美的解决方案，拥有世界一流的用户体验，卓越的访问速度和负载能力。功能特点：1，PC手机自适应，URL路径完全
发送消息:
DefaultMQProducer producer = new DefaultMQProducer("group_name"); producer.setNamesrvAddr("your_namesrv_address"); producer.start(); Message msg = new Message("TopicTest", "TagA", "OrderID001", "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); msg.putUserProperty("age", String.valueOf(18)); // 设置用户属性 SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); producer.shutdown();- 消费消息:
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4"); consumer.setNamesrvAddr("your_namesrv_address"); // 使用MessageSelector指定SQL表达式 consumer.subscribe("TopicTest", MessageSelector.bySql("age > 10 AND age < 20")); // 订阅age大于10且小于20的消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(Listmsgs, ConsumeConcurrentlyContext context) { System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); 注意点:
- SQL表达式过滤功能需要Broker支持。
- SQL表达式的语法有限制，只能使用
AND,OR,NOT,=,>,,>=,,IN,BETWEEN等运算符。 - 支持的数据类型包括
NUMERIC,BOOLEAN,STRING。 - SQL过滤的性能比Tag过滤略低，但灵活性更高。
-
如何选择合适的过滤方式？
如果只需要简单的消息分类，Tag过滤更简单高效。如果需要基于消息属性进行更复杂的过滤，SQL表达式过滤更适合。 实际应用中，可以结合使用这两种方式，例如先使用Tag过滤缩小范围，再使用SQL表达式过滤精确匹配。
RocketMQ消息过滤的性能优化策略有哪些？
-
减少过滤表达式的复杂度: 复杂的SQL表达式会增加Broker的过滤负担，尽量简化表达式，避免使用过多的
AND和OR运算符。 - 合理设置Tag数量: Tag数量过多会导致Broker的索引变大，影响性能。根据实际情况，合理划分Tag。
- 开启Broker的SQL过滤缓存: RocketMQ Broker可以缓存SQL过滤结果，减少重复计算。可以通过配置参数开启缓存。
- 优化消息属性: 消息属性的数据类型和大小会影响过滤性能。尽量使用简单的数据类型，避免使用过大的字符串。
- 监控Broker性能: 通过监控Broker的CPU、内存和磁盘IO等指标，及时发现性能瓶颈。
如果SQL表达式过滤不起作用，应该如何排查？
-
确认Broker是否开启SQL过滤功能: 检查
broker.conf文件中是否配置了enablePropertyFilter=true，并重启了Broker。 - 检查SQL表达式语法是否正确: RocketMQ的SQL表达式语法有一定限制，确保表达式符合规范。可以参考RocketMQ官方文档。
- 检查消息属性是否设置正确: 确认消息中是否设置了SQL表达式中使用的属性，并且属性名称和数据类型是否正确。
- 检查消费者订阅的Topic和Tag是否正确: 确保消费者订阅的Topic和Tag与生产者发送的消息一致。
- 查看Broker日志: 查看Broker日志，查找是否有SQL过滤相关的错误信息。
-
使用简单的SQL表达式进行测试: 先使用简单的SQL表达式进行测试，例如
age > 10，如果可以正常工作，再逐步增加表达式的复杂度。
除了Tag和SQL表达式，还有没有其他的消息过滤方式？
虽然Tag和SQL表达式是最常用的过滤方式，但RocketMQ也支持自定义消息过滤。你可以通过实现MessageFilter接口，编写自己的过滤逻辑。
- 自定义MessageFilter:
public class MyMessageFilter implements MessageFilter {
@Override
public boolean match(MessageExt msg, FilterContext context) {
String propertyValue = msg.getUserProperty("your_property");
// 自定义过滤逻辑
return propertyValue != null && propertyValue.equals("your_value");
}
}- 消费者使用自定义MessageFilter:
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.setNamesrvAddr("your_namesrv_address");
// 使用自定义MessageFilter
consumer.subscribe("TopicTest", "*", new MyMessageFilter());
// ... 剩余代码自定义消息过滤提供了更高的灵活性，但也需要更多的开发工作。通常情况下，Tag和SQL表达式过滤已经可以满足大部分需求。
在实际应用中，选择合适的消息过滤方式，并进行适当的性能优化，可以有效地提高RocketMQ的消费效率，降低系统负载。
