From 1b683328e037890405443d2341e98fa84a405ce4 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 30 Aug 2012 12:39:40 +0200 Subject: [PATCH] Improved statistics information gathering --- .../dna/mqtt/bechnmark/ConsumerFuture.java | 35 +++++++++++++++-- .../java/org/dna/mqtt/bechnmark/Main.java | 2 +- .../java/org/dna/mqtt/bechnmark/Producer.java | 39 ++++++++++++++++--- 3 files changed, 67 insertions(+), 9 deletions(-) diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java index 1e91a40c..41c8c179 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java @@ -1,6 +1,10 @@ package org.dna.mqtt.bechnmark; +import java.io.*; import java.net.URISyntaxException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + import org.fusesource.mqtt.client.Future; import org.fusesource.mqtt.client.FutureConnection; import org.fusesource.mqtt.client.MQTT; @@ -19,9 +23,14 @@ public class ConsumerFuture implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(ConsumerFuture.class); private String m_clientID; + private static final String BENCHMARK_FILE = "consumer_bechmark.txt"; + private PrintWriter m_benchMarkOut; + private ByteArrayOutputStream m_baos = new ByteArrayOutputStream(1024 * 1024); public ConsumerFuture(String clientID) { m_clientID = clientID; + m_benchMarkOut = new PrintWriter(new BufferedWriter(new OutputStreamWriter(m_baos))); + m_benchMarkOut.println("msg ID, ns"); } public void run() { @@ -55,12 +64,21 @@ public class ConsumerFuture implements Runnable { LOG.error("Cant't PUSBLISH to the server", ex); return; } - + + Pattern p = Pattern.compile(".*!!(\\d+)"); Message message = null; + long startMillis = System.currentTimeMillis(); for (int i = 0; i < Producer.PUB_LOOP; i++) { Future futReceive = connection.receive(); try { message = futReceive.await(); + String content = new String(message.getPayload()); + + //metrics part + Matcher matcher = p.matcher(content); + matcher.matches(); + String numMsg = matcher.group(1); + m_benchMarkOut.println(String.format("%s, %d", numMsg, System.nanoTime())); } catch (Exception ex) { LOG.error(null, ex); return; @@ -68,8 +86,11 @@ public class ConsumerFuture implements Runnable { byte[] payload = message.getPayload(); StringBuffer sb = new StringBuffer().append("Topic: ").append(message.getTopic()) .append(", payload: ").append(new String(payload)); - LOG.info(sb.toString()); + LOG.debug(sb.toString()); } + + long time = System.currentTimeMillis() - startMillis; + LOG.info(String.format("Consumer %s received %d messages in %d ms", Thread.currentThread().getName(), Producer.PUB_LOOP, time)); Future f4 = connection.disconnect(); try { @@ -79,7 +100,15 @@ public class ConsumerFuture implements Runnable { } catch (Exception ex) { LOG.error("Cant't DISCONNECT to the server", ex); } - + + m_benchMarkOut.close(); + + try { + FileWriter fw = new FileWriter(BENCHMARK_FILE); + fw.write(m_baos.toString()); + } catch (IOException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } } } diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java index 1191d8af..197c6c71 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java @@ -19,7 +19,7 @@ public class Main { ExecutorService consumerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE); ExecutorService producerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE); - consumerPool.submit(new ConsumerBlocking("Cons1")); + consumerPool.submit(new ConsumerFuture("Cons1")); //force wait to let the consumer be registered before the producer Thread.sleep(1000); diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java index 451a26a8..943a9ba1 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java @@ -1,5 +1,6 @@ package org.dna.mqtt.bechnmark; +import java.io.*; import java.net.URISyntaxException; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; @@ -16,21 +17,28 @@ import org.slf4j.LoggerFactory; public class Producer implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(Producer.class); - + private String m_clientID; - public static final int PUB_LOOP = 1000000; + public static final int PUB_LOOP = 100000; private static int m_starIndex; private static int m_len; - + private long m_startMillis; + private static final String BENCHMARK_FILE = "producer_bechmark.txt"; + private PrintWriter m_benchMarkOut; + private ByteArrayOutputStream m_baos = new ByteArrayOutputStream(1024 * 1024); + public Producer(String clientID, int start, int len) { m_clientID = clientID; m_starIndex = start; m_len = len; + m_benchMarkOut = new PrintWriter(new BufferedWriter(new OutputStreamWriter(m_baos))); + m_benchMarkOut.println("msg ID, ns"); } public void run() { + m_startMillis = System.currentTimeMillis(); MQTT mqtt = new MQTT(); try { // mqtt.setHost("test.mosquitto.org", 1883); @@ -48,18 +56,27 @@ public class Producer implements Runnable { LOG.error("Cant't CONNECT to the server", ex); return; } - - //TODO loop + + long time = System.currentTimeMillis() - m_startMillis; + LOG.info(String.format("Producer %s connected in %d ms", Thread.currentThread().getName(), time)); + + m_startMillis = System.currentTimeMillis(); for (int i = m_starIndex; i < m_starIndex + m_len; i++) { try { // LOG.info("Publishing"); String payload = "Hello world MQTT!!" + i; connection.publish("/topic", payload.getBytes(), QoS.AT_MOST_ONCE, false); + m_benchMarkOut.println(String.format("%d, %d", i, System.nanoTime())); } catch (Exception ex) { LOG.error("Cant't PUBLISH to the server", ex); return; } } + + time = System.currentTimeMillis() - m_startMillis; + LOG.info(String.format("Producer %s published %d messages in %d ms", Thread.currentThread().getName(), m_len, time)); + + m_startMillis = System.currentTimeMillis(); try { LOG.info("Disconneting"); connection.disconnect(); @@ -67,6 +84,18 @@ public class Producer implements Runnable { } catch (Exception ex) { LOG.error("Cant't DISCONNECT to the server", ex); } + + time = System.currentTimeMillis() - m_startMillis; + LOG.info(String.format("Producer %s disconnected in %d ms", Thread.currentThread().getName(), time)); + + m_benchMarkOut.close(); + + try { + FileWriter fw = new FileWriter(BENCHMARK_FILE); + fw.write(m_baos.toString()); + } catch (IOException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } } }