详述RocketMQ消息过滤方式
近期在做一个需求，业务应用需要接收指定 tag 下订单支付成功消息，担心该 tag 下在大促期间会有大量消息发出，导致消息队列堆积而消费不及时，间接影响业务。由于中间件使用的是 Rocket MQ，自然想到 Rocket MQ 的消息过滤机制，于是问题来了：
- 问题一：Rocket MQ 提供了几种消息过滤机制？
- 问题二：Rocket MQ 消息过滤是发生在服务端还是客户端，如果发生在客户端，那么是不是可以通过在服务端过滤消息来减轻业务应用（客户端）消费消息的压力？
带着这两个问题结合 Rocket MQ 的源码来一探究竟。
Rocket MQ 提供了几种消息过滤机制？
消息过滤包括基于表达式与基于类模式的两种过滤方式，其中表达式又分为 tag 和 sql92 模式，如下图：
1_mq_filter.png
1）表达式过滤 tag 方式
最常用的方式，示例代码如下：
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
// 只订阅的消息 TagA 或 TagC
consumer.subscribe("TagFilterTest", "TagA || TagC");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
2）表达式过滤 sql92 方式
基于 sql92 表达式消息过滤，其实是对消息的属性运用 sql 过滤表达式进行条件匹配，所以消息发送时应该调用 putUserProperty 方法设置消息属性。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_1");
// 只有订阅的消息有这个属性a, a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
3）MessageFilter 类过滤方式
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_2");
// 使用 MessageFilter实现类，在服务器做消息过滤
String filterCode = MixAll.file2String("MessageFilterImpl.java");
consumer.subscribe("FilterTopicTest", "com.zqh.demo.filter.MessageFilterImpl", filterCode);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
Rocket MQ 消息过滤是发生在服务端还是客户端？
直接上结论：不论 Rocket MQ 消息过滤是基于表达式还是基于类模式，Rocket MQ 消息过滤都是发生在服务端。先从 Rocket MQ 消息发送与接口的原理说起，如下图（为了方便理解，图中做了简化）：
2_main.png
-
消息生产者把消息发送并存储到 Rocket MQ 的 broker 上，NameServer 用来发现和更新 broker。
-
消费者启动时会启动 PullMessageService 线程，PullMessageService 线程不断地从内部的队列中取 PullRequest，
然后使用 PullRequest 作为请求去拉取消息。 -
PullRequest 中的消息处理队列 ProcessQueue 是 MessageQueue 在消费端的重现、快照。PullMessageService 使用消费者（DefaultMQPushConsumerImpl）从消息服务器默认每次拉取 32 条消息，按消息的队列偏移量存放在 ProcessQueue 中，
然后消费者再将消息提交到消息消费线程池中（提交 ConsumeRequest），消息成功消费后从 ProcessQueue 中移除。
所以需要结合两个时机来看：服务端处理消息拉取请求时机、客户端拉到消息后处理时机。再结合源码来看：
时机一：Rocket MQ 服务端（Borker）接收到拉取消息命令（RequestCode.PULL_MESSAGE），会对消息进行过滤
- 代码类：PULL_MESSAGE 命令处理类：org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest
- 核心代码片断：
3_server_filter.png
时机二：Rocket MQ 消费端（一般是业务应用）拉取到消息后对消息的处理
- 代码类：org.apache.rocketmq.client.impl.consumer.PullAPIWrapper#processPullResult
- 核心代码片断：
4_tag_client_filter.png
为什么基于表达式 tag 会在客户端再进行一次过滤？
由于在消息服务端进行消息过滤是匹配消息 tag 的 hashcode，导致服务端过滤并不十分准确，从服务端返回的消息最终不一定是消息消费者订阅的消息，导致服务端过滤并不十分准确，这样也会造成网络带宽的浪费（可使用基于 MessageFilter 实现类模式的消息过滤）
总结
Rocket MQ 消息过滤包括基于表达式与基于类模式的两种过滤方式，其中表达式又分为 tag 和 sql92 模式（sql92 模式可以会对用户属性字段进行复杂的过滤），且都是在服务端对消息进行过滤。
