rocketmq源码解析之管理请求clone消费组的offset

时间 2019-11-11

标签 rocketmq 源码解析管理请求 clone 消费 offset 繁體版

原文原文链接

说在前面apache

管理请求之clone消费组的offset缓存

源码解析微信

进入这个方法org.apache.rocketmq.broker.processor.AdminBrokerProcessor#cloneGroupOffsetthis






private RemotingCommand cloneGroupOffset(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final RemotingCommand response = RemotingCommand.createResponseCommand(null);CloneGroupOffsetRequestHeader requestHeader =(CloneGroupOffsetRequestHeader) request.decodeCommandCustomHeader(CloneGroupOffsetRequestHeader.class);Set<String> topics;if (UtilAll.isBlank(requestHeader.getTopic())) {//            查询那些topic被这个消费组在消费=》topics = this.brokerController.getConsumerOffsetManager().whichTopicByConsumer(requestHeader.getSrcGroup());} else {topics = new HashSet<String>();topics.add(requestHeader.getTopic());}for (String topic : topics) {//            按topic查询到topic的配置信息TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(topic);if (null == topicConfig) {log.warn("[cloneGroupOffset], topic config not exist, {}", topic);continue;}if (!requestHeader.isOffline()) {//                查询topic、消费组的订阅信息SubscriptionData findSubscriptionData =this.brokerController.getConsumerManager().findSubscriptionData(requestHeader.getSrcGroup(), topic);if (this.brokerController.getConsumerManager().findSubscriptionDataCount(requestHeader.getSrcGroup()) > 0&& findSubscriptionData == null) {log.warn("[cloneGroupOffset], the consumer group[{}], topic[{}] not exist", requestHeader.getSrcGroup(), topic);continue;}}//            clone offset=》this.brokerController.getConsumerOffsetManager().cloneOffset(requestHeader.getSrcGroup(), requestHeader.getDestGroup(),requestHeader.getTopic());}response.setCode(ResponseCode.SUCCESS);response.setRemark(null);return response;}

进入这个方法org.apache.rocketmq.broker.offset.ConsumerOffsetManager#whichTopicByConsumer，查询那些topic被这个消费组在消费code


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的配置信息blog

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

进入这个方法，查询topic、消费组的订阅信息，org.apache.rocketmq.broker.client.ConsumerManager#findSubscriptionDataip

public SubscriptionData findSubscriptionData(final String group, final String topic) {//        按组从缓存中获取消费组信息ConsumerGroupInfo consumerGroupInfo = this.getConsumerGroupInfo(group);if (consumerGroupInfo != null) {//            按topic从缓存中获取订阅数据return consumerGroupInfo.findSubscriptionData(topic);}return null;}

进入这个方法，clone offset，org.apache.rocketmq.broker.offset.ConsumerOffsetManager#cloneOffsetget

public void cloneOffset(final String srcGroup, final String destGroup, final String topic) {ConcurrentMap<Integer, Long> offsets = this.offsetTable.get(topic + TOPIC_GROUP_SEPARATOR + srcGroup);if (offsets != null) {this.offsetTable.put(topic + TOPIC_GROUP_SEPARATOR + destGroup, new ConcurrentHashMap<Integer, Long>(offsets));}}

往上返回到这个方法，org.apache.rocketmq.broker.processor.AdminBrokerProcessor#cloneGroupOffset结束。源码

说在最后it

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

加入技术微信群

钉钉技术群