Use default exchange by default

acogoluegnes · acogoluegnes · commit 2ae0c793e133 · 2017-10-23T14:00:55.000+02:00
The default exchange is now the default if possible (e.g. when the routing key isn't specified and there's only one queue to publish too). For those cases, the routing key is the queue name and there's no need to create an exchange (a "direct"-named exchange was created previously). Fixes #45
diff --git a/src/main/java/com/rabbitmq/perf/Consumer.java b/src/main/java/com/rabbitmq/perf/Consumer.java
@@ -35,7 +35,7 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
 
     private ConsumerImpl                q;
     private final Channel               channel;
-    private final String                id;
+    private final String                routingKey;
     private final List<String>          queueNames;
     private final int                   txSize;
     private final boolean               autoAck;
@@ -47,14 +47,14 @@ public class Consumer extends ProducerConsumerBase implements Runnable {
     private final Map<String, String>   consumerTagBranchMap = Collections.synchronizedMap(new HashMap<String, String>());
     private final ConsumerLatency       consumerLatency;
 
-    public Consumer(Channel channel, String id,
+    public Consumer(Channel channel, String routingKey,
                     List<String> queueNames, int txSize, boolean autoAck,
                     int multiAckEvery, Stats stats, float rateLimit, int msgLimit, int timeLimit,
                     int consumerLatencyInMicroSeconds) {
 
         this.channel       = channel;
-        this.id            = id;
-        this.queueNames    = queueNames;
+        this.routingKey    = routingKey;
+        this.queueNames    = Collections.unmodifiableList(queueNames);
         this.rateLimit     = rateLimit;
         this.txSize        = txSize;
         this.autoAck       = autoAck;
@@ -129,7 +129,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
 
                 now = System.currentTimeMillis();
 
-                stats.handleRecv(id.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
+                stats.handleRecv(routingKey.equals(envelope.getRoutingKey()) ? (nano - msgNano) : 0L);
                 if (rateLimit > 0.0f) {
                     delay(now);
                 }
@@ -158,6 +158,10 @@ public void handleCancel(String consumerTag) throws IOException {
         }
     }
 
+    public List<String> getQueueNames() {
+        return this.queueNames;
+    }
+
     private interface ConsumerLatency {
 
         void simulateLatency();
diff --git a/src/main/java/com/rabbitmq/perf/MulticastParams.java b/src/main/java/com/rabbitmq/perf/MulticastParams.java
@@ -43,19 +43,19 @@ public class MulticastParams {
     private int producerMsgCount = 0;
     private int consumerMsgCount = 0;
 
-    private String exchangeName = "direct";
+    private String exchangeName = "";
     private String exchangeType = "direct";
-    private List<String> queueNames = new ArrayList<String>();
+    private List<String> queueNames = new ArrayList<>();
     private String routingKey = null;
     private boolean randomRoutingKey = false;
 
-    private List<?> flags = new ArrayList<Object>();
+    private List<?> flags = new ArrayList<>();
 
     private int multiAckEvery = 0;
     private boolean autoAck = true;
     private boolean autoDelete = false;
 
-    private List<String> bodyFiles = new ArrayList<String>();
+    private List<String> bodyFiles = new ArrayList<>();
     private String bodyContentType = null;
 
     private boolean predeclared;
@@ -74,9 +74,9 @@ public void setExchangeName(String exchangeName) {
 
     public void setQueueNames(List<String> queueNames) {
         if(queueNames == null) {
-            this.queueNames = new ArrayList<String>();
+            this.queueNames = new ArrayList<>();
         } else {
-            this.queueNames = new ArrayList<String>(queueNames);
+            this.queueNames = new ArrayList<>(queueNames);
         }
     }
 
@@ -219,21 +219,21 @@ public boolean getRandomRoutingKey() {
 
     public void setBodyFiles(List<String> bodyFiles) {
         if (bodyFiles == null) {
-            this.bodyFiles = new ArrayList<String>();
+            this.bodyFiles = new ArrayList<>();
         } else {
-            this.bodyFiles = new ArrayList<String>(bodyFiles);
+            this.bodyFiles = new ArrayList<>(bodyFiles);
         }
     }
 
     public void setBodyContentType(String bodyContentType) {
         this.bodyContentType = bodyContentType;
     }
 
-    public Producer createProducer(Connection connection, Stats stats, String id) throws IOException {
+    public Producer createProducer(Connection connection, Stats stats, String routingKey) throws IOException {
         Channel channel = connection.createChannel();
         if (producerTxSize > 0) channel.txSelect();
         if (confirm >= 0) channel.confirmSelect();
-        if (!predeclared || !exchangeExists(connection, exchangeName)) {
+        if (!predeclared && !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         MessageBodySource messageBodySource = null;
@@ -242,7 +242,7 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
         } else {
             messageBodySource = new TimeSequenceMessageBodySource(minMsgSize);
         }
-        final Producer producer = new Producer(channel, exchangeName, id,
+        final Producer producer = new Producer(channel, exchangeName, routingKey,
                                                randomRoutingKey, flags, producerTxSize,
                                                producerRateLimit, producerMsgCount,
                                                timeLimit,
@@ -252,26 +252,24 @@ public Producer createProducer(Connection connection, Stats stats, String id) th
         return producer;
     }
 
-    public Consumer createConsumer(Connection connection, Stats stats, String id) throws IOException {
+    public Consumer createConsumer(Connection connection, Stats stats, String routingKey) throws IOException {
         Channel channel = connection.createChannel();
         if (consumerTxSize > 0) channel.txSelect();
-        List<String> generatedQueueNames = configureQueues(connection, id);
+        List<String> generatedQueueNames = configureQueues(connection, routingKey);
         if (consumerPrefetch > 0) channel.basicQos(consumerPrefetch);
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
-        return new Consumer(channel, id, generatedQueueNames,
+        return new Consumer(channel, routingKey, generatedQueueNames,
                                          consumerTxSize, autoAck, multiAckEvery,
                                          stats, consumerRateLimit, consumerMsgCount, timeLimit, consumerLatencyInMicroseconds);
     }
 
     public boolean shouldConfigureQueues() {
-        // don't declare any queues when --predeclared is passed,
-        // otherwise unwanted server-named queues without consumers will pile up. MK.
         return consumerCount == 0 && !(queueNames.size() == 0);
     }
 
-    public List<String> configureQueues(Connection connection, String id) throws IOException {
+    public List<String> configureQueues(Connection connection, String routingKey) throws IOException {
         Channel channel = connection.createChannel();
-        if (!predeclared || !exchangeExists(connection, exchangeName)) {
+        if (!predeclared && !exchangeExists(connection, exchangeName)) {
             channel.exchangeDeclare(exchangeName, exchangeType);
         }
         // To ensure we get at-least 1 default queue:
@@ -295,7 +293,7 @@ public List<String> configureQueues(Connection connection, String id) throws IOE
             // skipping binding to default exchange,
             // as it's not possible to explicitly bind to it.
             if (!"".equals(exchangeName) && !"amq.default".equals(exchangeName)) {
-                channel.queueBind(qName, exchangeName, id);
+                channel.queueBind(qName, exchangeName, routingKey);
             }
         }
         channel.abort();
@@ -308,20 +306,12 @@ private static boolean exchangeExists(Connection connection, final String exchan
             // NB: default exchange always exists
             return true;
         } else {
-            return exists(connection, new Checker() {
-                public void check(Channel ch) throws IOException {
-                    ch.exchangeDeclarePassive(exchangeName);
-                }
-            });
+            return exists(connection, ch -> ch.exchangeDeclarePassive(exchangeName));
         }
     }
 
     private static boolean queueExists(Connection connection, final String queueName) throws IOException {
-        return queueName != null && exists(connection, new Checker() {
-            public void check(Channel ch) throws IOException {
-                ch.queueDeclarePassive(queueName);
-            }
-        });
+        return queueName != null && exists(connection, ch -> ch.queueDeclarePassive(queueName));
     }
 
     private interface Checker {
diff --git a/src/main/java/com/rabbitmq/perf/MulticastSet.java b/src/main/java/com/rabbitmq/perf/MulticastSet.java
@@ -22,13 +22,15 @@
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Random;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 
 public class MulticastSet {
-    private final String id;
+    private final String routingKey;
     private final Stats stats;
     private final ConnectionFactory factory;
     private final MulticastParams params;
@@ -40,9 +42,9 @@ public class MulticastSet {
     public MulticastSet(Stats stats, ConnectionFactory factory,
         MulticastParams params, List<String> uris) {
         if (params.getRoutingKey() == null) {
-            this.id = UUID.randomUUID().toString();
+            this.routingKey = UUID.randomUUID().toString();
         } else {
-            this.id = params.getRoutingKey();
+            this.routingKey = params.getRoutingKey();
         }
         this.stats = stats;
         this.factory = factory;
@@ -54,9 +56,9 @@ public MulticastSet(Stats stats, ConnectionFactory factory,
     public MulticastSet(Stats stats, ConnectionFactory factory,
         MulticastParams params, String testID, List<String> uris) {
         if (params.getRoutingKey() == null) {
-            this.id = UUID.randomUUID().toString();
+            this.routingKey = UUID.randomUUID().toString();
         } else {
-            this.id = params.getRoutingKey();
+            this.routingKey = params.getRoutingKey();
         }
         this.stats = stats;
         this.factory = factory;
@@ -71,6 +73,7 @@ public void run() throws IOException, InterruptedException, TimeoutException, No
 
     public void run(boolean announceStartup)
         throws IOException, InterruptedException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
+        Set<String> queueNames = new HashSet<>();
         Thread[] consumerThreads = new Thread[params.getConsumerThreadCount()];
         Connection[] consumerConnections = new Connection[params.getConsumerCount()];
         for (int i = 0; i < consumerConnections.length; i++) {
@@ -84,18 +87,30 @@ public void run(boolean announceStartup)
                 if (announceStartup) {
                     System.out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
                 }
-                Thread t = new Thread(params.createConsumer(conn, stats, id));
+                Consumer consumer = params.createConsumer(conn, stats, routingKey);
+                queueNames.addAll(consumer.getQueueNames());
+                Thread t = new Thread(consumer);
                 consumerThreads[(i * params.getConsumerChannelCount()) + j] = t;
             }
         }
 
         if (params.shouldConfigureQueues()) {
             setUri();
             Connection conn = factory.newConnection();
-            params.configureQueues(conn, id);
+            List<String> configuredQueues = params.configureQueues(conn, routingKey);
+            queueNames.addAll(configuredQueues);
             conn.close();
         }
 
+        String producersRoutingKey;
+        // if one queue, and no routing key, the routing key
+        // must be the queue name and we should use the default exchange
+        if (queueNames.size() == 1 && params.getRoutingKey() == null) {
+            producersRoutingKey = queueNames.iterator().next();
+        } else {
+            producersRoutingKey = this.routingKey;
+        }
+
         Thread[] producerThreads = new Thread[params.getProducerThreadCount()];
         Connection[] producerConnections = new Connection[params.getProducerCount()];
         for (int i = 0; i < producerConnections.length; i++) {
@@ -109,7 +124,7 @@ public void run(boolean announceStartup)
                 if (announceStartup) {
                     System.out.println("id: " + testID + ", starting producer #" + i + ", channel #" + j);
                 }
-                Thread t = new Thread(params.createProducer(conn, stats, id));
+                Thread t = new Thread(params.createProducer(conn, stats, producersRoutingKey));
                 producerThreads[(i * params.getProducerChannelCount()) + j] = t;
             }
         }
diff --git a/src/main/java/com/rabbitmq/perf/PerfTest.java b/src/main/java/com/rabbitmq/perf/PerfTest.java
@@ -59,7 +59,7 @@ public static void main(String[] args) {
                 getInstance().getTime());
             testID                   = strArg(cmd, 'd', "test-"+testID);
             String exchangeType      = strArg(cmd, 't', "direct");
-            String exchangeName      = getExchangeName(cmd, exchangeType);
+            String exchangeName      = getExchangeName(cmd, "");
             String queueNames        = strArg(cmd, 'u', null);
             String routingKey        = strArg(cmd, 'k', null);
             boolean randomRoutingKey = cmd.hasOption('K');
