RocketMQ的TAG过滤和SQL过滤机制

时间：2022-12-16 22:52:25

写作目的

项目中各个中台都使用同一个DB。而DB下会使用中间件监听binlog转换成MQ消息，而下游的各个中台去MQ去拿自己感兴趣的消息。

  • TAG
    如果使用TAG去获取自己感兴趣的消息，那么对于一条学生表变更binlog，最少要插入三条消息，比如TAG=学生表，比如TAG=UPDATE修改操作，比如TAG=学生状态为1，等等。想到的就三种。。。
    所以上面这种方式缺陷还是挺明显的。

  • SQL过滤
    如果使用SQL过滤的方式，我们可以对某些属性进行过滤，自己拼接SQL，灵活性就上来了。

但是我好奇的一点是SQL怎么加到TAG里呢？并且TAG只能支持一个属性值呀。所以接下来从源码和原理的角度进行分析和探讨。

总体来说Tag过滤和SQL过滤如下图所示
RocketMQ的TAG过滤和SQL过滤机制

代码展示

本着简单的原则出发

TAG过滤

当producer构建消息时消息时会构造方法里会有TAG的属性，如代码所示，Tag = Creative。

Message msg =
              new Message(
                  "CBeann", // topic
                  "Creative", // tag
                  "OrderID188", // key
                  "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); // msg body

当consumer订阅topic时要想监听Tag = creative的就可以如下图所示

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
		//主题，Tag
        consumer.subscribe("CBeann", "Creative");
        consumer.setNamesrvAddr("114.115.208.175:9876");
        consumer.setConsumerGroup("group1");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    // wrong time format 2017_0422_221800
    // consumer.setConsumeTimestamp("20181109221800");
    consumer.registerMessageListener(...)

SQL过滤

与Tag消息不同的是，produccer生产的msg需要放入一些属性，如下代码所示，放入age属性的值为18。

 Message msg =
              new Message(
                  "CBeann", // topic
                  "creative", // tag
                  "OrderID188", // key
                  "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); // msg body
          msg.putUserProperty("age", String.valueOf(18));
          SendResult sendResult = producer.send(msg);

consumer中则不能根据tag过滤了。需要使用MessageSelector

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
        MessageSelector messageSelector = MessageSelector.bySql("age >= 5");
        consumer.subscribe("TopicTest", messageSelector);
        //consumer.setNamesrvAddr("114.115.208.175:9876");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumerGroup("group1Sql");
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    consumer.registerMessageListener()

TAG过滤机制

过滤图

此处以Tag过滤机制为例，消息过滤存在两个地方，一个是服务器端，另一个是消息者端。
RocketMQ的TAG过滤和SQL过滤机制

假设消费者订阅的topic=CBeann,tag=creative，creative的hashCode =9527(假设一下)

而topic=CBeann的消息队列里有3条消息
msg1[tag=feed,tagHashCode= 9000]
msg2[tag=creative,tagHashCode= 9527]
msg3[tag=material,tagHashCode= 9527]

当consumer消费者给broker服务器发送获取topic=CBeann,tag=creative请求时，请求会转化为topic=CBeann,tagHashCode=9527。
因此对于上述的3条消息，经过tagHashCode匹配后会把msg2和msg3发送给consumer消息者。
而Consumer消费者会根据tag匹配后留下msg2

源码思路讲解

构建SubscriptionData

首先要了解一点，我们在consumer中设置订阅的topic和tag是什么样的一个数据结构呢？

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
consumer.subscribe("CBeann", "Creative");

其实一直往下跟subcribe方法，最后我们可以定位到FilterAPI#buildSubscriptionData方法。我们传入的topic=CBeann；tag=creative 被封装到SubscriptionData对象里,包括topic、tag、tagHashCode。
RocketMQ的TAG过滤和SQL过滤机制

brokder过滤逻辑

那么consumer消费端是存储着topic、tag、tagHashCode。而consumer会把topic和tagHashCode发送给Broker服务器。

当consumer消费者向broker服务端请求获取消息时，broker会从ConsumeQueue获取offset之后的所有如下所示的三元组。ConsumeQueue里的数据三元组如下图所示。
RocketMQ的TAG过滤和SQL过滤机制

其实三元组是解析出来的，解析的三个属性就是上图中的offsetPy、sizePy和tagCode。下面我们重点关注一下tagCode
RocketMQ的TAG过滤和SQL过滤机制
解析出来的tagsCode如果匹配成功，则保留，如果匹配失败，则continue。
RocketMQ的TAG过滤和SQL过滤机制

接下来看一下是怎么匹配呢？如果是*，则全匹配，否则就根据tagsCode匹配。此处不是根据tag匹配，所以会有hash冲突的数据也会匹配到
RocketMQ的TAG过滤和SQL过滤机制

结论：此时我们可以看到，broker服务器端是通过hashcode匹配的，哈希冲突的msg会被认为有效消息发送给consumer端。

consumer过滤逻辑

一般这种RPC的都是通过回调实现的，所以看完源码后定位到了一个CallBack方法。该CallBack方法如下所示，拿到Broker发送的消息后在经过processPullResult预处理后才会真正去判断消息是否获取到。
RocketMQ的TAG过滤和SQL过滤机制
拿到消息后再经过Tag过滤，如下图所示，则到达我们自定义的处理消息逻辑
RocketMQ的TAG过滤和SQL过滤机制
结论：此时我们可以看到，consumer消费者端是通过tag匹配的，二次过滤因为哈希导致消息Tag不准确的问题。

SQL过滤机制

SQL过滤和Tag过滤的消息有什么区别

结论：没区别，就是多了几个属性。比如下面的代码中的age属性

Message msg =
              new Message(
                  "CBeann", // topic
                  "creative", // tag
                  "OrderID188", // key
                  "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); // msg body
          msg.putUserProperty("age", String.valueOf(18));

如上面代码所示，msg的tag=creative, 属性age=18。
其实根据Message的构造方法和putUserProperty方法可以发现，最后都是放到Properties里
RocketMQ的TAG过滤和SQL过滤机制

构建SubscriptionData

SQL过滤和Tag过滤的consumer端有什么区别?

如下面代码所示，我们构造了一个MessageSelector

 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
        MessageSelector messageSelector = MessageSelector.bySql("age >= 5");
        consumer.subscribe("TopicTest", messageSelector);

那么subscribe方法同样是把MessageSelector也是构建成SubscriptionData。不过和Tag那种不同的是，SubscriptionData里面放的是SQL即subString属性和expressionType属性SQL92
RocketMQ的TAG过滤和SQL过滤机制

源码跟踪

broker过滤逻辑

SQL过滤和tag过滤都是经历下面的三个阶段，下面我们重点跟一下SQL过滤的代码块messageFilter#isMatchedByCommitLog。RocketMQ的TAG过滤和SQL过滤机制

debug了一下，如下图所示，从buffer里解析出properties来然后和SQL进行校验，返回校验结果
RocketMQ的TAG过滤和SQL过滤机制

consumer过滤逻辑

一般这种RPC的都是通过回调实现的，所以看完源码后定位到了一个CallBack方法。该CallBack方法如下所示，拿到Broker发送的消息后在经过processPullResult预处理后才会真正去判断消息是否获取到。
下面的这个图其实在上面也出现过，这个处理方法里并没有SQL过滤的逻辑，因此在consumer不过滤。
RocketMQ的TAG过滤和SQL过滤机制

总结

  • 特殊的分表方式
    tag作为msg的properties,这个其实映射到数据库分库分表中。比如db的一条记录需要新增一个字段，我们完全可以新增一个setting表，存储这个properties属性。阿里这边的很多项目DB设计都是这么做的。
  • SQL过滤比Tag过滤慢的原因：比较慢，解析慢
    Tag过滤是直接等于，而SQL过滤还要通过表达式计算，SQL复杂的计算必然不如直接等于快。
    SQL过滤的时候需要解析properties，本身就是一种资源消耗。
标签：

相关文章