From 824cc56802587bee694d808972d7da2e463a9f63 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 30 Aug 2012 10:45:01 +0200 Subject: [PATCH] Created a Consumer for non blocking IO in benchmarks --- .../{Consumer.java => ConsumerBlocking.java} | 6 +- .../dna/mqtt/bechnmark/ConsumerFuture.java | 85 +++++++++++++++++++ .../java/org/dna/mqtt/bechnmark/Main.java | 14 +-- .../java/org/dna/mqtt/bechnmark/Producer.java | 9 +- 4 files changed, 104 insertions(+), 10 deletions(-) rename benchmarking/src/main/java/org/dna/mqtt/bechnmark/{Consumer.java => ConsumerBlocking.java} (94%) create mode 100644 benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerBlocking.java similarity index 94% rename from benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java rename to benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerBlocking.java index 2c9f1a90..dcb0c214 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Consumer.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerBlocking.java @@ -13,13 +13,13 @@ import org.slf4j.LoggerFactory; * * @author andrea */ -public class Consumer implements Runnable { +public class ConsumerBlocking implements Runnable { - private static final Logger LOG = LoggerFactory.getLogger(Consumer.class); + private static final Logger LOG = LoggerFactory.getLogger(ConsumerBlocking.class); private String m_clientID; - public Consumer(String clientID) { + public ConsumerBlocking(String clientID) { m_clientID = clientID; } diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java new file mode 100644 index 00000000..1e91a40c --- /dev/null +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java @@ -0,0 +1,85 @@ +package org.dna.mqtt.bechnmark; + +import java.net.URISyntaxException; +import org.fusesource.mqtt.client.Future; +import org.fusesource.mqtt.client.FutureConnection; +import org.fusesource.mqtt.client.MQTT; +import org.fusesource.mqtt.client.Message; +import org.fusesource.mqtt.client.QoS; +import org.fusesource.mqtt.client.Topic; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class ConsumerFuture implements Runnable { + + private static final Logger LOG = LoggerFactory.getLogger(ConsumerFuture.class); + + private String m_clientID; + + public ConsumerFuture(String clientID) { + m_clientID = clientID; + } + + public void run() { + MQTT mqtt = new MQTT(); + try { +// mqtt.setHost("test.mosquitto.org", 1883); + mqtt.setHost("localhost", 1883); + } catch (URISyntaxException ex) { + LOG.error(null, ex); + return; + } + + mqtt.setClientId(m_clientID); + FutureConnection connection = mqtt.futureConnection(); + Future futConn = connection.connect(); + + try { + futConn.await(); + } catch (Exception ex) { + LOG.error("Cant't CONNECT to the server", ex); + return; + } + + + Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)}; + Future futSub = connection.subscribe(topics); + try { + byte[] qoses = futSub.await(); + LOG.info("Subscribed to topic"); + } catch (Exception ex) { + LOG.error("Cant't PUSBLISH to the server", ex); + return; + } + + Message message = null; + for (int i = 0; i < Producer.PUB_LOOP; i++) { + Future futReceive = connection.receive(); + try { + message = futReceive.await(); + } catch (Exception ex) { + LOG.error(null, ex); + return; + } + byte[] payload = message.getPayload(); + StringBuffer sb = new StringBuffer().append("Topic: ").append(message.getTopic()) + .append(", payload: ").append(new String(payload)); + LOG.info(sb.toString()); + } + + Future f4 = connection.disconnect(); + try { + LOG.info("Disconneting"); + f4.await(); + LOG.info("Disconnected"); + } catch (Exception ex) { + LOG.error("Cant't DISCONNECT to the server", ex); + } + + } + +} 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 f6dc7b68..1191d8af 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java @@ -13,17 +13,21 @@ import java.util.concurrent.Executors; */ public class Main { private static final int PUBLISHER_POOL_SIZE = 2; + private static final int NUM_CONCURRENT_PRODS = 2; public static void main(String[] args) throws InterruptedException { - ExecutorService pool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE); + ExecutorService consumerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE); + ExecutorService producerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE); - pool.submit(new Consumer("Cons1")); + consumerPool.submit(new ConsumerBlocking("Cons1")); //force wait to let the consumer be registered before the producer Thread.sleep(1000); + int len = Producer.PUB_LOOP / NUM_CONCURRENT_PRODS; + producerPool.submit(new Producer("Prod1", 0, len)); + producerPool.submit(new Producer("Prod2", len, len)); - pool.submit(new Producer("Prod1")); - - pool.shutdown(); + consumerPool.shutdown(); + producerPool.shutdown(); } } 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 a0e879dc..451a26a8 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java @@ -21,8 +21,13 @@ public class Producer implements Runnable { public static final int PUB_LOOP = 1000000; - public Producer(String clientID) { + private static int m_starIndex; + private static int m_len; + + public Producer(String clientID, int start, int len) { m_clientID = clientID; + m_starIndex = start; + m_len = len; } public void run() { @@ -45,7 +50,7 @@ public class Producer implements Runnable { } //TODO loop - for (int i = 0; i < PUB_LOOP; i++) { + for (int i = m_starIndex; i < m_starIndex + m_len; i++) { try { // LOG.info("Publishing"); String payload = "Hello world MQTT!!" + i;