@Test public void testExecute() throws SubCommandException, IllegalAccessException, NoSuchFieldException { Field field = BrokerConsumeStatsSubCommad.class.getDeclaredField("defaultMQAdminExt"); field.setAccessible(true); field.set(cmd, defaultMQAdminExt); Options options = ServerUtil.buildCommandlineOptions(new Options()); String[] subargs = new String[] {"-b 127.0.0.1:10911", "-t 3000", "-l 5", "-o true"}; final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, cmd.buildCommandlineOptions(options), new PosixParser()); cmd.execute(commandLine, options, null); }
@Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { try { defaultMQAdminExt = createMQAdminExt(rpcHook); String brokerAddr = commandLine.getOptionValue('b').trim(); boolean isOrder = false; long timeoutMillis = 50000; long diffLevel = 0; if (commandLine.hasOption('o')) { isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim()); } if (commandLine.hasOption('t')) { timeoutMillis = Long.parseLong(commandLine.getOptionValue('t').trim()); } if (commandLine.hasOption('l')) { diffLevel = Long.parseLong(commandLine.getOptionValue('l').trim()); } ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis); System.out.printf("%-32s %-32s %-32s %-4s %-20s %-20s %-20s %s%n", "#Topic", "#Group", "#Broker Name", "#QID", "#Broker Offset", "#Consumer Offset", "#Diff", "#LastTime"); for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) { for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) { String group = entry.getKey(); List<ConsumeStats> consumeStatsArray = entry.getValue(); for (ConsumeStats consumeStats : consumeStatsArray) { List<MessageQueue> mqList = new LinkedList<MessageQueue>(); mqList.addAll(consumeStats.getOffsetTable().keySet()); Collections.sort(mqList); for (MessageQueue mq : mqList) { OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq); long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset(); if (diff < diffLevel) { continue; } String lastTime = "-"; try { lastTime = UtilAll.formatDate(new Date(offsetWrapper.getLastTimestamp()), UtilAll.YYYY_MM_DD_HH_MM_SS); } catch (Exception ignored) { } if (offsetWrapper.getLastTimestamp() > 0) System.out.printf("%-32s %-32s %-32s %-4d %-20d %-20d %-20d %s%n", UtilAll.frontStringAtLeast(mq.getTopic(), 32), group, UtilAll.frontStringAtLeast(mq.getBrokerName(), 32), mq.getQueueId(), offsetWrapper.getBrokerOffset(), offsetWrapper.getConsumerOffset(), diff, lastTime ); } } } } System.out.printf("%nDiff Total: %d%n", consumeStatsList.getTotalDiff()); } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }
BrokerConsumeStatsSubCommad implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { try { defaultMQAdminExt = createMQAdminExt(rpcHook); String brokerAddr = commandLine.getOptionValue('b').trim(); boolean isOrder = false; long timeoutMillis = 50000; long diffLevel = 0; if (commandLine.hasOption('o')) { isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim()); } if (commandLine.hasOption('t')) { timeoutMillis = Long.parseLong(commandLine.getOptionValue('t').trim()); } if (commandLine.hasOption('l')) { diffLevel = Long.parseLong(commandLine.getOptionValue('l').trim()); } ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis); System.out.printf("%-32s %-32s %-32s %-4s %-20s %-20s %-20s %s%n", "#Topic", "#Group", "#Broker Name", "#QID", "#Broker Offset", "#Consumer Offset", "#Diff", "#LastTime"); for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) { for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) { String group = entry.getKey(); List<ConsumeStats> consumeStatsArray = entry.getValue(); for (ConsumeStats consumeStats : consumeStatsArray) { List<MessageQueue> mqList = new LinkedList<MessageQueue>(); mqList.addAll(consumeStats.getOffsetTable().keySet()); Collections.sort(mqList); for (MessageQueue mq : mqList) { OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq); long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset(); if (diff < diffLevel) { continue; } String lastTime = "-"; try { lastTime = UtilAll.formatDate(new Date(offsetWrapper.getLastTimestamp()), UtilAll.YYYY_MM_DD_HH_MM_SS); } catch (Exception ignored) { } if (offsetWrapper.getLastTimestamp() > 0) System.out.printf("%-32s %-32s %-32s %-4d %-20d %-20d %-20d %s%n", UtilAll.frontStringAtLeast(mq.getTopic(), 32), group, UtilAll.frontStringAtLeast(mq.getBrokerName(), 32), mq.getQueueId(), offsetWrapper.getBrokerOffset(), offsetWrapper.getConsumerOffset(), diff, lastTime ); } } } } System.out.printf("%nDiff Total: %d%n", consumeStatsList.getTotalDiff()); } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } } }
BrokerConsumeStatsSubCommad implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { try { defaultMQAdminExt = createMQAdminExt(rpcHook); String brokerAddr = commandLine.getOptionValue('b').trim(); boolean isOrder = false; long timeoutMillis = 50000; long diffLevel = 0; if (commandLine.hasOption('o')) { isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim()); } if (commandLine.hasOption('t')) { timeoutMillis = Long.parseLong(commandLine.getOptionValue('t').trim()); } if (commandLine.hasOption('l')) { diffLevel = Long.parseLong(commandLine.getOptionValue('l').trim()); } ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis); System.out.printf("%-32s %-32s %-32s %-4s %-20s %-20s %-20s %s%n", "#Topic", "#Group", "#Broker Name", "#QID", "#Broker Offset", "#Consumer Offset", "#Diff", "#LastTime"); for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) { for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) { String group = entry.getKey(); List<ConsumeStats> consumeStatsArray = entry.getValue(); for (ConsumeStats consumeStats : consumeStatsArray) { List<MessageQueue> mqList = new LinkedList<MessageQueue>(); mqList.addAll(consumeStats.getOffsetTable().keySet()); Collections.sort(mqList); for (MessageQueue mq : mqList) { OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq); long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset(); if (diff < diffLevel) { continue; } String lastTime = "-"; try { lastTime = UtilAll.formatDate(new Date(offsetWrapper.getLastTimestamp()), UtilAll.YYYY_MM_DD_HH_MM_SS); } catch (Exception ignored) { } if (offsetWrapper.getLastTimestamp() > 0) System.out.printf("%-32s %-32s %-32s %-4d %-20d %-20d %-20d %s%n", UtilAll.frontStringAtLeast(mq.getTopic(), 32), group, UtilAll.frontStringAtLeast(mq.getBrokerName(), 32), mq.getQueueId(), offsetWrapper.getBrokerOffset(), offsetWrapper.getConsumerOffset(), diff, lastTime ); } } } } System.out.printf("%nDiff Total: %d%n", consumeStatsList.getTotalDiff()); } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  }
BrokerConsumeStatsSubCommad implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { try { defaultMQAdminExt = createMQAdminExt(rpcHook); String brokerAddr = commandLine.getOptionValue('b').trim(); boolean isOrder = false; long timeoutMillis = 50000; long diffLevel = 0; if (commandLine.hasOption('o')) { isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim()); } if (commandLine.hasOption('t')) { timeoutMillis = Long.parseLong(commandLine.getOptionValue('t').trim()); } if (commandLine.hasOption('l')) { diffLevel = Long.parseLong(commandLine.getOptionValue('l').trim()); } ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis); System.out.printf("%-32s %-32s %-32s %-4s %-20s %-20s %-20s %s%n", "#Topic", "#Group", "#Broker Name", "#QID", "#Broker Offset", "#Consumer Offset", "#Diff", "#LastTime"); for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) { for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) { String group = entry.getKey(); List<ConsumeStats> consumeStatsArray = entry.getValue(); for (ConsumeStats consumeStats : consumeStatsArray) { List<MessageQueue> mqList = new LinkedList<MessageQueue>(); mqList.addAll(consumeStats.getOffsetTable().keySet()); Collections.sort(mqList); for (MessageQueue mq : mqList) { OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq); long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset(); if (diff < diffLevel) { continue; } String lastTime = "-"; try { lastTime = UtilAll.formatDate(new Date(offsetWrapper.getLastTimestamp()), UtilAll.YYYY_MM_DD_HH_MM_SS); } catch (Exception ignored) { } if (offsetWrapper.getLastTimestamp() > 0) System.out.printf("%-32s %-32s %-32s %-4d %-20d %-20d %-20d %s%n", UtilAll.frontStringAtLeast(mq.getTopic(), 32), group, UtilAll.frontStringAtLeast(mq.getBrokerName(), 32), mq.getQueueId(), offsetWrapper.getBrokerOffset(), offsetWrapper.getConsumerOffset(), diff, lastTime ); } } } } System.out.printf("%nDiff Total: %d%n", consumeStatsList.getTotalDiff()); } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(Options options); @Override void execute(CommandLine commandLine, Options options, RPCHook rpcHook); }
BrokerConsumeStatsSubCommad implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { try { defaultMQAdminExt = createMQAdminExt(rpcHook); String brokerAddr = commandLine.getOptionValue('b').trim(); boolean isOrder = false; long timeoutMillis = 50000; long diffLevel = 0; if (commandLine.hasOption('o')) { isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim()); } if (commandLine.hasOption('t')) { timeoutMillis = Long.parseLong(commandLine.getOptionValue('t').trim()); } if (commandLine.hasOption('l')) { diffLevel = Long.parseLong(commandLine.getOptionValue('l').trim()); } ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis); System.out.printf("%-32s %-32s %-32s %-4s %-20s %-20s %-20s %s%n", "#Topic", "#Group", "#Broker Name", "#QID", "#Broker Offset", "#Consumer Offset", "#Diff", "#LastTime"); for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) { for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) { String group = entry.getKey(); List<ConsumeStats> consumeStatsArray = entry.getValue(); for (ConsumeStats consumeStats : consumeStatsArray) { List<MessageQueue> mqList = new LinkedList<MessageQueue>(); mqList.addAll(consumeStats.getOffsetTable().keySet()); Collections.sort(mqList); for (MessageQueue mq : mqList) { OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq); long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset(); if (diff < diffLevel) { continue; } String lastTime = "-"; try { lastTime = UtilAll.formatDate(new Date(offsetWrapper.getLastTimestamp()), UtilAll.YYYY_MM_DD_HH_MM_SS); } catch (Exception ignored) { } if (offsetWrapper.getLastTimestamp() > 0) System.out.printf("%-32s %-32s %-32s %-4d %-20d %-20d %-20d %s%n", UtilAll.frontStringAtLeast(mq.getTopic(), 32), group, UtilAll.frontStringAtLeast(mq.getBrokerName(), 32), mq.getQueueId(), offsetWrapper.getBrokerOffset(), offsetWrapper.getConsumerOffset(), diff, lastTime ); } } } } System.out.printf("%nDiff Total: %d%n", consumeStatsList.getTotalDiff()); } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(Options options); @Override void execute(CommandLine commandLine, Options options, RPCHook rpcHook);  }