rocketmq源碼解析之管理請求獲取消費者狀態

時間 2019-11-11

標籤 rocketmq 源碼解析管理請求獲取消費者狀態简体版

原文原文鏈接

說在前面java

管理請求 GET_CONSUME_STATS 獲取消費者狀態apache

源碼解析緩存

進入到這個方法org.apache.rocketmq.broker.processor.AdminBrokerProcessor#getConsumeStats獲取消費者狀態安全

private RemotingCommand getConsumeStats(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final GetConsumeStatsRequestHeader requestHeader =
            (GetConsumeStatsRequestHeader) request.decodeCommandCustomHeader(GetConsumeStatsRequestHeader.class);
        ConsumeStats consumeStats = new ConsumeStats();
        Set<String> topics = new HashSet<String>();
        if (UtilAll.isBlank(requestHeader.getTopic())) {
//            獲取消費者所在組的topics=》
            topics = this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(requestHeader.getConsumerGroup());
        } else {
            topics.add(requestHeader.getTopic());
        }

        for (String topic : topics) {
//            按topic名稱從本地緩存中獲取topic配置信息
            TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
            if (null == topicConfig) {
                log.warn("consumeStats, topic config not exist, {}", topic);
                continue;
            }

            {
//                按消費組和topic獲取訂閱數據=》
                SubscriptionData findSubscriptionData =
                    this.brokerController.getConsumerManager().findSubscriptionData(requestHeader.getConsumerGroup(), topic);
                if (null == findSubscriptionData
                    && this.brokerController.getConsumerManager().findSubscriptionDataCount(requestHeader.getConsumerGroup()) > 0) {
                    log.warn("consumeStats, the consumer group[{}], topic[{}] not exist", requestHeader.getConsumerGroup(), topic);
                    continue;
                }
            }

//            默認讀隊列數量16
            for (int i = 0; i < topicConfig.getReadQueueNums(); i++) {
                MessageQueue mq = new MessageQueue();
                mq.setTopic(topic);
                mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());
                mq.setQueueId(i);
                OffsetWrapper offsetWrapper = new OffsetWrapper();
//                =》按topic和queueId查詢broker的offset
                long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
                if (brokerOffset < 0)
                    brokerOffset = 0;
//                按消費組、topic、queueId查詢消費者的offset=》
                long consumerOffset = this.brokerController.getConsumerOffsetManager().queryOffset(
                    requestHeader.getConsumerGroup(),
                    topic,
                    i);
                if (consumerOffset < 0)
                    consumerOffset = 0;
                offsetWrapper.setBrokerOffset(brokerOffset);
                offsetWrapper.setConsumerOffset(consumerOffset);
                long timeOffset = consumerOffset - 1;
                if (timeOffset >= 0) {
//                    查詢最後時間 =》
                    long lastTimestamp = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, timeOffset);
                    if (lastTimestamp > 0) {
                        offsetWrapper.setLastTimestamp(lastTimestamp);
                    }
                }

                consumeStats.getOffsetTable().put(mq, offsetWrapper);
            }

//            按消費組和topic獲取消費者的tps
            double consumeTps = this.brokerController.getBrokerStatsManager().tpsGroupGetNums(requestHeader.getConsumerGroup(), topic);
            consumeTps += consumeStats.getConsumeTps();
            consumeStats.setConsumeTps(consumeTps);
        }

        byte[] body = consumeStats.encode();
        response.setBody(body);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

進入到這個方法org.apache.rocketmq.broker.offset.ConsumerOffsetManager#whichTopicByConsumer找到消費組訂閱的topic微信

public Set<String> whichTopicByConsumer(final String group) {
        Set<String> topics = new HashSet<String>();
//        遍歷消費者組、topic緩存信息
        Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, ConcurrentMap<Integer, Long>> next = it.next();
            String topicAtGroup = next.getKey();
            String[] arrays = topicAtGroup.split(TOPIC_GROUP_SEPARATOR);
            if (arrays.length == 2) {
                if (group.equals(arrays[1])) {
                    topics.add(arrays[0]);
                }
            }
        }

        return topics;
    }

進入這個方法org.apache.rocketmq.broker.topic.TopicConfigManager#selectTopicConfig從topic配置信息中獲取當前topic的配置併發

public TopicConfig selectTopicConfig(final String topic) {
//        從topic配置緩存信息中查詢當前topic的配置
        return this.topicConfigTable.get(topic);
    }

進入這個方法org.apache.rocketmq.broker.client.ConsumerManager#findSubscriptionData按消費組、topic查詢消息組訂閱數據信息app

public SubscriptionData findSubscriptionData(final String group, final String topic) {
//        按組從緩存中獲取消費組信息
        ConsumerGroupInfo consumerGroupInfo = this.getConsumerGroupInfo(group);
        if (consumerGroupInfo != null) {
//            按topic從緩存中獲取訂閱數據
            return consumerGroupInfo.findSubscriptionData(topic);
        }

        return null;
    }

進入這個方法org.apache.rocketmq.store.DefaultMessageStore#getMaxOffsetInQueue按topic、queueId查詢隊列中最大的offsetide

