Skip to content

Commit

Permalink
[ISSUE #6803] Benchmark support reportInterval option (#6804)
Browse files Browse the repository at this point in the history
  • Loading branch information
@DL1231
DL1231 authored May 27, 2023
1 parent ea32980 commit e545e3b
Show file tree
Hide file tree
Showing 4 changed files with 39 additions and 13 deletions.
19 changes: 15 additions & 4 deletions example/src/main/java/org/apache/rocketmq/example/benchmark/BatchProducer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,18 +74,19 @@ public static void main(String[] args) throws MQClientException {
final boolean msgTraceEnable = getOptionValue(commandLine, 'm', false);
final boolean aclEnable = getOptionValue(commandLine, 'a', false);
final boolean enableCompress = commandLine.hasOption('c') && Boolean.parseBoolean(commandLine.getOptionValue('c'));
final int reportInterval = commandLine.hasOption("ri") ? Integer.parseInt(commandLine.getOptionValue("ri")) : 10000;

System.out.printf("topic: %s, threadCount: %d, messageSize: %d, batchSize: %d, keyEnable: %s, propertySize: %d, tagCount: %d, traceEnable: %s, " +
"aclEnable: %s%n compressEnable: %s%n",
topic, threadCount, messageSize, batchSize, keyEnable, propertySize, tagCount, msgTraceEnable, aclEnable, enableCompress);
"aclEnable: %s%n compressEnable: %s, reportInterval: %d%n",
topic, threadCount, messageSize, batchSize, keyEnable, propertySize, tagCount, msgTraceEnable, aclEnable, enableCompress, reportInterval);

StringBuilder sb = new StringBuilder(messageSize);
for (int i = 0; i < messageSize; i++) {
sb.append(RandomStringUtils.randomAlphanumeric(1));
}
msgBody = sb.toString().getBytes(StandardCharsets.UTF_8);

final StatsBenchmarkBatchProducer statsBenchmark = new StatsBenchmarkBatchProducer();
final StatsBenchmarkBatchProducer statsBenchmark = new StatsBenchmarkBatchProducer(reportInterval);
statsBenchmark.start();

RPCHook rpcHook = null;
Expand Down Expand Up @@ -253,6 +254,10 @@ public static Options buildCommandlineOptions(final Options options) {
opt.setRequired(false);
options.addOption(opt);

opt = new Option("ri", "reportInterval", true, "The number of ms between reports, Default: 10000");
opt.setRequired(false);
options.addOption(opt);

return options;
}

Expand Down Expand Up @@ -359,6 +364,12 @@ class StatsBenchmarkBatchProducer {

private final LinkedList<Long[]> snapshotList = new LinkedList<>();

private final int reportInterval;

public StatsBenchmarkBatchProducer(int reportInterval) {
this.reportInterval = reportInterval;
}

public Long[] createSnapshot() {
Long[] snap = new Long[] {
System.currentTimeMillis(),
Expand Down Expand Up @@ -432,7 +443,7 @@ public void run() {
e.printStackTrace();
}
}
}, 10000, 10000, TimeUnit.MILLISECONDS);
}, reportInterval, reportInterval, TimeUnit.MILLISECONDS);
}

public void shutdown() {
Expand Down
11 changes: 8 additions & 3 deletions example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,14 +68,15 @@ public static void main(String[] args) throws MQClientException, IOException {
final boolean msgTraceEnable = commandLine.hasOption('m') && Boolean.parseBoolean(commandLine.getOptionValue('m'));
final boolean aclEnable = commandLine.hasOption('a') && Boolean.parseBoolean(commandLine.getOptionValue('a'));
final boolean clientRebalanceEnable = commandLine.hasOption('c') ? Boolean.parseBoolean(commandLine.getOptionValue('c')) : true;
final int reportInterval = commandLine.hasOption("ri") ? Integer.parseInt(commandLine.getOptionValue("ri")) : 10000;

String group = groupPrefix;
if (Boolean.parseBoolean(isSuffixEnable)) {
group = groupPrefix + "_" + (System.currentTimeMillis() % 100);
}

System.out.printf("topic: %s, threadCount %d, group: %s, suffix: %s, filterType: %s, expression: %s, msgTraceEnable: %s, aclEnable: %s%n",
topic, threadCount, group, isSuffixEnable, filterType, expression, msgTraceEnable, aclEnable);
System.out.printf("topic: %s, threadCount %d, group: %s, suffix: %s, filterType: %s, expression: %s, msgTraceEnable: %s, aclEnable: %s, reportInterval: %d%n",
topic, threadCount, group, isSuffixEnable, filterType, expression, msgTraceEnable, aclEnable, reportInterval);

final StatsBenchmarkConsumer statsBenchmarkConsumer = new StatsBenchmarkConsumer();

Expand Down Expand Up @@ -124,7 +125,7 @@ public void run() {
e.printStackTrace();
}
}
}, 10000, 10000, TimeUnit.MILLISECONDS);
}, reportInterval, reportInterval, TimeUnit.MILLISECONDS);

