anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved statistics information gathering

Browse Source
This commit is contained in:
selva.andre
2012-08-30 12:39:40 +02:00
parent 6888d37fb6
commit 1b683328e0
3 changed files with 67 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
+32 -3
View File
@@ -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<Message> 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<Void> 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.
}
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java
+1 -1
View File
@@ -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);
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java
+34 -5
View File
@@ -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.
}
}
}