import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
private boolean isBrokerAddrExistInTopicRouteTable(final String addr) {
    Iterator<Entry<String, TopicRouteData>> it = this.topicRouteTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, TopicRouteData> entry = it.next();
        TopicRouteData topicRouteData = entry.getValue();
        List<BrokerData> bds = topicRouteData.getBrokerDatas();
        for (BrokerData bd : bds) {
            if (bd.getBrokerAddrs() != null) {
                boolean exist = bd.getBrokerAddrs().containsValue(addr);
                if (exist)
                    return true;
            }
        }
    }

    return false;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException, InterruptedException,
        MQBrokerException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    TopicStatsTable topicStatsTable = new TopicStatsTable();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            TopicStatsTable tst = this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, timeoutMillis);
            topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
        }
    }

    if (topicStatsTable.getOffsetTable().isEmpty()) {
        throw new MQClientException("Not found the topic stats info", null);
    }

    return topicStatsTable;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ConsumeStats examineConsumeStats(String consumerGroup, String topic) throws RemotingException, MQClientException,
        InterruptedException, MQBrokerException {
    String retryTopic = MixAll.getRetryTopic(consumerGroup);
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
    ConsumeStats result = new ConsumeStats();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            ConsumeStats consumeStats =
                    this.mqClientInstance.getMQClientAPIImpl().getConsumeStats(addr, consumerGroup, topic, timeoutMillis * 3);
            result.getOffsetTable().putAll(consumeStats.getOffsetTable());
            double value = result.getConsumeTps() + consumeStats.getConsumeTps();
            result.setConsumeTps(value);
        }
    }

    if (result.getOffsetTable().isEmpty()) {
        throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
            "Not found the consumer group consume stats, because return offset table is empty, maybe the consumer not consume any message");
    }

    return result;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup) throws InterruptedException, MQBrokerException,
        RemotingException, MQClientException {
    String topic = MixAll.getRetryTopic(consumerGroup);
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    ConsumerConnection result = new ConsumerConnection();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mqClientInstance.getMQClientAPIImpl().getConsumerConnectionList(addr, consumerGroup, timeoutMillis);
        }
    }

    if (result.getConnectionSet().isEmpty()) {
        throw new MQClientException(ResponseCode.CONSUMER_NOT_ONLINE, "Not found the consumer group connection");
    }

    return result;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ProducerConnection examineProducerConnectionInfo(String producerGroup, final String topic) throws RemotingException,
        MQClientException, InterruptedException, MQBrokerException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    ProducerConnection result = new ProducerConnection();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mqClientInstance.getMQClientAPIImpl().getProducerConnectionList(addr, producerGroup, timeoutMillis);
        }
    }

    if (result.getConnectionSet().isEmpty()) {
        throw new MQClientException("Not found the consumer group connection", null);
    }

    return result;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce, boolean isC)
        throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();
    Map<MessageQueue, Long> allOffsetTable = new HashMap<MessageQueue, Long>();
    if (brokerDatas != null) {
        for (BrokerData brokerData : brokerDatas) {
            String addr = brokerData.selectBrokerAddr();
            if (addr != null) {
                Map<MessageQueue, Long> offsetTable =
                        this.mqClientInstance.getMQClientAPIImpl().invokeBrokerToResetOffset(addr, topic, group, timestamp, isForce,
                            timeoutMillis, isC);
                if (offsetTable != null) {
                    allOffsetTable.putAll(offsetTable);
                }
            }
        }
    }
    return allOffsetTable;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException, RemotingException,
        MQClientException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mqClientInstance.getMQClientAPIImpl().queryTopicConsumeByWho(addr, topic, timeoutMillis);
        }

        break;
    }

    return null;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack) throws RemotingException,
        MQClientException, InterruptedException {
    String topic = MixAll.RETRY_GROUP_TOPIC_PREFIX + consumerGroup;
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();
    if (brokerDatas != null) {
        for (BrokerData brokerData : brokerDatas) {
            String addr = brokerData.selectBrokerAddr();
            if (addr != null) {
                return this.mqClientInstance.getMQClientAPIImpl().getConsumerRunningInfo(addr, consumerGroup, clientId, jstack,
                    timeoutMillis * 3);
            }
        }
    }
    return null;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public Set<String> getTopicClusterList(final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException{
    Set<String> clusterSet = new HashSet<String>();
    ClusterInfo clusterInfo = examineBrokerClusterInfo();
    TopicRouteData topicRouteData = examineTopicRouteInfo(topic);
    BrokerData brokerData = topicRouteData.getBrokerDatas().get(0);
    String brokerName = brokerData.getBrokerName();
    Iterator<Map.Entry<String, Set<String>>> it = clusterInfo.getClusterAddrTable().entrySet().iterator();
    while (it.hasNext()) {
        Map.Entry<String, Set<String>> next = it.next();
        if (next.getValue().contains(brokerName)) {
            clusterSet.add(next.getKey());
        }
    }
    return clusterSet;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
private String findTopicBelongToWhichCluster(final String topic, final ClusterInfo clusterInfo,
        final DefaultMQAdminExt defaultMQAdminExt) throws RemotingException, MQClientException,
        InterruptedException {
    TopicRouteData topicRouteData = defaultMQAdminExt.examineTopicRouteInfo(topic);

    BrokerData brokerData = topicRouteData.getBrokerDatas().get(0);

    String brokerName = brokerData.getBrokerName();

    Iterator<Entry<String, Set<String>>> it = clusterInfo.getClusterAddrTable().entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, Set<String>> next = it.next();
        if (next.getValue().contains(brokerName)) {
            return next.getKey();
        }
    }

    return null;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
    public Map<String, Object> list() {
        try {
            Map<String, Object> resultMap = Maps.newHashMap();
            ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            logger.info("op=look_clusterInfo {}", JsonUtil.obj2String(clusterInfo));
            Map<String/*brokerName*/, Map<Long/* brokerId */, Object/* brokerDetail */>> brokerServer = Maps.newHashMap();
            for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
                Map<Long, Object> brokerMasterSlaveMap = Maps.newHashMap();
                for (Map.Entry<Long/* brokerId */, String/* broker address */> brokerAddr : brokerData.getBrokerAddrs().entrySet()) {
                    KVTable kvTable = mqAdminExt.fetchBrokerRuntimeStats(brokerAddr.getValue());
//                KVTable kvTable = mqAdminExt.fetchBrokerRuntimeStats("127.0.0.1:10911");
                    brokerMasterSlaveMap.put(brokerAddr.getKey(), kvTable.getTable());
                }
                brokerServer.put(brokerData.getBrokerName(), brokerMasterSlaveMap);
            }
            resultMap.put("clusterInfo", clusterInfo);
            resultMap.put("brokerServer", brokerServer);
            return resultMap;
        } catch (Exception err) {
            throw Throwables.propagate(err);
        }
    }
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
    public List<TopicConfigInfo> examineTopicConfig(String topic) {
        List<TopicConfigInfo> topicConfigInfoList = Lists.newArrayList();
        TopicRouteData topicRouteData = route(topic);
        for (BrokerData brokerData : topicRouteData.getBrokerDatas()) {
            TopicConfigInfo topicConfigInfo = new TopicConfigInfo();
            TopicConfig topicConfig = examineTopicConfig(topic, brokerData.getBrokerName());
            BeanUtils.copyProperties(topicConfig, topicConfigInfo);
            boolean hasSameTopicConfig = false;
//            for (TopicConfigInfo topicConfigInfoExist : topicConfigInfoList) {
//                if (topicConfigInfoExist.equals(topicConfigInfo)) {
//                    topicConfigInfoExist.getBrokerNameList().add(brokerData.getBrokerName());
//                    hasSameTopicConfig = true;
//                    break;
//                }
//            } //每一个broker的配置单独展示 变更 交互可以优化下
            if (!hasSameTopicConfig) {
                topicConfigInfo.setBrokerNameList(Lists.newArrayList(brokerData.getBrokerName()));
                topicConfigInfoList.add(topicConfigInfo);
            }
        }
        return topicConfigInfoList;
    }
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
public boolean consumed(final MessageExt msg, final String group) throws RemotingException, MQClientException, InterruptedException,
        MQBrokerException {
    ConsumeStats cstats = this.examineConsumeStats(group);

    ClusterInfo ci = this.examineBrokerClusterInfo();

    Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
    while (it.hasNext()) {
        Entry<MessageQueue, OffsetWrapper> next = it.next();
        MessageQueue mq = next.getKey();
        if (mq.getTopic().equals(msg.getTopic()) && mq.getQueueId() == msg.getQueueId()) {
            BrokerData brokerData = ci.getBrokerAddrTable().get(mq.getBrokerName());
            if (brokerData != null) {
                String addr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                if (addr.equals(RemotingUtil.socketAddress2String(msg.getStoreHost()))) {
                    if (next.getValue().getConsumerOffset() > msg.getQueueOffset()) {
                        return true;
                    }
                }
            }
        }
    }

    return false;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException,
        InterruptedException, MQBrokerException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    TopicStatsTable topicStatsTable = new TopicStatsTable();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            TopicStatsTable tst =
                    this.mQClientFactory.getMQClientAPIImpl().getTopicStatsInfo(addr, topic, 3000);
            topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
        }
    }

    if (topicStatsTable.getOffsetTable().isEmpty()) {
        throw new MQClientException("Not found the topic stats info", null);
    }

    return topicStatsTable;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup)
        throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
    String topic = MixAll.getRetryTopic(consumerGroup);
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    ConsumerConnection result = new ConsumerConnection();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mQClientFactory.getMQClientAPIImpl().getConsumerConnectionList(addr,
                consumerGroup, 3000);
        }
    }

    if (result.getConnectionSet().isEmpty()) {
        throw new MQClientException("Not found the consumer group connection", null);
    }

    return result;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public ProducerConnection examineProducerConnectionInfo(String producerGroup, final String topic)
        throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    ProducerConnection result = new ProducerConnection();

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mQClientFactory.getMQClientAPIImpl().getProducerConnectionList(addr,
                producerGroup, 300);
        }
    }

    if (result.getConnectionSet().isEmpty()) {
        throw new MQClientException("Not found the consumer group connection", null);
    }

    return result;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp,
        boolean isForce) throws RemotingException, MQBrokerException, InterruptedException,
        MQClientException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
    List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();
    // 每个 broker 上有所有的 consumer 连接，故只需要在一个 broker 执行即可。
    if (brokerDatas != null && brokerDatas.size() > 0) {
        String addr = brokerDatas.get(0).selectBrokerAddr();
        if (addr != null) {
            return this.mQClientFactory.getMQClientAPIImpl().invokeBrokerToResetOffset(addr, topic,
                group, timestamp, isForce, 5000);
        }
    }
    return Collections.EMPTY_MAP;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