RPCHook rpcHook = null;
if (aclEnable) {
Expand Down Expand Up @@ -235,6 +236,10 @@ public static Options buildCommandlineOptions(final Options options) {
opt.setRequired(false);
options.addOption(opt);

opt = new Option("ri", "reportInterval", true, "The number of ms between reports, Default: 10000");
opt.setRequired(false);
options.addOption(opt);

return options;
}

Expand Down
11 changes: 8 additions & 3 deletions example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,12 +82,13 @@ public static void main(String[] args) throws MQClientException {
final boolean asyncEnable = commandLine.hasOption('y') && Boolean.parseBoolean(commandLine.getOptionValue('y'));
final int threadCount = asyncEnable ? 1 : commandLine.hasOption('w') ? Integer.parseInt(commandLine.getOptionValue('w')) : 64;
final boolean enableCompress = commandLine.hasOption('c') && Boolean.parseBoolean(commandLine.getOptionValue('c'));
final int reportInterval = commandLine.hasOption("ri") ? Integer.parseInt(commandLine.getOptionValue("ri")) : 10000;

System.out.printf("topic: %s, threadCount: %d, messageSize: %d, keyEnable: %s, propertySize: %d, tagCount: %d, " +
"traceEnable: %s, aclEnable: %s, messageQuantity: %d, delayEnable: %s, delayLevel: %s, " +
"asyncEnable: %s%n compressEnable: %s%n",
"asyncEnable: %s%n compressEnable: %s, reportInterval: %d%n",
topic, threadCount, messageSize, keyEnable, propertySize, tagCount, msgTraceEnable, aclEnable, messageNum,
delayEnable, delayLevel, asyncEnable, enableCompress);
delayEnable, delayLevel, asyncEnable, enableCompress, reportInterval);

StringBuilder sb = new StringBuilder(messageSize);
for (int i = 0; i < messageSize; i++) {
Expand Down Expand Up @@ -139,7 +140,7 @@ public void run() {
e.printStackTrace();
}
}
}, 10000, 10000, TimeUnit.MILLISECONDS);
}, reportInterval, reportInterval, TimeUnit.MILLISECONDS);

RPCHook rpcHook = null;
if (aclEnable) {
Expand Down Expand Up @@ -370,6 +371,10 @@ public static Options buildCommandlineOptions(final Options options) {
opt.setRequired(false);
options.addOption(opt);

opt = new Option("ri", "reportInterval", true, "The number of ms between reports, Default: 10000");
opt.setRequired(false);
options.addOption(opt);

return options;
}

Expand Down
11 changes: 8 additions & 3 deletions example/src/main/java/org/apache/rocketmq/example/benchmark/TransactionProducer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ public static void main(String[] args) throws MQClientException, UnsupportedEnco
config.sendInterval = commandLine.hasOption("i") ? Integer.parseInt(commandLine.getOptionValue("i")) : 0;
config.aclEnable = commandLine.hasOption('a') && Boolean.parseBoolean(commandLine.getOptionValue('a'));
config.msgTraceEnable = commandLine.hasOption('m') && Boolean.parseBoolean(commandLine.getOptionValue('m'));
config.reportInterval = commandLine.hasOption("ri") ? Integer.parseInt(commandLine.getOptionValue("ri")) : 10000;

final ExecutorService sendThreadPool = Executors.newFixedThreadPool(config.threadCount);

Expand All @@ -105,8 +106,7 @@ private void printStats() {
Snapshot begin = snapshotList.getFirst();
Snapshot end = snapshotList.getLast();

final long sendCount = (end.sendRequestSuccessCount - begin.sendRequestSuccessCount)
+ (end.sendRequestFailedCount - begin.sendRequestFailedCount);
final long sendCount = end.sendRequestSuccessCount - begin.sendRequestSuccessCount;
final long sendTps = (sendCount * 1000L) / (end.endTime - begin.endTime);
final double averageRT = (end.sendMessageTimeTotal - begin.sendMessageTimeTotal) / (double) (end.sendRequestSuccessCount - begin.sendRequestSuccessCount);

Expand All @@ -131,7 +131,7 @@ public void run() {
e.printStackTrace();
}
}
}, 10000, 10000, TimeUnit.MILLISECONDS);
}, config.reportInterval, config.reportInterval, TimeUnit.MILLISECONDS);

RPCHook rpcHook = null;
if (config.aclEnable) {
Expand Down Expand Up @@ -291,6 +291,10 @@ public static Options buildCommandlineOptions(final Options options) {
opt.setRequired(false);
options.addOption(opt);

opt = new Option("ri", "reportInterval", true, "The number of ms between reports, Default: 10000");
opt.setRequired(false);
options.addOption(opt);

return options;
}
}
Expand Down Expand Up @@ -475,6 +479,7 @@ class TxSendConfig {
int sendInterval;
boolean aclEnable;
boolean msgTraceEnable;
int reportInterval;
}

class LRUMap<K, V> extends LinkedHashMap<K, V> {
Expand Down

0 comments on commit e545e3b

Please sign in to comment.