RocketMQ5.0消息过滤-编程知识

前言

消费者订阅了某个主题后，RocketMQ 会将该主题中的所有消息投递给消费者。若消费者只需要关注部分消息，可通过设置过滤条件在 Broker 端进行过滤，只获取到需要关注的消息子集，避免接收到大量无效的消息。

以电商交易场景为例，用户从下单到拿到商品，中间会产生很多消息，被不同的下游系统订阅消费。下游系统往往只关心自己需要处理的消息，比如支付系统只关心支付消息，这时候生产者就可以在发送消息的时候给消息打上标签，下游系统按需订阅即可。

过滤方式

RocketMQ 支持两种消息过滤方式。

Tag标签过滤

生产者在发送消息前，可以先给消息打上标签，每条消息最多设置一个 Tag 标签：

Message message = provider.newMessageBuilder().setTopic("Trade_Topic").setTag("pay").setBody("xxx".getBytes()).build();
producer.send(message);

消费者配置 Tag 标签过滤规则：

consumer.subscribe("Trade_Topic", new FilterExpression("pay", FilterExpressionType.TAG));

Tag 标签过滤规则：

单 Tag 匹配：过滤表达式为目标 Tag，相同 Tag 的消息才会投递给消费者
多 Tag 匹配：过滤表达式为多个目标 Tag 用||分割，消息符合任一 Tag 就会被投递
全部匹配：过滤表达式为*，所有消息都会投递

SQL属性过滤

SQL 属性过滤是 RocketMQ 提供的高级消息过滤方式，每个消息都可以额外设置用户属性和系统属性，消费者订阅时可设置 SQL 语法的过滤表达式过滤多个属性。

SQL 过滤也可以实现 Tag 标签过滤的效果，Tag 属于系统属性，属性名称是 TAGS

首先，生产者发送消息前给消息设置自定义属性：

Message message = provider.newMessageBuilder().setTopic("Trade_Topic").setBody("xxx".getBytes()).addProperty("price", "99800").addProperty("region", "杭州").build();
producer.send(message);

消费者配置 SQL 过滤规则，这里以杭州区域价格大于 100 的订单为例：

consumer.subscribe("Trade_Topic", new FilterExpression("region='杭州' AND price>10000", FilterExpressionType.SQL92));

SQL 属性过滤使用 SQL92 语法作为过滤规则表达式，语法规范如下：

如何选择

尽量用 Tag 标签过滤，实现更加轻量级，效率更高，在扫描 ConsumeQueue 时就可以先通过 TagHash 过滤一遍。而消息属性是存储在 CommitLog 文件里的，意味着 SQL 属性过滤必须读到完整的消息才能判断是否要过滤，性能较差。

设计实现

org.apache.rocketmq.store.MessageFilter是 RocketMQ 抽象出来的消息过滤接口，两个方法：

isMatchedByConsumeQueue：通过 ConsumeQueue 里的 tagsCode 先匹配一次，也就是 Tag 标签的哈希码，tagsCode 不同 Tag 肯定不同
isMatchedByCommitLog：根据 CommitLog 里的完整消息属性匹配

public interface MessageFilter {boolean isMatchedByConsumeQueue(final Long tagsCode,final ConsumeQueueExt.CqExtUnit cqExtUnit);boolean isMatchedByCommitLog(final ByteBuffer msgBuffer,final Map<String, String> properties);
}

RocketMQ 的处理逻辑是：先根据 ConsumeQueue 里的 tagsCode 过滤，通过了再读取 CommitLog 里的完整消息走 SQL 属性过滤，实现类会根据配置的过滤规则在不关心的过滤方法里直接返回 true。

