rocketmq源码解析之管理请求查询消费时间①

时间 2019-11-11

标签 rocketmq 源码解析管理请求查询消费时间繁體版

原文原文链接

说在前面java

管理请求 QUERY_CONSUME_TIME_SPAN 查询消费时间apache

源码解析缓存

进入这个方法org.apache.rocketmq.broker.processor.AdminBrokerProcessor#queryConsumeTimeSpan查询消费时间微信

private RemotingCommand queryConsumeTimeSpan(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        QueryConsumeTimeSpanRequestHeader requestHeader =
            (QueryConsumeTimeSpanRequestHeader) request.decodeCommandCustomHeader(QueryConsumeTimeSpanRequestHeader.class);
        final String topic = requestHeader.getTopic();
//        按topic从缓存中查询出topic配置信息=》
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);
        if (null == topicConfig) {
            response.setCode(ResponseCode.TOPIC_NOT_EXIST);
            response.setRemark("topic[" + topic + "] not exist");
            return response;
        }

        List<QueueTimeSpan> timeSpanSet = new ArrayList<QueueTimeSpan>();
        for (int i = 0; i < topicConfig.getWriteQueueNums(); i++) {
            QueueTimeSpan timeSpan = new QueueTimeSpan();
            MessageQueue mq = new MessageQueue();
            mq.setTopic(topic);
            mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());
            mq.setQueueId(i);
            timeSpan.setMessageQueue(mq);
//            按topic、queueId查询最先消息的时间=》
            long minTime = this.brokerController.getMessageStore().getEarliestMessageTime(topic, i);
            timeSpan.setMinTimeStamp(minTime);
//            按topic、queueId查询最大的offset=》
            long max = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
//            按topic、queueId、consumerOffset查询最大时间=》
            long maxTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, max - 1);
            timeSpan.setMaxTimeStamp(maxTime);
            long consumeTime;
//            按consumerGroup、topic、queueId查询consumerOffset=》
            long consumerOffset = this.brokerController.getConsumerOffsetManager().queryOffset(
                requestHeader.getGroup(), topic, i);
            if (consumerOffset > 0) {
//                按topic、queueId、consumerOffset查询消费时间=》
                consumeTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, consumerOffset - 1);
            } else {
                consumeTime = minTime;
            }
            timeSpan.setConsumeTimeStamp(consumeTime);
//            按topic、queueId查询最大的offset=》
            long maxBrokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(requestHeader.getTopic(), i);
            if (consumerOffset < maxBrokerOffset) {
//                按topic、queueId、consumerOffset查询消费时间=》
                long nextTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, consumerOffset);
                timeSpan.setDelayTime(System.currentTimeMillis() - nextTime);
            }
            timeSpanSet.add(timeSpan);
        }

        QueryConsumeTimeSpanBody queryConsumeTimeSpanBody = new QueryConsumeTimeSpanBody();
        queryConsumeTimeSpanBody.setConsumeTimeSpanSet(timeSpanSet);
        response.setBody(queryConsumeTimeSpanBody.encode());
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

进入这个方法org.apache.rocketmq.broker.topic.TopicConfigManager#selectTopicConfig按topic查询topic配置信息app

public TopicConfig selectTopicConfig(final String topic) {
//        从topic配置缓存信息中查询当前topic的配置
        return this.topicConfigTable.get(topic);
    }

往上返回进入这个方法org.apache.rocketmq.store.DefaultMessageStore#getEarliestMessageTime(java.lang.String, int)按topic、queueId查询最先消息的时间ide

@Override
    public long getEarliestMessageTime(String topic, int queueId) {
//        按topic和queueId查询消费者队列
        ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
        if (logicQueue != null) {
//            获取最小的物理offset
            long minLogicOffset = logicQueue.getMinLogicOffset();
//            根据最小的offset获取SelectMappedBufferResult =》
            SelectMappedBufferResult result = logicQueue.getIndexBuffer(minLogicOffset / ConsumeQueue.CQ_STORE_UNIT_SIZE);
//            根据SelectMappedBufferResult查找存储时间 =》
            return getStoreTime(result);
        }

        return -1;
    }

按topic和queueId查询消费者队列，进入这个方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueuethis

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;
    }

根据最小的offset获取SelectMappedBufferResult，进入这个方法org.apache.rocketmq.store.ConsumeQueue#getIndexBuffercode

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;
    }

根据offset查询映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)blog

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;
    }

根据SelectMappedBufferResult查找存储时间，往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getStoreTime索引

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;
    }

根据SelectMappedBufferResult的offset和大小查找存储时间，进入这个方法org.apache.rocketmq.store.CommitLog#pickupStoreTimestamp

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

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;
    }

根据offset找到映射文件，进入这个方法org.apache.rocketmq.store.MappedFileQueue#findMappedFileByOffset(long, boolean)

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;
    }

按topic、queueId查询最大的offset，往上返回到这个方法org.apache.rocketmq.store.DefaultMessageStore#getMaxOffsetInQueue

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;
    }

根据topic和queueId找到消费者队列，进入这个方法org.apache.rocketmq.store.DefaultMessageStore#findConsumeQueue

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;
    }

往上返回到这个方法获取最大的offset

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;
}

接下篇

说在最后

本次解析仅表明我的观点，仅供参考。

加入技术微信群

钉钉技术群