IDEA2023.1.3破解,IDEA破解,IDEA 2023.1破解,最新IDEA激活码

文章永久连接：https://tech.souyunku.com/?p=6024

作者：唯有坚持不懈 | 出处：https://blog.csdn.net/prestigeding/article/details/78888290

继上篇源码分析了 Tag 过滤机制实现原理，本文主要阐述 RocketMQ SQL92 表达式与 ClassFilte r过滤机制实现。

1、RocketMQ SQL92实现原理分析

入口：PullMessageProcessor#processRequest

if (!ExpressionType.isTagType(subscriptionData.getExpressionType())) {
consumerFilterData = ConsumerFilterManager.build(
                        requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getSubscription(),
                        requestHeader.getExpressionType(), requestHeader.getSubVersion()
     );
    assert consumerFilterData != null;

首先构建ConsumeFilterData数据结构：

consumeGroup: 消费组
topic ：消息主题
expresstion:消息过滤表达式，例如SQL92表达式，或过滤类全路径名
expresseionType : 表达式类型，可取值TAG、SQL92
compiledExpression:编译后的表达式对象
bornTime: ConsumerFilterData 对象创建创建时间
deadTime: ConsumerFilterData 对象死亡时间，默认0，表示一直有效。
BloomFilterData bloomFilterData
clientVersion:客户端版本

接下来我们先重点分析 ConsumerFilterManager#build方法：

/**
     * Build consumer filter data.Be care, bloom filter data is not included.
     *
     * @return maybe null
     */
    public static ConsumerFilterData build(final String topic, final String consumerGroup,
        final String expression, final String type,
        final long clientVersion) {   // @1 
        if (ExpressionType.isTagType(type)) {  // @2
            return null;
        }

        ConsumerFilterData consumerFilterData = new ConsumerFilterData();
        consumerFilterData.setTopic(topic);
        consumerFilterData.setConsumerGroup(consumerGroup);
        consumerFilterData.setBornTime(System.currentTimeMillis());
        consumerFilterData.setDeadTime(0);
        consumerFilterData.setExpression(expression);
        consumerFilterData.setExpressionType(type);
        consumerFilterData.setClientVersion(clientVersion);
        try {
            consumerFilterData.setCompiledExpression(
                FilterFactory.INSTANCE.get(type).compile(expression)
            );  // @3
        } catch (Throwable e) {
            log.error("parse error: expr={}, topic={}, group={}, error={}", expression, topic, consumerGroup, e.getMessage());
            return null;
        }

        return consumerFilterData;

该方法构建 ConsumerFilterData 数据。不包含 BoomFilter 的构建。
代码@1，首先我们来看一下参数：

topic : 主题。
consumerGroup ：消费组。
expression ：表达式。
type ：消息过滤类型,TAG或SQL92。
clientVersion : 客户端版本号。

代码@2：如果是 TAG 模式，则返回 null,表明 ConsumerFilterData 不针对TAG模式。
代码@3：编译表达式，这里我们就不再深入去看如何解析SQL语句，例如 a >= 1 and b <=10 这些语法，如果大家有兴趣，可以重点看一下 SqlFilter–>SelectorParser。这里会返回一个BooleanExpression的实现类，用于判断是否匹配。

我们再转入到 MessageFilter#isMatchedByCommitLog 的实现中来，因为 SQL92 是根据消息体中的属性进行过滤，故需要访问CommitLog 文件，也就是消息体。

public boolean isMatchedByCommitLog(ByteBuffer msgBuffer, Map<String, String> properties) {
        if (subscriptionData == null) {
            return true;
        }

        if (subscriptionData.isClassFilterMode()) {
            return true;
        }

        if (ExpressionType.isTagType(subscriptionData.getExpressionType())) {
            return true;
        }

        ConsumerFilterData realFilterData = this.consumerFilterData;
        Map<String, String> tempProperties = properties;

        // no expression
        if (realFilterData == null || realFilterData.getExpression() == null
            || realFilterData.getCompiledExpression() == null) {
            return true;
        }

        if (tempProperties == null && msgBuffer != null) {  // @1
            tempProperties = MessageDecoder.decodeProperties(msgBuffer);
        }

        Object ret = null;
        try {
            MessageEvaluationContext context = new MessageEvaluationContext(tempProperties);

            ret = realFilterData.getCompiledExpression().evaluate(context);  // @2
        } catch (Throwable e) {
            log.error("Message Filter error, " + realFilterData + ", " + tempProperties, e);
        }

        log.debug("Pull eval result: {}, {}, {}", ret, realFilterData, tempProperties);

        if (ret == null || !(ret instanceof Boolean)) {
            return false;
        }

        return (Boolean) ret;

代码@1：从消息体中解码出属性。
代码@2：然后对表达式进行匹配，上下文环境为消息体中的属性，如果匹配，则返回true,否则返回false。

2、ClassFilter 消息过滤机制 FilterServer详解

从图中可以看出，如果使用了类模式过滤，Consumer 不是直接从Broker拉取，而是从FilterServer上拉取。那么问题来了，FilterServer 是什么、Consume 如何与 FilterServer 打交道。

我们知道，一个客户端，一个专门的消息拉取线程(PullMessageService)专门负责拉取消息，多种过滤模式公用一套消息拉取机制【消息队列负载机制】，那 ClassFilter 模式是如何工作呢？首先，ClassFilter 模式，顾名思义就是消费端可以上传一个Class类文件到 FilterServer, 然后 FilterServer 从 Broker 拉取消息，执行过滤逻辑然后再返回给Consumer。

ClassFilter模式过滤机制，本文从如下三个方面展开。
1）ClassFilter注册（消费端如何提交自己的消息过滤实现类、以及消费订阅信息注册）。
2）消费端如何路由到FilterServer上拉取消息。
3）FilterServer消息拉取与消息消费。

2.1 ClassFilter模式消息过滤类注册机制

DefaultMQPushConsumerImpl#subscribe

public void subscribe(String topic, String fullClassName, String filterClassSource) throws MQClientException {  // @1
        try {
            SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
                topic, "*");
            subscriptionData.setSubString(fullClassName);
            subscriptionData.setClassFilterMode(true);     // @2
            subscriptionData.setFilterClassSource(filterClassSource);
            this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);    // @3
            if (this.mQClientFactory != null) {
                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();     // @4
            }

        } catch (Exception e) {
            throw new MQClientException("subscription exception", e);
        }

代码@1：topic : 主题，fullClassName ：过滤类全类路径名，filterClassSource：过滤类内容。
代码@2：设置 classFilterMode 为 true,表示类过滤机制。
代码@3：将该主题的订阅信息放入到 RebalanceImp l对象中，一个消费者各自维护一个 RebalanceImpl 对象，用于创建消息拉取任务。
代码@4：sendHeartbeatToAllBrokerWithLock，关键，发送心跳到所有Broker。
MQClientInstance#sendHeartbeatToAllBrokerWithLock

重点关注MQClientInstance#uploadFilterClassSource方法：

继续进入MQClientInstance#uploadFilterClassToAllFilterServer方法。

private void uploadFilterClassToAllFilterServer(final String consumerGroup, final String fullClassName,
        final String topic,
        final String filterClassSource) throws UnsupportedEncodingException {
        byte[] classBody = null;
        int classCRC = 0;
        try {
            classBody = filterClassSource.getBytes(MixAll.DEFAULT_CHARSET);
            classCRC = UtilAll.crc32(classBody);
        } catch (Exception e1) {
            log.warn("uploadFilterClassToAllFilterServer Exception, ClassName: {} {}",
                fullClassName,
                RemotingHelper.exceptionSimpleDesc(e1));
        }  // @1

        TopicRouteData topicRouteData = this.topicRouteTable.get(topic);    
        if (topicRouteData != null
            && topicRouteData.getFilterServerTable() != null && !topicRouteData.getFilterServerTable().isEmpty()) {    // @2
            Iterator<Entry<String, List<String>>> it = topicRouteData.getFilterServerTable().entrySet().iterator();     
            while (it.hasNext()) {
                Entry<String, List<String>> next = it.next();
                List<String> value = next.getValue();
                for (final String fsAddr : value) {
                    try {
                        this.mQClientAPIImpl.registerMessageFilterClass(fsAddr, consumerGroup, topic, fullClassName, classCRC, classBody,
                            5000);    // @3

                        log.info("register message class filter to {} OK, ConsumerGroup: {} Topic: {} ClassName: {}", fsAddr, consumerGroup,
                            topic, fullClassName);

                    } catch (Exception e) {
                        log.error("uploadFilterClassToAllFilterServer Exception", e);
                    }
                }
            }
        } else {
            log.warn("register message class filter failed, because no filter server, ConsumerGroup: {} Topic: {} ClassName: {}",
                consumerGroup, topic, fullClassName);
        }

代码@1：将代码转换成字节数值。
代码@2：根据主题找到路由信息，如果路由信息中的filterServerTable不为空，则通过网络将classname,class内容注册到FilterServer中。这里不免有一个疑问：TopicRouteInfo中的 filterserver 地址从何而来？我们先简单了解一下代码@3，再来分析这个问题，也就是FilterServer注册机制。
代码@3：registerMessageFilterClass，向路由信息中包含的 FilterServer 服务器注册过滤类，该方法主要是构建RequestCode.REGISTER_MESSAGE_FILTER_CLASS 消息，发往FilterServer。具体处理逻辑，在FilterServer端。

2.1.1 FilterClassManager 源码分析

FilterServer收到REGISTER_MESSAGE_FILTER_CLASS，完成类的注册与类加载。

2、1.1.1 FilterClassManager#registerFilterClass

public boolean registerFilterClass(final String consumerGroup, final String topic,
        final String className, final int classCRC, final byte[] filterSourceBinary) {   // @1
        final String key = buildKey(consumerGroup, topic);                                        // @2

        boolean registerNew = false;
        FilterClassInfo filterClassInfoPrev = this.filterClassTable.get(key);
        if (null == filterClassInfoPrev) {                                                                       // @3
            registerNew = true;
        } else {
            if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {    // @4
                if (filterClassInfoPrev.getClassCRC() != classCRC && classCRC != 0) {
                    registerNew = true;
                }
            }
        }

        if (registerNew) {  // @5
            synchronized (this.compileLock) {
                filterClassInfoPrev = this.filterClassTable.get(key);
                if (null != filterClassInfoPrev && filterClassInfoPrev.getClassCRC() == classCRC) {  //@6
                    return true;
                }

                try {

                    FilterClassInfo filterClassInfoNew = new FilterClassInfo();
                    filterClassInfoNew.setClassName(className);
                    filterClassInfoNew.setClassCRC(0);
                    filterClassInfoNew.setMessageFilter(null);

                    if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {   // @7
                        String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
                        Class<?> newClass = DynaCode.compileAndLoadClass(className, javaSource);
                        Object newInstance = newClass.newInstance();
                        filterClassInfoNew.setMessageFilter((MessageFilter) newInstance);
                        filterClassInfoNew.setClassCRC(classCRC);
                    }

                    this.filterClassTable.put(key, filterClassInfoNew);
                } catch (Throwable e) {
                    String info =
                        String
                            .format(
                                "FilterServer, registerFilterClass Exception, consumerGroup: %s topic: %s className: %s",
                                consumerGroup, topic, className);
                    log.error(info, e);
                    return false;
                }
            }
        }

        return true;
}

代码@1：consumerGroup 消费组名称；topic：消费主题；className：过滤类; classCRC：过滤类crc,filterSourceBinary 过滤类内容字节数组。

代码@2：构建 FilterClass 信息的缓存key,主题名 + “@” + 消费组名。

代码@3：如果当前不存在该key的过滤器信息，则认为是第一次注册。

代码@4：如果允许客户端编译上传的类，并且原先的过滤信息的crc与新的额crc不一样，也认为是第一次注册，将覆盖原先的注册信息。

代码@5：加锁，防止并发修改注册信息Map。

代码@6：这里是双重检查（并发编程通用的手段），例如，同一个消费组多个消费者同时注册，进行排队，一个处理好了之后，其他的获取锁，再检查一次，避免重复操作。

代码@7：如果允许客户端编译上传的类(clientUploadFilterClassEnable=true)，则根据过滤类名，过滤类源代码，利用jdk提供的编译API（JavaCompiler），具体封装类(DynaCode),将类编译好，如果不允许编译的话（clientUploadFilterClassEnable=false），就只是收集这些信息，真正的类加载需要去服务器去下载，然后再编译。这里主要是基于安全考虑，因为允许消费者（应用程序）直接上传JAVA类，本身就是一件危险的事情。那如果clientUploadFilterClassEnable=false,那如何编译呢？

2、1.1.2 FilterClassManager#fetchClassFromRemoteHost

public void start() {
        if (!this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    fetchClassFromRemoteHost();
                }
            }, 1, 1, TimeUnit.MINUTES);         // @1
        }
    }

    private void fetchClassFromRemoteHost() {
        Iterator<Entry<String, FilterClassInfo>> it = this.filterClassTable.entrySet().iterator();
        while (it.hasNext()) {
            try {
                Entry<String, FilterClassInfo> next = it.next();
                FilterClassInfo filterClassInfo = next.getValue();
                String[] topicAndGroup = next.getKey().split("@");
                String responseStr =
                    this.filterClassFetchMethod.fetch(topicAndGroup[0], topicAndGroup[1],
                        filterClassInfo.getClassName());    // @2
                byte[] filterSourceBinary = responseStr.getBytes("UTF-8");
                int classCRC = UtilAll.crc32(responseStr.getBytes("UTF-8"));
                if (classCRC != filterClassInfo.getClassCRC()) {
                    String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
                    Class<?> newClass =
                        DynaCode.compileAndLoadClass(filterClassInfo.getClassName(), javaSource);
                    Object newInstance = newClass.newInstance();
                    filterClassInfo.setMessageFilter((MessageFilter) newInstance);
                    filterClassInfo.setClassCRC(classCRC);

                    log.info("fetch Remote class File OK, {} {}", next.getKey(),
                        filterClassInfo.getClassName());
                }
            } catch (Exception e) {
                log.error("fetchClassFromRemoteHost Exception", e);
            }
        }

代码@1：首先，如果 isClientUploadFilterClassEnable 设置为 false, 则开启一个定时任务，每一分钟尝试从远程服务器，isClientUploadFilterClassEnable=false 时，需要配置 filterClassRepertoryUr l属性，根据类名拉取类的源码。

代码@2：通过HTTP协议从远程服务器拉取代码，远程服务器URL: filterClassRepertoryUrl/classname.java。

2.2 ClassFilter模式 FilterServer注册机制

上文提到，消费者采用类过滤机制时，需要向 FilterServe r服务器列表注册过滤器类，FilterServer 的列表是从 topic的路由信息TopicInfo 中获取, topicInfo 中的 FilterServer 从何而来呢？本节将揭开谜底.

RocketMQ路由信息维护，Broker 通过心跳向 NameServer 注册。再结合此图，应该能想到 FilterServer 向 Broker 注册，然后Broker再发送给 NameServer。

2.2.1 FilterServer注册

FiltersrvController#initialize

FilterServerOuterAPI#registerFilterServerToBroker，该方法就是向 Broker 发送 RequestCode.REGISTER_FILTER_SERVER 请求。
Broker端：AdminBrokerProcessor#registerFilterServer。

这里的关键点就是调用 brokerControl.getFilterServerManager().registerFillterServer 方法，将该注册信息存入到 brokerControl 的 FilerServerManager 路由表中（this.filterServerTable.put(channel, filterServerInfo)），完成 FilterServer 到 Broker 的注册过程。

我们知道，消费者调用 subscribe 方法，会首先将订阅信息存储在消费者中，其实存储在消费端 RebalanceImpl 的订阅消息中。然后通过心跳来注册消息到 Broker, 然后 Broker 通过心跳发送到 NameServer, 将 FilterServer 服务器列表加入到路由信息中。

由于该部分源码简单，就不一一介绍了，NameServer 接收 Broker 心跳，并注册 FilterServer 服务器的代码入口：
DefaultRequestProcessor#registerBrokerWithFilterServer，接下来进入到消息拉取环节。

2.3 消息拉取

进入本篇最后一个议题，消息拉取。

消息拉取大家应该不陌生了吧，一个 MQClientInstance 一个拉取消息线程。

消息拉去以推模式来讲解，拉模式是应用程序手动去拉。并不是通过 PullMessageServer 线程。

PullMessageServer 线程在执行消息拉取时，最终将进入到：PullAPIWrapper#pullKernelImpl。

核心代码：如果使用了类过滤消息模式，将改变拉取地址为FilterServer地址：

从该Broker中的FilterServer地址列表中，随机选择一个，进行拉取。

在FilterServer端，就是通过拉模式根据拉取需求从Broker上拉取消息，然后执行消息过滤逻辑。

4、总结

本文重点分析了FilterServer类消息过滤的实现原理，总结如下：
1、订阅消息与FilterServer注册流程

消息消费者先用类过滤模式将过滤器类，过滤器类内容添加到订阅消息中（RebalanceImpl)。
消费者拉取消息时，向FilterServer发送请求，那FilterServer怎么得到FilterServer服务器地址呢？通过主题路由信息，向NameServer获取。消费者在得到FilterServervf服务地址的时候，会将自己过滤器类代码发送到FilterServer,方便FilterServer在拉取消息时执行过滤逻辑。
FilterServer启动时向Broker注册Filter服务地址信息。（服务列表），然后Broker将该信息通过心跳向NameServer注册，将FilterServer服务列表保存在主题的路由信息中。

