opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-19 10:54:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

Command line tool QueryMsgByKey& QueryMsgTraceById：time range and number of data bars can be specified

Browse Source
This commit is contained in:
loboxu
2023-02-27 12:46:48 +08:00
committed by yukon
parent 3bdab84c58
commit 254deb079a
2 changed files with 57 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByKeySubCommand.java
+27 -3
View File
@@ -49,6 +49,18 @@ public class QueryMsgByKeySubCommand implements SubCommand {
opt.setRequired(true);
options.addOption(opt);
opt = new Option("b", "beginTimestamp", true, "Begin timestamp(ms). default:0, eg:1676730526212");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("e", "endTimestamp", true, "End timestamp(ms). default:Long.MAX_VALUE, eg:1676730526212");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("num", "maxNum", true, "The maximum number of messages returned by the query, default:64");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -62,7 +74,19 @@ public class QueryMsgByKeySubCommand implements SubCommand {
final String topic = commandLine.getOptionValue('t').trim();
final String key = commandLine.getOptionValue('k').trim();
this.queryByKey(defaultMQAdminExt, topic, key);
long beginTimestamp = 0;
long endTimestamp = Long.MAX_VALUE;
int maxNum = 64;
if(commandLine.hasOption("b")) {
beginTimestamp = Long.parseLong(commandLine.getOptionValue("b").trim());
}
if(commandLine.hasOption("e")) {
endTimestamp = Long.parseLong(commandLine.getOptionValue("e").trim());
}
if(commandLine.hasOption("num")) {
maxNum = Integer.parseInt(commandLine.getOptionValue("num").trim());
}
this.queryByKey(defaultMQAdminExt, topic, key, maxNum, beginTimestamp, endTimestamp);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
} finally {
@@ -70,11 +94,11 @@ public class QueryMsgByKeySubCommand implements SubCommand {
}
}
private void queryByKey(final DefaultMQAdminExt admin, final String topic, final String key)
private void queryByKey(final DefaultMQAdminExt admin, final String topic, final String key, int maxNum, long begin, long end)
throws MQClientException, InterruptedException {
admin.start();
QueryResult queryResult = admin.queryMessage(topic, key, 64, 0, Long.MAX_VALUE);
QueryResult queryResult = admin.queryMessage(topic, key, maxNum, begin, end);
System.out.printf("%-50s %4s %40s%n",
"#Message ID",
"#QID",
tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgTraceByIdSubCommand.java
+30 -3
View File
@@ -47,6 +47,19 @@ public class QueryMsgTraceByIdSubCommand implements SubCommand {
opt = new Option("t", "traceTopic", true, "The name value of message trace topic");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("b", "beginTimestamp", true, "Begin timestamp(ms). default:0, eg:1676730526212");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("e", "endTimestamp", true, "End timestamp(ms). default:Long.MAX_VALUE, eg:1676730526212");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("num", "maxNum", true, "The maximum number of messages returned by the query, default:64");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -78,7 +91,21 @@ public class QueryMsgTraceByIdSubCommand implements SubCommand {
if (commandLine.hasOption('n')) {
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
}
this.queryTraceByMsgId(defaultMQAdminExt, traceTopic, msgId);
long beginTimestamp = 0;
long endTimestamp = Long.MAX_VALUE;
int maxNum = 64;
if(commandLine.hasOption("b")) {
beginTimestamp = Long.parseLong(commandLine.getOptionValue("b").trim());
}
if(commandLine.hasOption("e")) {
endTimestamp = Long.parseLong(commandLine.getOptionValue("e").trim());
}
if(commandLine.hasOption("num")) {
maxNum = Integer.parseInt(commandLine.getOptionValue("num").trim());
}
this.queryTraceByMsgId(defaultMQAdminExt, traceTopic, msgId, maxNum, beginTimestamp, endTimestamp);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + "command failed", e);
} finally {
@@ -86,10 +113,10 @@ public class QueryMsgTraceByIdSubCommand implements SubCommand {
}
}
private void queryTraceByMsgId(final DefaultMQAdminExt admin, String traceTopic, String msgId)
private void queryTraceByMsgId(final DefaultMQAdminExt admin, String traceTopic, String msgId, int maxNum, long begin, long end)
throws MQClientException, InterruptedException {
admin.start();
QueryResult queryResult = admin.queryMessage(traceTopic, msgId, 64, 0, System.currentTimeMillis());
QueryResult queryResult = admin.queryMessage(traceTopic, msgId, maxNum, begin, end);
List<MessageExt> messageList = queryResult.getMessageList();
List<TraceView> traceViews = new ArrayList<>();
for (MessageExt message : messageList) {