public long getMaxOffsetInQueue(String topic, int queueId) {
//        根據topic和queueId找到消費者隊列=》
        ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
        if (logic != null) {
//            獲取最大的offset =》
            long offset = logic.getMaxOffsetInQueue();
            return offset;
        }

//        若是不存在指定topic和queueId的消費隊列直接返回0
        return 0;
    }

進入這個方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueue 按topic、queueId查詢消息隊列性能

public ConsumeQueue findConsumeQueue(String topic, int queueId) {
//        找到topic的全部消息隊列
        ConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
        if (null == map) {
            ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
            ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
            if (oldMap != null) {
                map = oldMap;
            } else {
                map = newMap;
            }
        }

//        按queue id查找消費者隊列
        ConsumeQueue logic = map.get(queueId);
        if (null == logic) {
            ConsumeQueue newLogic = new ConsumeQueue(
                topic,
                queueId,
//                消費者隊列存儲地址 user.home/store/consumequeue
                StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
//                每一個文件存儲默認30W
                this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
                this);
            ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
            if (oldLogic != null) {
                logic = oldLogic;
            } else {
                logic = newLogic;
            }
        }

        return logic;
    }

往上返回到這個方法org.apache.rocketmq.store.ConsumeQueue#getMaxOffsetInQueue 查詢消息隊列中最大的offsetthis

public long getMaxOffsetInQueue() {
//        =》
        return this.mappedFileQueue.getMaxOffset() / CQ_STORE_UNIT_SIZE;
    }

進入到這個方法org.apache.rocketmq.store.MappedFileQueue#getMaxOffset

public long getMaxOffset() {
//        獲取存儲映射文件隊列中索引位置最大的映射文件=》
        MappedFile mappedFile = getLastMappedFile();
        if (mappedFile != null) {
//            映射文件的起始offset+映射文件的可讀取的索引位置
            return mappedFile.getFileFromOffset() + mappedFile.getReadPosition();
        }
//        若是隊列中沒有存儲映射文件直接返回0
        return 0;
    }

進入這個方法org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile()獲取映射文件隊列中最後的映射文件

public MappedFile getLastMappedFile() {
    MappedFile mappedFileLast = null;
    while (!this.mappedFiles.isEmpty()) {
        try {
            mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
            break;
        } catch (IndexOutOfBoundsException e) {
            //continue;
        } catch (Exception e) {
            log.error("getLastMappedFile has exception.", e);
            break;
        }
    }

    return mappedFileLast;
}

這裏的映射文件隊列是CopyOnWriteArrayList實現，讀多寫少的場景併發性能比較好

//    併發線程安全隊列存儲映射文件
    private final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();

往上返回進入這個方法org.apache.rocketmq.broker.offset.ConsumerOffsetManager#queryOffset(java.lang.String, java.lang.String, int) 按消費組、topic、queueId查詢消費者的offset

public long queryOffset(final String group, final String topic, final int queueId) {
    // topic@group 從本地offset緩存中查詢
    String key = topic + TOPIC_GROUP_SEPARATOR + group;
    ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
    if (null != map) {
        Long offset = map.get(queueId);
        if (offset != null)
            return offset;
    }

    return -1;
}

進入這個方法org.apache.rocketmq.store.DefaultMessageStore#getMessageStoreTimeStamp 按topic、queueId、consumeQueueOffset查詢消費者offset的最後更新時間

@Override
    public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
//        按topic和queueId查詢到消費隊列=》
        ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
        if (logicQueue != null) {
//            按消費者的offset查詢存儲時間所在的buffer=》
            SelectMappedBufferResult result = logicQueue.getIndexBuffer(consumeQueueOffset);
//            =》
            return getStoreTime(result);
        }

        return -1;
    }

進入這個方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueue 按topic、queueId查詢消息隊列

public ConsumeQueue findConsumeQueue(String topic, int queueId) {
//        找到topic的全部消息隊列
        ConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
        if (null == map) {
            ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
            ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
            if (oldMap != null) {
                map = oldMap;
            } else {
                map = newMap;
            }
        }

//        按queue id查找消費者隊列
        ConsumeQueue logic = map.get(queueId);
        if (null == logic) {
            ConsumeQueue newLogic = new ConsumeQueue(
                topic,
                queueId,
//                消費者隊列存儲地址 user.home/store/consumequeue
                StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
//                每一個文件存儲默認30W
                this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
                this);
            ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
            if (oldLogic != null) {
                logic = oldLogic;
            } else {
                logic = newLogic;
            }
        }

        return logic;
    }

進入這個方法org.apache.rocketmq.store.ConsumeQueue#getIndexBuffer按consumeQueueOffset查詢SelectMappedBufferResult

public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
        int mappedFileSize = this.mappedFileSize;
//        獲取最小的物理offset
        long offset = startIndex * CQ_STORE_UNIT_SIZE;
        if (offset >= this.getMinLogicOffset()) {
//            根據offset查詢映射文件 =》
            MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
            if (mappedFile != null) {
                SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
                return result;
            }
        }
        return null;
    }

進入這個方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)按consumeQueueOffset查詢MappedFile