2、clientUploadFilterClassEnable 参数，如果设置为false, 则需要一个url，去下载过滤类：url/classname.java 这样的请求，FilterServer要能拿到过滤类代码。

3、类过滤模式，上传的类必须实现MessageFilter接口。

备注：本文是《RocketMQ技术内幕》的前期素材，建议关注笔者的书籍：《RocketMQ技术内幕》。

干货推荐

十三、RocketMQ源码分析消息过滤机制下篇-FilterServer、ClassFilter模式详解

1、RocketMQ SQL92实现原理分析

2、ClassFilter 消息过滤机制 FilterServer详解

2.1 ClassFilter模式消息过滤类注册机制

2.1.1 FilterClassManager 源码分析

2.2 ClassFilter模式 FilterServer注册机制

2.2.1 FilterServer注册

2.3 消息拉取

4、总结

干货推荐

附录：RocketMQ源码分析，系列文章

相关推荐

评论抢沙发

学习必备：精选优质专栏

大前端WP主题更专业更方便

觉得文章有用就打赏一下文章作者

微信扫一扫打赏

微信扫一扫打赏

QQ咨询

回顶部

1、RocketMQ SQL92实现原理分析

2、ClassFilter 消息过滤机制 FilterServer详解

2.1 ClassFilter模式 消息过滤类注册机制

2.1.1 FilterClassManager 源码分析

2.2 ClassFilter模式 FilterServer注册机制

2.2.1 FilterServer注册

2.3 消息拉取

4、总结

干货推荐

附录：RocketMQ源码分析，系列文章

相关推荐

评论 抢沙发

学习必备：精选优质专栏

大前端WP主题 更专业 更方便

觉得文章有用就打赏一下文章作者

微信扫一扫打赏

微信扫一扫打赏

QQ咨询

回顶部

2.1 ClassFilter模式消息过滤类注册机制

评论抢沙发

大前端WP主题更专业更方便