final boolean keyEnable = args.length >= 3 ? Boolean.parseBoolean(args[2]) : false;
System.out
.printf("threadCount %d messageSize %d keyEnable %s\n", threadCount, messageSize, keyEnable);
final Message msg = buildMessage(messageSize);
final ExecutorService sendThreadPool = Executors.newFixedThreadPool(threadCount);
final StatsBenchmarkProducer statsBenchmark = new StatsBenchmarkProducer();
final Timer timer = new Timer("BenchmarkTimerThread", true);
final LinkedList<Long[]> snapshotList = new LinkedList<Long[]>();
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
snapshotList.addLast(statsBenchmark.createSnapshot());
if (snapshotList.size() > 10) {
snapshotList.removeFirst();
}
}
}, 1000, 1000);
timer.scheduleAtFixedRate(new TimerTask() {
private void printStats() {
if (snapshotList.size() >= 10) {
Long[] begin = snapshotList.getFirst();
Long[] end = snapshotList.getLast();
final long sendTps =
(long) (((end[3] - begin[3]) / (double) (end[0] - begin[0])) * 1000L);
final double averageRT = ((end[5] - begin[5]) / (double) (end[3] - begin[3]));
System.out.printf(
"Send TPS: %d Max RT: %d Average RT: %7.3f Send Failed: %d Response Failed: %d\n"//
, sendTps//
, statsBenchmark.getSendMessageMaxRT().get()//
, averageRT//
, end[2]//
, end[4]//
);
}
}
@Override
public void run() {
try {
this.printStats();
}
catch (Exception e) {
e.printStackTrace();
}
}
}, 10000, 10000);
final DefaultMQProducer producer = new DefaultMQProducer("benchmark_producer");
producer.setInstanceName(Long.toString(System.currentTimeMillis()));
producer.setCompressMsgBodyOverHowmuch(Integer.MAX_VALUE);
producer.start();
for (int i = 0; i < threadCount; i++) {
sendThreadPool.execute(new Runnable() {
@Override
public void run() {
while (true) {
try {
final long beginTimestamp = System.currentTimeMillis();
if (keyEnable) {
msg.setKeys(String.valueOf(beginTimestamp / 1000));
}
producer.send(msg);
statsBenchmark.getSendRequestSuccessCount().incrementAndGet();
statsBenchmark.getReceiveResponseSuccessCount().incrementAndGet();
final long currentRT = System.currentTimeMillis() - beginTimestamp;