optimize the RabbitMQ connection pool

src/DotNetCore.CAP.RabbitMQ/CAP.RabbitMQCapOptionsExtension.cs

@@ -23,7 +23,7 @@ namespace DotNetCore.CAP
             services.AddSingleton(options);
 
             services.AddSingleton<IConsumerClientFactory, RabbitMQConsumerClientFactory>();
-            services.AddSingleton<ConnectionPool>();
+            services.AddSingleton<IConnectionChannelPool, ConnectionChannelPool>();
             services.AddSingleton<IQueueExecutor, PublishQueueExecutor>();
             services.AddSingleton<IPublishExecutor, PublishQueueExecutor>();
         }

src/DotNetCore.CAP.RabbitMQ/ConnectionPool.cs → src/DotNetCore.CAP.RabbitMQ/ConnectionChannelPool.cs

@@ -2,38 +2,50 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Threading;
+using Microsoft.Extensions.Logging;
 using RabbitMQ.Client;
 
 namespace DotNetCore.CAP.RabbitMQ
 {
-    public class ConnectionPool : IConnectionPool, IDisposable
+    public class ConnectionChannelPool : IConnectionChannelPool, IDisposable
     {
         private const int DefaultPoolSize = 15;
+        private readonly Func<IConnection> _connectionActivator;
+        private readonly ILogger<ConnectionChannelPool> _logger;
+        private readonly ConcurrentQueue<IModel> _pool = new ConcurrentQueue<IModel>();
+        private IConnection _connection;
 
-        private readonly Func<IConnection> _activator;
-
-        private readonly ConcurrentQueue<IConnection> _pool = new ConcurrentQueue<IConnection>();
         private int _count;
-
         private int _maxSize;
 
-        public ConnectionPool(RabbitMQOptions options)
+        public ConnectionChannelPool(ILogger<ConnectionChannelPool> logger,
+            RabbitMQOptions options)
         {
+            _logger = logger;
             _maxSize = DefaultPoolSize;
 
-            _activator = CreateActivator(options);
+            _connectionActivator = CreateConnection(options);
         }
 
-        IConnection IConnectionPool.Rent()
+        IModel IConnectionChannelPool.Rent()
         {
             return Rent();
         }
 
-        bool IConnectionPool.Return(IConnection connection)
+        bool IConnectionChannelPool.Return(IModel connection)
         {
             return Return(connection);
         }
 
+        public IConnection GetConnection()
+        {
+            if (_connection != null && _connection.IsOpen)
+                return _connection;
+            _connection = _connectionActivator();
+            _connection.ConnectionShutdown += RabbitMQ_ConnectionShutdown;
+            return _connection;
+        }
+
         public void Dispose()
         {
             _maxSize = 0;
@@ -42,7 +54,7 @@ namespace DotNetCore.CAP.RabbitMQ
                 context.Dispose();
         }
 
-        private static Func<IConnection> CreateActivator(RabbitMQOptions options)
+        private static Func<IConnection> CreateConnection(RabbitMQOptions options)
         {
             var factory = new ConnectionFactory
             {
@@ -59,23 +71,28 @@ namespace DotNetCore.CAP.RabbitMQ
             return () => factory.CreateConnection();
         }
 
-        public virtual IConnection Rent()
+        private void RabbitMQ_ConnectionShutdown(object sender, ShutdownEventArgs e)
+        {
+            _logger.LogWarning($"RabbitMQ client connection closed! {e}");
+        }
+
+        public virtual IModel Rent()
         {
-            if (_pool.TryDequeue(out var connection))
+            if (_pool.TryDequeue(out var model))
             {
                 Interlocked.Decrement(ref _count);
 
                 Debug.Assert(_count >= 0);
 
-                return connection;
+                return model;
             }
 
-            connection = _activator();
+            model = GetConnection().CreateModel();
 
-            return connection;
+            return model;
         }
 
-        public virtual bool Return(IConnection connection)
+        public virtual bool Return(IModel connection)
         {
             if (Interlocked.Increment(ref _count) <= _maxSize)
             {

src/DotNetCore.CAP.RabbitMQ/IConnectionChannelPool.cs

@@ -0,0 +1,13 @@
+using RabbitMQ.Client;
+
+namespace DotNetCore.CAP.RabbitMQ
+{
+    public interface IConnectionChannelPool
+    {
+        IConnection GetConnection();
+
+        IModel Rent();
+
+        bool Return(IModel context);
+    }
+}

src/DotNetCore.CAP.RabbitMQ/IConnectionPool.cs

@@ -1,11 +0,0 @@
-using RabbitMQ.Client;
-
-namespace DotNetCore.CAP.RabbitMQ
-{
-    public interface IConnectionPool
-    {
-        IConnection Rent();
-
-        bool Return(IConnection context);
-    }
-}

src/DotNetCore.CAP.RabbitMQ/PublishQueueExecutor.cs

@@ -9,41 +9,34 @@ namespace DotNetCore.CAP.RabbitMQ
 {
     internal sealed class PublishQueueExecutor : BasePublishQueueExecutor
     {
-        private readonly ConnectionPool _connectionPool;
+        private readonly IConnectionChannelPool _connectionChannelPool;
         private readonly ILogger _logger;
         private readonly RabbitMQOptions _rabbitMQOptions;
 
-        public PublishQueueExecutor(
-            CapOptions options,
-            IStateChanger stateChanger,
-            ConnectionPool connectionPool,
-            RabbitMQOptions rabbitMQOptions,
-            ILogger<PublishQueueExecutor> logger)
+        public PublishQueueExecutor(ILogger<PublishQueueExecutor> logger, CapOptions options,
+            RabbitMQOptions rabbitMQOptions, IConnectionChannelPool connectionChannelPool, IStateChanger stateChanger)
             : base(options, stateChanger, logger)
         {
             _logger = logger;
-            _connectionPool = connectionPool;
+            _connectionChannelPool = connectionChannelPool;
             _rabbitMQOptions = rabbitMQOptions;
         }
 
         public override Task<OperateResult> PublishAsync(string keyName, string content)
         {
-            var connection = _connectionPool.Rent();
-
+            var channel = _connectionChannelPool.Rent();
             try
             {
-                using (var channel = connection.CreateModel())
-                {
-                    var body = Encoding.UTF8.GetBytes(content);
+                var body = Encoding.UTF8.GetBytes(content);
+
+                channel.ExchangeDeclare(_rabbitMQOptions.TopicExchangeName, RabbitMQOptions.ExchangeType, true);
+                channel.BasicPublish(_rabbitMQOptions.TopicExchangeName,
+                    keyName,
+                    null,
+                    body);
 
-                    channel.ExchangeDeclare(_rabbitMQOptions.TopicExchangeName, RabbitMQOptions.ExchangeType, true);
-                    channel.BasicPublish(_rabbitMQOptions.TopicExchangeName,
-                        keyName,
-                        null,
-                        body);
+                _logger.LogDebug($"RabbitMQ topic message [{keyName}] has been published.");
 
-                    _logger.LogDebug($"RabbitMQ topic message [{keyName}] has been published.");
-                }
                 return Task.FromResult(OperateResult.Success);
             }
             catch (Exception ex)
@@ -60,7 +53,7 @@ namespace DotNetCore.CAP.RabbitMQ
             }
             finally
             {
-                _connectionPool.Return(connection);
+                _connectionChannelPool.Return(channel);
             }
         }
     }

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs

@@ -10,7 +10,7 @@ namespace DotNetCore.CAP.RabbitMQ
 {
     internal sealed class RabbitMQConsumerClient : IConsumerClient
     {
-        private readonly ConnectionPool _connectionPool;
+        private readonly IConnectionChannelPool _connectionChannelPool;
         private readonly string _exchageName;
         private readonly string _queueName;
         private readonly RabbitMQOptions _rabbitMQOptions;
@@ -19,11 +19,11 @@ namespace DotNetCore.CAP.RabbitMQ
         private ulong _deliveryTag;
 
         public RabbitMQConsumerClient(string queueName,
-            ConnectionPool connectionPool,
+            IConnectionChannelPool connectionChannelPool,
             RabbitMQOptions options)
         {
             _queueName = queueName;
-            _connectionPool = connectionPool;
+            _connectionChannelPool = connectionChannelPool;
             _rabbitMQOptions = options;
             _exchageName = options.TopicExchangeName;
 
@@ -69,7 +69,7 @@ namespace DotNetCore.CAP.RabbitMQ
 
         private void InitClient()
         {
-            var connection = _connectionPool.Rent();
+            var connection = _connectionChannelPool.GetConnection();
 
             _channel = connection.CreateModel();
 
@@ -82,8 +82,6 @@ namespace DotNetCore.CAP.RabbitMQ
                 { "x-message-ttl", _rabbitMQOptions.QueueMessageExpires }
             };
             _channel.QueueDeclare(_queueName, true, false, false, arguments);
-
-            _connectionPool.Return(connection);
         }
 
         private void OnConsumerReceived(object sender, BasicDeliverEventArgs e)

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClientFactory.cs

@@ -2,19 +2,19 @@
 {
     internal sealed class RabbitMQConsumerClientFactory : IConsumerClientFactory
     {
-        private readonly ConnectionPool _connectionPool;
+        private readonly IConnectionChannelPool _connectionChannelPool;
         private readonly RabbitMQOptions _rabbitMQOptions;
 
 
-        public RabbitMQConsumerClientFactory(RabbitMQOptions rabbitMQOptions, ConnectionPool pool)
+        public RabbitMQConsumerClientFactory(RabbitMQOptions rabbitMQOptions, IConnectionChannelPool channelPool)
         {
             _rabbitMQOptions = rabbitMQOptions;
-            _connectionPool = pool;
+            _connectionChannelPool = channelPool;
         }
 
         public IConsumerClient Create(string groupId)
         {
-            return new RabbitMQConsumerClient(groupId, _connectionPool, _rabbitMQOptions);
+            return new RabbitMQConsumerClient(groupId, _connectionChannelPool, _rabbitMQOptions);
         }
     }
 }