public GetMessageResult getMessage(){......// 先通过consumequeue里的tagsCode过滤if (messageFilter != null&& !messageFilter.isMatchedByConsumeQueue(cqUnit.getValidTagsCodeAsLong(), cqUnit.getCqExtUnit())) {if (getResult.getBufferTotalSize() == 0) {status = GetMessageStatus.NO_MATCHED_MESSAGE;}continue;}// 再从CommitLog读取完整消息SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);// 再执行SQL属性过滤if (messageFilter != null&& !messageFilter.isMatchedByCommitLog(selectResult.getByteBuffer().slice(), null)) {if (getResult.getBufferTotalSize() == 0) {status = GetMessageStatus.NO_MATCHED_MESSAGE;}selectResult.release();continue;}......
}

Tag 标签过滤的实现
Broker 把消息写入 CommitLog 后，ReputMessageService 线程会每隔 1ms 把新消息写入到 consumequeue 文件，以加速消费者的消费效率。ConsumeQueue 文件由若干个 CqUnit 组成，每个 CqUnit 占用固定的 20 个字节：

CqUnit{long offset; // 消息在 CommitLog 偏移量int size; // 消息长度long tagsCode; // Tag哈希码
}

消费者在消费 ConsumeQueue 时就可以直接通过 tagsCode 进行标签过滤：

public boolean isMatchedByConsumeQueue(Long tagsCode, ConsumeQueueExt.CqExtUnit cqExtUnit) {// by tags code.if (ExpressionType.isTagType(subscriptionData.getExpressionType())) {if (tagsCode == null) {return true;}// '*' 订阅所有if (subscriptionData.getSubString().equals(SubscriptionData.SUB_ALL)) {return true;}// 消息的tagsCode是否包含在消费者订阅的Tags里面return subscriptionData.getCodeSet().contains(tagsCode.intValue());}
}

因为是哈希码，所以 tagsCode 存在哈希冲突的可能性，不过概率极小。万一冲突了，Broker 还是会继续投递消息，RocketMQ 5.0 版本会由 Proxy 再进行一次 Tag 的精准匹配，如果不匹配不会投递给消费者；RocketMQ 4.x 版本由消费者收到消息后自行判断，Tag 不匹配的消息会直接丢弃。

SQL 属性过滤的实现
为了执行 SQL 语法实现属性过滤，SQL 语法会先被编译成 Expression 对象，再由Expression#evaluate方法得出执行结果。

Expression expression = FilterFactory.INSTANCE.get(ExpressionType.SQL92).compile("a>10 AND b<10 OR c=10");
expression.evaluate(context);

要对消息属性过滤，首先要把消息属性提取出来，消息属性由若干个 String 类型的键值对组成，然后执行 SQL。

public boolean isMatchedByCommitLog(ByteBuffer msgBuffer, Map<String, String> properties) {// tag过滤 直接返回trueif (ExpressionType.isTagType(subscriptionData.getExpressionType())) {return true;}ConsumerFilterData realFilterData = this.consumerFilterData;// 消息属性Map<String, String> tempProperties = properties;// 没有SQL表达式if (realFilterData == null || realFilterData.getExpression() == null|| realFilterData.getCompiledExpression() == null) {return true;}if (tempProperties == null && msgBuffer != null) {// 从CommitLog解码出消息属性tempProperties = MessageDecoder.decodeProperties(msgBuffer);}Object ret = null;try {MessageEvaluationContext context = new MessageEvaluationContext(tempProperties);// 执行SQL92表达式过滤ret = realFilterData.getCompiledExpression().evaluate(context);} catch (Throwable e) {log.error("Message Filter error, " + realFilterData + ", " + tempProperties, e);}if (ret == null || !(ret instanceof Boolean)) {return false;}return (Boolean) ret;
}

尾巴

消息过滤是 RocketMQ 防止 Broker 端因为投递大量消费者不感兴趣的消息而导致资源浪费的一种手段，消费者可以根据自己感兴趣的消息类型配置过滤规则，分为 Tag 标签过滤和 SQL 属性过滤两种方式。Tag 标签过滤效率高，因为 Broker 在构建 consumequeue 文件时会写入消息 Tag 的哈希码，直接比较哈希码可以避免通过 CommitLog 读取完整消息。SQL 针对消息属性过滤，此时必须读取到完整的消息才能过滤，效率较低。