in example/src/main/java/org/apache/rocketmq/example/benchmark/Consumer.java [53:193]
public static void main(String[] args) throws MQClientException, IOException {
System.setProperty(RemotingCommand.SERIALIZE_TYPE_PROPERTY, SerializeType.ROCKETMQ.name());
Options options = ServerUtil.buildCommandlineOptions(new Options());
CommandLine commandLine = ServerUtil.parseCmdLine("benchmarkConsumer", args, buildCommandlineOptions(options), new DefaultParser());
if (null == commandLine) {
System.exit(-1);
}
final String topic = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : "BenchmarkTest";
final int threadCount = commandLine.hasOption('w') ? Integer.parseInt(commandLine.getOptionValue('w')) : 20;
final String groupPrefix = commandLine.hasOption('g') ? commandLine.getOptionValue('g').trim() : "benchmark_consumer";
final String isSuffixEnable = commandLine.hasOption('p') ? commandLine.getOptionValue('p').trim() : "false";
final String filterType = commandLine.hasOption('f') ? commandLine.getOptionValue('f').trim() : null;
final String expression = commandLine.hasOption('e') ? commandLine.getOptionValue('e').trim() : null;
final double failRate = commandLine.hasOption('r') ? Double.parseDouble(commandLine.getOptionValue('r').trim()) : 0.0;
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, reportInterval: %d%n",
topic, threadCount, group, isSuffixEnable, filterType, expression, msgTraceEnable, aclEnable, reportInterval);
final StatsBenchmarkConsumer statsBenchmarkConsumer = new StatsBenchmarkConsumer();
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("BenchmarkTimerThread-%d").daemon(true).build());
final LinkedList<Long[]> snapshotList = new LinkedList<>();
executorService.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
snapshotList.addLast(statsBenchmarkConsumer.createSnapshot());
if (snapshotList.size() > 10) {
snapshotList.removeFirst();
}
}
}, 1000, 1000, TimeUnit.MILLISECONDS);
executorService.scheduleAtFixedRate(new TimerTask() {
private void printStats() {
if (snapshotList.size() >= 10) {
Long[] begin = snapshotList.getFirst();
Long[] end = snapshotList.getLast();
final long consumeTps =
(long) (((end[1] - begin[1]) / (double) (end[0] - begin[0])) * 1000L);
final double averageB2CRT = (end[2] - begin[2]) / (double) (end[1] - begin[1]);
final double averageS2CRT = (end[3] - begin[3]) / (double) (end[1] - begin[1]);
final long failCount = end[4] - begin[4];
final long b2cMax = statsBenchmarkConsumer.getBorn2ConsumerMaxRT().get();
final long s2cMax = statsBenchmarkConsumer.getStore2ConsumerMaxRT().get();
statsBenchmarkConsumer.getBorn2ConsumerMaxRT().set(0);
statsBenchmarkConsumer.getStore2ConsumerMaxRT().set(0);
System.out.printf("Current Time: %s | Consume TPS: %d | AVG(B2C) RT(ms): %7.3f | AVG(S2C) RT(ms): %7.3f | MAX(B2C) RT(ms): %d | MAX(S2C) RT(ms): %d | Consume Fail: %d%n",
UtilAll.timeMillisToHumanString2(System.currentTimeMillis()), consumeTps, averageB2CRT, averageS2CRT, b2cMax, s2cMax, failCount);
}
}
@Override
public void run() {
try {
this.printStats();
} catch (Exception e) {
e.printStackTrace();
}
}
}, reportInterval, reportInterval, TimeUnit.MILLISECONDS);
RPCHook rpcHook = null;
if (aclEnable) {
String ak = commandLine.hasOption("ak") ? String.valueOf(commandLine.getOptionValue("ak")) : AclClient.ACL_ACCESS_KEY;
String sk = commandLine.hasOption("sk") ? String.valueOf(commandLine.getOptionValue("sk")) : AclClient.ACL_SECRET_KEY;
rpcHook = AclClient.getAclRPCHook(ak, sk);
}
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group, rpcHook, new AllocateMessageQueueAveragely(), msgTraceEnable, null);
if (commandLine.hasOption('n')) {
String ns = commandLine.getOptionValue('n');
consumer.setNamesrvAddr(ns);
}
consumer.setConsumeThreadMin(threadCount);
consumer.setConsumeThreadMax(threadCount);
consumer.setInstanceName(Long.toString(System.currentTimeMillis()));
consumer.setClientRebalance(clientRebalanceEnable);
if (filterType == null || expression == null) {
consumer.subscribe(topic, "*");
} else {
if (ExpressionType.TAG.equals(filterType)) {
String expr = MixAll.file2String(expression);
System.out.printf("Expression: %s%n", expr);
consumer.subscribe(topic, MessageSelector.byTag(expr));
} else if (ExpressionType.SQL92.equals(filterType)) {
String expr = MixAll.file2String(expression);
System.out.printf("Expression: %s%n", expr);
consumer.subscribe(topic, MessageSelector.bySql(expr));
} else {
throw new IllegalArgumentException("Not support filter type! " + filterType);
}
}
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
MessageExt msg = msgs.get(0);
long now = System.currentTimeMillis();
statsBenchmarkConsumer.getReceiveMessageTotalCount().increment();
long born2ConsumerRT = now - msg.getBornTimestamp();
statsBenchmarkConsumer.getBorn2ConsumerTotalRT().add(born2ConsumerRT);
long store2ConsumerRT = now - msg.getStoreTimestamp();
statsBenchmarkConsumer.getStore2ConsumerTotalRT().add(store2ConsumerRT);
compareAndSetMax(statsBenchmarkConsumer.getBorn2ConsumerMaxRT(), born2ConsumerRT);
compareAndSetMax(statsBenchmarkConsumer.getStore2ConsumerMaxRT(), store2ConsumerRT);
if (ThreadLocalRandom.current().nextDouble() < failRate) {
statsBenchmarkConsumer.getFailCount().increment();
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
} else {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}