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过滤 直接返回true
    if (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 针对消息属性过滤，此时必须读取到完整的消息才能过滤，效率较低。