@Override
public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException,
        RemotingException, MQClientException {
    TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);

    for (BrokerData bd : topicRouteData.getBrokerDatas()) {
        String addr = bd.selectBrokerAddr();
        if (addr != null) {
            return this.mQClientFactory.getMQClientAPIImpl().queryTopicConsumeByWho(addr, topic, 3000);
        }

        break;
    }

    return null;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
public List<QueueTimeSpan> queryConsumeTimeSpan(final String topic) throws RemotingException,
        MQClientException, InterruptedException, MQBrokerException {
    List<QueueTimeSpan> queueTimeSpan = new ArrayList<QueueTimeSpan>();
    TopicRouteData routeData =
            this.mQClientFactory.getMQClientAPIImpl().getTopicRouteInfoFromNameServer(topic, 3000);
    for (BrokerData brokerData : routeData.getBrokerDatas()) {
        String addr = brokerData.selectBrokerAddr();
        queueTimeSpan.addAll(this.mQClientFactory.getMQClientAPIImpl().queryConsumeTimeSpan(addr, topic,
                groupName(), 3000l));
    }

    return queueTimeSpan;
}
 

import com.alibaba.rocketmq.common.protocol.route.BrokerData; //导入依赖的package包/类
public String findBrokerAddrByTopic(final String topic) {
    TopicRouteData topicRouteData = this.topicRouteTable.get(topic);
    if (topicRouteData != null) {
        List<BrokerData> brokers = topicRouteData.getBrokerDatas();
        if (!brokers.isEmpty()) {
            BrokerData bd = brokers.get(0);
            return bd.selectBrokerAddr();
        }
    }

    return null;
}