Use 1-thread executor per connection to dispatch messages to consumers

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -73,6 +73,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final String name;
   private final Lock instanceLock = new ReentrantLock();
   private final boolean filterExpressionsSupported;
+  private volatile ExecutorService dispatchingExecutorService;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
     super(builder.listeners());
@@ -547,22 +548,21 @@ Session nativeSession(boolean check) {
     if (check) {
       checkOpen();
     }
+
     Session result = this.nativeSession;
-    if (result == null) {
-      this.instanceLock.lock();
-      try {
-        result = this.nativeSession;
-        if (result == null) {
-          if (check) {
-            checkOpen();
-          }
-          this.nativeSession = result = this.openSession(this.nativeConnection);
-        }
-      } finally {
-        this.instanceLock.unlock();
+    if (result != null) {
+      return result;
+    }
+
+    this.instanceLock.lock();
+    try {
+      if (this.nativeSession == null) {
+        this.nativeSession = this.openSession(this.nativeConnection);
       }
+      return this.nativeSession;
+    } finally {
+      this.instanceLock.unlock();
     }
-    return result;
   }
 
   private Session openSession(org.apache.qpid.protonj2.client.Connection connection) {
@@ -585,6 +585,27 @@ ScheduledExecutorService scheduledExecutorService() {
     return this.environment.scheduledExecutorService();
   }
 
+  ExecutorService dispatchingExecutorService() {
+    checkOpen();
+
+    ExecutorService result = this.dispatchingExecutorService;
+    if (result != null) {
+      return result;
+    }
+
+    this.instanceLock.lock();
+    try {
+      if (this.dispatchingExecutorService == null) {
+        this.dispatchingExecutorService =
+            Executors.newSingleThreadExecutor(
+                Utils.threadFactory("dispatching-" + this.name + "-"));
+      }
+      return this.dispatchingExecutorService;
+    } finally {
+      this.instanceLock.unlock();
+    }
+  }
+
   Clock clock() {
     return this.environment.clock();
   }
@@ -714,6 +735,14 @@ private void close(Throwable cause) {
       for (AmqpConsumer consumer : this.consumers) {
         consumer.close();
       }
+      try {
+        this.dispatchingExecutorService.shutdownNow();
+      } catch (Exception e) {
+        LOGGER.info(
+            "Error while shutting down dispatching executor service for connection '{}': {}",
+            this.name(),
+            e.getMessage());
+      }
       try {
         org.apache.qpid.protonj2.client.Connection nc = this.nativeConnection;
         if (nc != null) {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -69,7 +69,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private final SessionHandler sessionHandler;
   private final AtomicLong unsettledMessageCount = new AtomicLong(0);
   private final Runnable replenishCreditOperation = this::replenishCreditIfNeeded;
-  private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+  private final ExecutorService dispatchingExecutorService;
   private final java.util.function.Consumer<Delivery> nativeHandler;
   private final java.util.function.Consumer<ClientException> nativeReceiverCloseHandler;
   // native receiver internal state, accessed only in the native executor/scheduler
@@ -99,10 +99,11 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
     this.connection = builder.connection();
     this.sessionHandler = this.connection.createSessionHandler();
 
+    this.dispatchingExecutorService = connection.dispatchingExecutorService();
     this.nativeHandler = createNativeHandler(messageHandler);
     this.nativeReceiverCloseHandler =
         e ->
-            this.executorService.submit(
+            this.dispatchingExecutorService.submit(
                 () -> {
                   // get result to make spotbugs happy
                   boolean ignored = maybeCloseConsumerOnException(this, e);
@@ -228,7 +229,7 @@ private java.util.function.Consumer<Delivery> createNativeHandler(MessageHandler
     return delivery -> {
       this.unsettledMessageCount.incrementAndGet();
       this.metricsCollector.consume();
-      this.executorService.submit(
+      this.dispatchingExecutorService.submit(
           () -> {
             AmqpMessage message;
             try {
@@ -330,11 +331,6 @@ private void close(Throwable cause) {
     if (this.closed.compareAndSet(false, true)) {
       this.state(CLOSING, cause);
       this.connection.removeConsumer(this);
-      try {
-        this.executorService.shutdownNow();
-      } catch (Exception e) {
-        LOGGER.warn("Error while closing consumer executor service");
-      }
       try {
         this.nativeReceiver.close();
         this.sessionHandler.close();

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -82,7 +82,6 @@ class AmqpManagement implements Management {
   private final TopologyListener topologyListener;
   private final Supplier<String> nameSupplier;
   private final AtomicReference<State> state = new AtomicReference<>(CREATED);
-  //  private final AtomicBoolean initializing = new AtomicBoolean(false);
   private volatile boolean initializing = false;
   private final Lock initializationLock = new ReentrantLock();
   private final Duration receiveLoopIdleTimeout;
@@ -208,7 +207,7 @@ void init() {
       if (!this.initializing) {
         try {
           initializationLock.lock();
-          if (!this.initializing) {
+          if (!this.initializing && this.state() != OPEN) {
             this.initializing = true;
             LOGGER.debug("Initializing management ({}).", this);
             this.state(UNAVAILABLE);

src/test/java/com/rabbitmq/client/amqp/perf/AmqpPerfTest.java

@@ -17,10 +17,12 @@
 // [email protected].
 package com.rabbitmq.client.amqp.perf;
 
+import static com.rabbitmq.client.amqp.Management.ExchangeType.DIRECT;
 import static com.rabbitmq.client.amqp.Management.QueueType.QUORUM;
 import static com.rabbitmq.client.amqp.impl.TestUtils.environmentBuilder;
 
 import com.rabbitmq.client.amqp.*;
+import com.rabbitmq.client.amqp.impl.TestUtils;
 import com.rabbitmq.client.amqp.metrics.MetricsCollector;
 import com.rabbitmq.client.amqp.metrics.MicrometerMetricsCollector;
 import com.sun.net.httpserver.HttpServer;
@@ -33,7 +35,6 @@
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,9 +56,8 @@ public static void main(String[] args) throws IOException {
     PrintWriter out = new PrintWriter(System.out, true);
     PerformanceMetrics metrics = new PerformanceMetrics(registry, executorService, out);
 
-    //    String e = TestUtils.name(AmqpPerfTest.class, "main");
-    //    String q = TestUtils.name(AmqpPerfTest.class, "main");
-    String q = "qq";
+    String e = TestUtils.name(AmqpPerfTest.class, "main");
+    String q = TestUtils.name(AmqpPerfTest.class, "main");
     String rk = "foo";
     Environment environment = environmentBuilder().metricsCollector(collector).build();
     Connection connection = environment.connectionBuilder().build();
@@ -76,19 +76,18 @@ public static void main(String[] args) throws IOException {
             metrics.close();
             executorService.shutdownNow();
             shutdownLatch.countDown();
-            //            management.queueDeletion().delete(q);
-            //            management.exchangeDeletion().delete(e);
+            management.queueDeletion().delete(q);
+            management.exchangeDeletion().delete(e);
             management.close();
           }
         };
 
     Runtime.getRuntime().addShutdownHook(new Thread(shutdownSequence::run));
     try {
-      //      management.exchange().name(e).type(DIRECT).declare();
+      management.exchange().name(e).type(DIRECT).declare();
       management.queue().name(q).type(QUORUM).declare();
-      //      management.binding().sourceExchange(e).destinationQueue(q).key(rk).bind();
+      management.binding().sourceExchange(e).destinationQueue(q).key(rk).bind();
 
-      AtomicInteger count = new AtomicInteger(0);
       connection
           .consumerBuilder()
           .listeners(
@@ -102,17 +101,13 @@ public static void main(String[] args) throws IOException {
           .messageHandler(
               (context, message) -> {
                 context.accept();
-                if (count.incrementAndGet() == 1_000_000) {
-                  shutdownLatch.countDown();
+                try {
+                  long time = readLong(message.body());
+                  metrics.latency(System.currentTimeMillis() - time, TimeUnit.MILLISECONDS);
+                } catch (Exception ex) {
+                  // not able to read the body, maybe not a message from the
+                  // tool
                 }
-                //                try {
-                //                  long time = readLong(message.body());
-                //                  metrics.latency(System.currentTimeMillis() - time,
-                // TimeUnit.MILLISECONDS);
-                //                } catch (Exception ex) {
-                //                  // not able to read the body, maybe not a message from the
-                //                  // tool
-                //                }
               })
           .build();
 
@@ -122,13 +117,12 @@ public static void main(String[] args) throws IOException {
             Publisher publisher =
                 connection
                     .publisherBuilder()
-                    .queue(q)
-                    //                    .exchange(e)
-                    //                    .key(rk)
+                    .exchange(e)
+                    .key(rk)
                     .listeners(
                         context -> {
                           if (context.currentState() == Resource.State.OPEN) {
-                            //                            shouldPublish.set(true);
+                            shouldPublish.set(true);
                           } else {
                             if (context.currentState() == Resource.State.RECOVERING) {
                               LOGGER.info("Publisher is recovering...");
@@ -146,9 +140,6 @@ public static void main(String[] args) throws IOException {
                   } catch (Exception ex) {
                     // not able to read the body, should not happen
                   }
-                  if (count.incrementAndGet() == 1_000_000) {
-                    shutdownLatch.countDown();
-                  }
                 };
             int msgSize = 10;
             while (!Thread.currentThread().isInterrupted()) {