CousumeQueue中tag的作用

2023-02-25 62

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 问题的提出存在就是有意义的，那么ConsumeQueue中存消息tag的hashcode是什么目的呢？

consumer.subscribe("TopicTest", "TagA");

那么在消息过滤是在broker还是Conumser呢？按照常理是在broker,因为在broker可以减少流量，实际情况是在broker过滤大部分，Consumer过滤一小部分

ConsumeQueue的结构

ConsumeQueue存的是主题的逻辑信息，如下图所示，代表一条记录。其中记录的信息存储在commitLog中，位置是CommitLog Offset。

Offset用于标记消息在CommitLog中的位置

Size用于标记消息的大小

HashCode用于过滤消息

源码跟踪

SubscriptionData的构建（Consumer启动）

Consumer一般会有订阅的主题和tag

consumer.subscribe("TopicTest", "TagA");

跟进去会跟到FilterAPI的buildSubscriptionData方法

public static SubscriptionData buildSubscriptionData(String topic, String subString) throws Exception {
        SubscriptionData subscriptionData = new SubscriptionData();
        subscriptionData.setTopic(topic);
        subscriptionData.setSubString(subString);
        if (null == subString || subString.equals(SubscriptionData.SUB_ALL) || subString.length() == 0) {
            subscriptionData.setSubString(SubscriptionData.SUB_ALL);
        } else {
            String[] tags = subString.split("\\|\\|");
            if (tags.length > 0) {
                for (String tag : tags) {
                    if (tag.length() > 0) {
                        String trimString = tag.trim();
                        if (trimString.length() > 0) {
                            //添加tag的set
                            //添加tag的set
                            //添加tag的set
                            subscriptionData.getTagsSet().add(trimString);
                            //添加tag的hashcode的set
                            //添加tag的hashcode的set
                            //添加tag的hashcode的set
                            subscriptionData.getCodeSet().add(trimString.hashCode());
                        }
                    }
                }
            } else {
                throw new Exception("subString split error");
            }
        }
        return subscriptionData;
    }

总结：SubscriptionData包含了tag列表和tag的hashcode列表

broker过滤消息

首先Consumer给broker发送消息，请求code是 RequestCode.PULL_MESSAGE ，因此我们可以跟borker里对这个请求码的处理的processor,最后定位到

PullMessageProcessor###processRequest方法，方法里有如下的代码

final GetMessageResult getMessageResult =
            this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(), requestHeader.getTopic(),
                requestHeader.getQueueId(), requestHeader.getQueueOffset(), requestHeader.getMaxMsgNums(), messageFilter);

跟DefaultMessageStore###getMessage方法

 public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
        final int maxMsgNums,
        final MessageFilter messageFilter) {
        //省略        
                        for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                            //获取消息的偏移量
                            long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
                             //获取消息的大小
                            int sizePy = bufferConsumeQueue.getByteBuffer().getInt();
                            //获取消息的tag的hashcode
                            long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();
                            maxPhyOffsetPulling = offsetPy;
                            if (nextPhyFileStartOffset != Long.MIN_VALUE) {
                                if (offsetPy < nextPhyFileStartOffset)
                                    continue;
                            }
                            //省略
                            //省略
                            //省略
                             //查看消息tag是否匹配,此时在broker实现过滤
                             //查看消息tag是否匹配,此时在broker实现过滤
                             //查看消息tag是否匹配,此时在broker实现过滤
                            if (messageFilter != null
                                && !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) {
                                if (getResult.getBufferTotalSize() == 0) {
                                    status = GetMessageStatus.NO_MATCHED_MESSAGE;
                                }
                                continue;
                            }
                          //省略
                          //省略
                          //省略
        return getResult;
    }

跟进匹配方法,此时能发现过滤方法是看subscriptionData里是否有包含tagsCode

//ExpressionMessageFilter###
```java
 @Override
    public boolean isMatchedByConsumeQueue(Long tagsCode, ConsumeQueueExt.CqExtUnit cqExtUnit) {
            //省略
            //省略
            //省略
            //订阅主题里是否包含这个hashcode
            return subscriptionData.getCodeSet().contains(tagsCode.intValue());
        } else {
           //省略
    }

总结：broker是根据subscriptionData里的tag的hashcode列表去过滤消息，判断从ConsumeQueue中读取的tag的hashcode是否在subscriptionData里的tag的hashcode列表中。

Consumer过滤消息

Consumer开始跟的地方在DefaultMQPushConsumerImpl###pullMessage方法里有一个PullCallback，此方法是一个给broker发送拉取消息后的一个回调方法

PullCallback pullCallback = new PullCallback() {
@Override
            public void onSuccess(PullResult pullResult) {
                if (pullResult != null) {
                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                        subscriptionData);
         //省略
}

跟一下PullAPIWrapper###processPullResult方法

public PullResult processPullResult(final MessageQueue mq, final PullResult pullResult,
        final SubscriptionData subscriptionData) {
              //省略
                for (MessageExt msg : msgList) {
                    if (msg.getTags() != null) {
                    //Consumer端过滤消息
                    //Consumer端过滤消息
                    //Consumer端过滤消息
                        if (subscriptionData.getTagsSet().contains(msg.getTags())) {
                            msgListFilterAgain.add(msg);
                        }
                    }
                }
            }
           //省略
        return pullResult;
    }