public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
        try {
//            獲取隊列中第一個映射文件
            MappedFile firstMappedFile = this.getFirstMappedFile();
//            獲取隊列中最後一個映射文件
            MappedFile lastMappedFile = this.getLastMappedFile();
            if (firstMappedFile != null && lastMappedFile != null) {
//                若是offset不在索引文件的offset範圍內
                if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
                    LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                        offset,
                        firstMappedFile.getFileFromOffset(),
                        lastMappedFile.getFileFromOffset() + this.mappedFileSize,
                        this.mappedFileSize,
                        this.mappedFiles.size());
                } else {
//                   找到映射文件在隊列中的索引位置
                    int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
                    MappedFile targetFile = null;
                    try {
//                        獲取索引文件
                        targetFile = this.mappedFiles.get(index);
                    } catch (Exception ignored) {
                    }

//                    offset在目標文件的起始offset和結束offset範圍內
                    if (targetFile != null && offset >= targetFile.getFileFromOffset()
                        && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
                        return targetFile;
                    }

//                    若是按索引在隊列中找不到映射文件就遍歷隊列查找映射文件
                    for (MappedFile tmpMappedFile : this.mappedFiles) {
                        if (offset >= tmpMappedFile.getFileFromOffset()
                            && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
                            return tmpMappedFile;
                        }
                    }
                }

//                若是offset=0獲取隊列中第一個映射文件，我的感受這個邏輯是否放在前面判斷更爲合理，仍是放在這裏另有深意
                if (returnFirstOnNotFound) {
                    return firstMappedFile;
                }
            }
        } catch (Exception e) {
            log.error("findMappedFileByOffset Exception", e);
        }

        return null;
    }

進入這個方法org.apache.rocketmq.store.DefaultMessageStore#getStoreTime按SelectMappedBufferResult查詢存儲時間

private long getStoreTime(SelectMappedBufferResult result) {
        if (result != null) {
            try {
                final long phyOffset = result.getByteBuffer().getLong();
                final int size = result.getByteBuffer().getInt();
//                根據SelectMappedBufferResult的offset和大小查找存儲時間=》
                long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
                return storeTime;
            } catch (Exception e) {
            } finally {
                result.release();
            }
        }
        return -1;
    }

進入這個方法org.apache.rocketmq.store.CommitLog#pickupStoreTimestamp按物理offset、文件大小查詢存儲時間

public long pickupStoreTimestamp(final long offset, final int size) {
        if (offset >= this.getMinOffset()) {
//            =》
            SelectMappedBufferResult result = this.getMessage(offset, size);
            if (null != result) {
                try {
//                    獲取消息存儲時間
                    return result.getByteBuffer().getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
                } finally {
                    result.release();
                }
            }
        }

        return -1;
    }

進入這個方法org.apache.rocketmq.store.CommitLog#getMessage 按物理offset和文件大小查詢SelectMappedBufferResult

public SelectMappedBufferResult getMessage(final long offset, final int size) {
        int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();
//        根據offset找到映射文件 =》
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
        if (mappedFile != null) {
            int pos = (int) (offset % mappedFileSize);
            return mappedFile.selectMappedBuffer(pos, size);
        }
        return null;
    }

進入這個方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)按物理offset查詢MappedFile

public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
        try {
//            獲取隊列中第一個映射文件
            MappedFile firstMappedFile = this.getFirstMappedFile();
//            獲取隊列中最後一個映射文件
            MappedFile lastMappedFile = this.getLastMappedFile();
            if (firstMappedFile != null && lastMappedFile != null) {
//                若是offset不在索引文件的offset範圍內
                if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
                    LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                        offset,
                        firstMappedFile.getFileFromOffset(),
                        lastMappedFile.getFileFromOffset() + this.mappedFileSize,
                        this.mappedFileSize,
                        this.mappedFiles.size());
                } else {
//                   找到映射文件在隊列中的索引位置
                    int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
                    MappedFile targetFile = null;
                    try {
//                        獲取索引文件
                        targetFile = this.mappedFiles.get(index);
                    } catch (Exception ignored) {
                    }

//                    offset在目標文件的起始offset和結束offset範圍內
                    if (targetFile != null && offset >= targetFile.getFileFromOffset()
                        && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
                        return targetFile;
                    }

//                    若是按索引在隊列中找不到映射文件就遍歷隊列查找映射文件
                    for (MappedFile tmpMappedFile : this.mappedFiles) {
                        if (offset >= tmpMappedFile.getFileFromOffset()
                            && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
                            return tmpMappedFile;
                        }
                    }
                }

//                若是offset=0獲取隊列中第一個映射文件，我的感受這個邏輯是否放在前面判斷更爲合理，仍是放在這裏另有深意
                if (returnFirstOnNotFound) {
                    return firstMappedFile;
                }
            }
        } catch (Exception e) {
            log.error("findMappedFileByOffset Exception", e);
        }

        return null;
    }

往上返回到這個方法org.apache.rocketmq.broker.processor.AdminBrokerProcessor#getConsumeStats結束

說在最後

本次解析僅表明我的觀點，僅供參考。

加入技術微信羣