前面分析了rocketmq-namesrv的源码

其实很简单，lock+map 把注册中心的事情做了。  因为nameSrv 之间不用保证数据一致性。 每个节点都是独立的

本文接着学习 rocketmq-tools, 这个包主要实现了对mqadmin的支持

1.admin

admin 也是一个client端

DefaultMQAdminExt 和 DefaultMQAdminExtImpl  封装了一些mqadmin命令实现

2.command

把命令行的操作转换成java类，基于commons-cli 来做的一些封装

/** * @author shijia.wxr */ public interface SubCommand { public String commandName(); public String commandDesc(); public Options buildCommandlineOptions(final Options options); public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook); }

MQAdminStartup

public static void initCommand() { initCommand(new UpdateTopicSubCommand()); initCommand(new DeleteTopicSubCommand()); initCommand(new UpdateSubGroupSubCommand()); initCommand(new DeleteSubscriptionGroupCommand()); initCommand(new UpdateBrokerConfigSubCommand()); initCommand(new UpdateTopicPermSubCommand()); initCommand(new TopicRouteSubCommand()); initCommand(new TopicStatusSubCommand()); initCommand(new TopicClusterSubCommand()); initCommand(new BrokerStatusSubCommand()); initCommand(new QueryMsgByIdSubCommand()); initCommand(new QueryMsgByKeySubCommand()); initCommand(new QueryMsgByUniqueKeySubCommand()); initCommand(new QueryMsgByOffsetSubCommand()); initCommand(new QueryMsgByUniqueKeySubCommand()); initCommand(new PrintMessageSubCommand()); initCommand(new SendMsgStatusCommand()); initCommand(new BrokerConsumeStatsSubCommad()); initCommand(new ProducerConnectionSubCommand()); initCommand(new ConsumerConnectionSubCommand()); initCommand(new ConsumerProgressSubCommand()); initCommand(new ConsumerStatusSubCommand()); initCommand(new CloneGroupOffsetCommand()); initCommand(new ClusterListSubCommand()); initCommand(new TopicListSubCommand()); initCommand(new UpdateKvConfigCommand()); initCommand(new DeleteKvConfigCommand()); initCommand(new WipeWritePermSubCommand()); initCommand(new ResetOffsetByTimeCommand()); initCommand(new UpdateOrderConfCommand()); initCommand(new CleanExpiredCQSubCommand()); initCommand(new CleanUnusedTopicCommand()); initCommand(new StartMonitoringSubCommand()); initCommand(new StatsAllSubCommand()); initCommand(new SyncDocsToGithubSubCommand()); initCommand(new AllocateMQSubCommand()); initCommand(new CheckMsgSendRTCommand()); initCommand(new CLusterSendMsgRTCommand()); }

3.github

上传wiki文件

4.monitor 

public void start() throws MQClientException { this.defaultMQPullConsumer.start(); this.defaultMQAdminExt.start(); this.defaultMQPushConsumer.start(); this.startScheduleTask(); }

private void startScheduleTask() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { MonitorService.this.doMonitorWork(); } catch (Exception e) { log.error("doMonitorWork Exception", e); } } }, 1000 * 20, this.monitorConfig.getRoundInterval(), TimeUnit.MILLISECONDS); } public void doMonitorWork() throws RemotingException, MQClientException, InterruptedException { long beginTime = System.currentTimeMillis(); this.monitorListener.beginRound(); TopicList topicList = defaultMQAdminExt.fetchAllTopicList(); for (String topic : topicList.getTopicList()) { if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { String consumerGroup = topic.substring(MixAll.RETRY_GROUP_TOPIC_PREFIX.length()); try { this.reportUndoneMsgs(consumerGroup); } catch (Exception e) { // log.error("reportUndoneMsgs Exception", e); } try { this.reportConsumerRunningInfo(consumerGroup); } catch (Exception e) { // log.error("reportConsumerRunningInfo Exception", e); } } } this.monitorListener.endRound(); long spentTimeMills = System.currentTimeMillis() - beginTime; log.info("Execute one round monitor work, spent timemills: {}", spentTimeMills);

}

reportUndoneMsgs

private void reportUndoneMsgs(final String consumerGroup) { ConsumeStats cs = null; try { cs = defaultMQAdminExt.examineConsumeStats(consumerGroup); } catch (Exception e) { return; } ConsumerConnection cc = null; try { cc = defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup); } catch (Exception e) { return; } if (cs != null) { HashMap<String/* Topic */, ConsumeStats> csByTopic = new HashMap<String, ConsumeStats>(); { Iterator<Entry<MessageQueue, OffsetWrapper>> it = cs.getOffsetTable().entrySet().iterator(); while (it.hasNext()) { Entry<MessageQueue, OffsetWrapper> next = it.next(); MessageQueue mq = next.getKey(); OffsetWrapper ow = next.getValue(); ConsumeStats csTmp = csByTopic.get(mq.getTopic()); if (null == csTmp) { csTmp = new ConsumeStats(); csByTopic.put(mq.getTopic(), csTmp); } csTmp.getOffsetTable().put(mq, ow); } } { Iterator<Entry<String, ConsumeStats>> it = csByTopic.entrySet().iterator(); while (it.hasNext()) { Entry<String, ConsumeStats> next = it.next(); UndoneMsgs undoneMsgs = new UndoneMsgs(); undoneMsgs.setConsumerGroup(consumerGroup); undoneMsgs.setTopic(next.getKey()); this.computeUndoneMsgs(undoneMsgs, next.getValue()); this.monitorListener.reportUndoneMsgs(undoneMsgs); this.reportFailedMsgs(consumerGroup, next.getKey()); } } } }

reportConsumerRunningInfo

public void reportConsumerRunningInfo(final String consumerGroup) throws InterruptedException, MQBrokerException, RemotingException, MQClientException { ConsumerConnection cc = defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup); TreeMap<String, ConsumerRunningInfo> infoMap = new TreeMap<String, ConsumerRunningInfo>(); for (Connection c : cc.getConnectionSet()) { String clientId = c.getClientId(); if (c.getVersion() < MQVersion.Version.V3_1_8_SNAPSHOT.ordinal()) { continue; } try { ConsumerRunningInfo info = defaultMQAdminExt.getConsumerRunningInfo(consumerGroup, clientId, false); infoMap.put(clientId, info); } catch (Exception e) { } } if (!infoMap.isEmpty()) { this.monitorListener.reportConsumerRunningInfo(infoMap); } }