连接池实现连接和通道的复用。但还没有实现消费者绑定队列获取消息的功能。

ROMYIM · ROMYIM · commit a918dca0d00c · 2020-02-19T09:52:53.000+08:00
diff --git a/Abtraction/AbractPool.cs b/Abtraction/AbractPool.cs
@@ -0,0 +1,38 @@
+﻿using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Text;
+
+namespace RabbitMq.Abtraction
+{
+    public abstract class AbractPool<T> : IPool<T> where T : IPoolable
+    {
+        protected readonly ConcurrentQueue<T> _pool = new ConcurrentQueue<T>();
+
+        protected int _maxSize;
+
+        protected int _createdCount;
+
+        protected bool _disposed;
+
+        protected ILogger _logger;
+
+        protected virtual int Count { get => _pool.Count; }
+
+        public virtual void Dispose()
+        {
+            if (!_pool.IsEmpty)
+            {
+                while (_pool.TryDequeue(out T item))
+                {
+                    item?.Dispose();
+                }
+            }
+        }
+
+        public abstract T Rent();
+
+        public abstract void Return(T t);
+    }
+}
diff --git a/Abtraction/IPool.cs b/Abtraction/IPool.cs
@@ -0,0 +1,13 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace RabbitMq.Abtraction
+{
+    public interface IPool<T> : IDisposable where T : IDisposable
+    {
+        T Rent();
+
+        void Return(T t);
+    }
+}
diff --git a/Abtraction/IPoolable.cs b/Abtraction/IPoolable.cs
@@ -0,0 +1,11 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace RabbitMq.Abtraction
+{
+    public interface IPoolable : IDisposable
+    {
+        void SetPool(object pool);
+    }
+}
diff --git a/Events/ConfigEventArgs.cs b/Events/ConfigEventArgs.cs
@@ -0,0 +1,11 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace RabbitMq.Events
+{
+    public class ConfigEventArgs : EventArgs
+    {
+        public RabbitMqOptions Options { get; set; }
+    }
+}
diff --git a/Extensions/ConnectionExtension.cs b/Extensions/ConnectionExtension.cs
@@ -0,0 +1,25 @@
+﻿using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace RabbitMq.Extensions
+{
+    public static class ConnectionExtension
+    {
+        public static Task SendMessageAsync(this RabbitMqConnection connection, string exchangeName, string routeKey, byte[] message)
+        {
+            return Task.Run(() =>
+            {
+                var channel = connection.Rent();
+                if (!channel.SendMessage(exchangeName, routeKey, message))
+                {
+                    connection.Logger.LogError("send message failed");
+                }
+                connection.Return(channel);
+            });
+
+        }
+    }
+}
diff --git a/Extensions/DependencyInjectionExtension.cs b/Extensions/DependencyInjectionExtension.cs
@@ -0,0 +1,24 @@
+﻿using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace RabbitMq.DependencyInjection
+{
+    public static class DependencyInjectionExtension
+    {
+        public static IServiceCollection AddRabbitMqHelper(this IServiceCollection services, Action<RabbitMqOptions> build, IConfiguration configuration)
+        {
+            var options = new RabbitMqOptions();
+            options.Configuration = configuration;
+            build(options);
+
+            services.AddSingleton(options);
+            services.AddSingleton<RabbitMqConnectionTable>();
+            services.AddSingleton<RabbitMqHelper>();
+
+            return services;
+        }
+    }
+}
diff --git a/RabbitMq.csproj b/RabbitMq.csproj
@@ -0,0 +1,15 @@
+﻿<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Core\Core.csproj" />
+  </ItemGroup>
+
+</Project>
diff --git a/RabbitMqChannel.cs b/RabbitMqChannel.cs
@@ -0,0 +1,122 @@
+﻿using Core.Extensions;
+using Microsoft.Extensions.Logging;
+using RabbitMq.Abtraction;
+using RabbitMq.Events;
+using RabbitMQ.Client;
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace RabbitMq
+{
+    public class RabbitMqChannel : IDisposable, IPoolable
+    {
+        private RabbitMqConnection _connection;
+
+        private IModel _channel;
+
+        private bool _disposed;
+
+        private readonly RabbitMqOptions _options;
+
+        private readonly ILogger _logger;
+
+        public IModel Channel { get => _channel; }
+        /// <summary>
+        /// 通道配置的事件回调。可以由用户自己配置再进行与mq服务的消息传递。
+        /// 调用<see cref="HandleChannelConfig(RabbitMqOptions)"/>使配置生效。
+        /// </summary>
+        public EventHandler<ConfigEventArgs> ChannelConfiged { get; set; }
+
+        public bool Configed { get; private set; }
+
+        public RabbitMqChannel(RabbitMqConnection connection)
+        {
+            Check.NotNull(connection, nameof(connection));
+
+            _logger = RabbitMqConnectionTable.LoggerFactory.CreateLogger(GetType());
+            _channel = connection.Connection?.CreateModel();
+            _options = connection.Options;
+            SetPool(connection);
+        }
+
+        public Task<bool> SendAndDeleteAsync(Exchange exchange, Queue queue, byte[] body, IBasicProperties properties = default, bool mandatory = default)
+        {
+            if (Channel.IsOpen)
+            {
+                Channel.ExchangeDeclare(exchange.Name, exchange.Type, exchange.Durable, exchange.AutoDelete, null);
+                Channel.QueueDeclare(queue.Name, queue.Durable, queue.Exclusive, queue.AutoDelete, null);
+                Channel.QueueBind(queue.Name, exchange.Name, exchange.RouteKey, null);
+                Channel.BasicPublish(exchange.Name, exchange.RouteKey, mandatory, properties, body);
+                return Task.Run(() =>
+                {
+                    
+                    var result = Channel.WaitForConfirms();
+                    Channel.QueueDelete(queue.Name, true, true);
+                    Channel.ExchangeDelete(exchange.Name, true);
+                    return result;
+                });
+            }
+            else
+            {
+                _logger.LogInformation("class-id:{0}\nreply-code:{1}\nreply-text:{2}", Channel.CloseReason.ClassId, Channel.CloseReason.ReplyCode, Channel.CloseReason.ReplyText);
+                return Task.FromResult(false);
+            }
+        }
+
+        public bool SendMessage(string exchangeName, string routeKey, byte[] body,IBasicProperties properties = default, bool mandatory = default)
+        {
+            if (Channel.IsOpen)
+            {
+                Channel.BasicPublish(exchangeName, routeKey, mandatory, properties, body);
+                return Channel.WaitForConfirms();
+            }
+            else
+            {
+                _logger.LogInformation("class-id:{0}\nreply-code:{1}\nreply-text:{2}", Channel.CloseReason.ClassId, Channel.CloseReason.ReplyCode, Channel.CloseReason.ReplyText);
+                return false;
+            }
+        }
+
+        public virtual void Dispose()
+        {
+            if (!_disposed)
+            {
+                _channel?.Close();
+                _channel = null;
+                _connection = null;
+                _disposed = true;
+            }
+        }
+
+        public virtual void SetPool(object pool)
+        {
+            Check.NotNull(pool, nameof(pool));
+            Check.IsType<RabbitMqConnection>(pool, nameof(pool));
+
+            _connection = pool as RabbitMqConnection;
+        }
+
+        /// <summary>
+        /// 执行通道(channel)通道配置事件。调用事件<see cref="ChannelConfiged"/>，并设置<see cref="Configed"/>为true
+        /// </summary>
+        /// <param name="options">配置选项</param>
+        /// <seealso cref="ChannelConfiged"/>
+        public void HandleChannelConfig(RabbitMqOptions options)
+        {
+            Check.NotNull(options, nameof(options));
+
+            var source = new ConfigEventArgs()
+            {
+                Options = options
+            };
+
+            if (ChannelConfiged != null)
+            {
+                ChannelConfiged(this, source);
+                Configed = true;
+            }
+        }
+    }
+}
diff --git a/RabbitMqConnection.cs b/RabbitMqConnection.cs
@@ -0,0 +1,129 @@
+﻿using Core.Extensions;
+using Microsoft.Extensions.Logging;
+using RabbitMq.Abtraction;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Framing.Impl;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using static RabbitMq.RabbitMqConnectionTable;
+
+namespace RabbitMq
+{
+    public class RabbitMqConnection : AbractPool<RabbitMqChannel>, IPoolable
+    {
+        private IConnection _connection;
+
+        private ConnectionPool _connectionPool;
+
+        private readonly RabbitMqOptions _options;
+
+        public RabbitMqOptions Options { get => _options; }
+
+        public IConnection Connection { get => _connection; }
+
+        public ILogger Logger { get => _logger; }
+
+        public RabbitMqConnection(ConnectionPool parentPool, ushort channleMax = 50)
+        {
+            Check.NotNull(parentPool, nameof(parentPool));
+
+            _options = parentPool.Options;
+            _connection = parentPool.ConnectionFactory?.CreateConnection() ?? throw new ArgumentNullException(nameof(parentPool));
+            _logger = RabbitMqConnectionTable.LoggerFactory.CreateLogger(GetType());
+            _maxSize = channleMax;
+            SetPool(parentPool);
+        }
+
+        public RabbitMqChannel CreateChannel()
+        {
+            var channel = new RabbitMqChannel(this);
+            return channel;
+        }
+
+        public void ExchangesDeclare(string serverName)
+        {
+            var channel = Rent();
+            var model = channel.Channel;
+            foreach (var exchange in _options.Exchanges.Where(e => e.ServerName == serverName))
+            {
+                model.ExchangeDeclare(exchange.Name, exchange.Type, exchange.Durable, exchange.AutoDelete, exchange.Arguements);
+                foreach (var queue in exchange.BindedQueues)
+                {
+                    model.QueueDeclare(queue.Name, queue.Durable, queue.Exclusive, queue.AutoDelete, queue.Arguements);
+                    model.QueueBind(queue.Name, exchange.Name, exchange.RouteKey, exchange.Arguements);
+                }
+            }
+            Return(channel);
+        }
+
+        public void QueuesDeclare(string serverName)
+        {
+            var channel = Rent();
+            var model = channel.Channel;
+            foreach (var queue in _options.Queues.Where(q => q.ServerName == serverName))
+            {
+                model.QueueDeclare(queue.Name, queue.Durable, queue.Exclusive, queue.AutoDelete, queue.Arguements);
+            }
+            Return(channel);
+        }
+
+        public override RabbitMqChannel Rent()
+        {
+            if (_pool.TryDequeue(out RabbitMqChannel channel))
+            {
+                return channel;
+            }
+
+            else if (Interlocked.Increment(ref _createdCount) <= _maxSize)
+            {
+                channel = CreateChannel();
+                channel.SetPool(this);
+                return channel;
+            }
+
+            else
+            {
+                Interlocked.Decrement(ref _createdCount);
+                return default;
+            }
+        }
+
+        public override void Return(RabbitMqChannel model)
+        {
+            int currentCount = Count;
+            if (Interlocked.Increment(ref currentCount) <= _maxSize)
+            {
+                _pool.Enqueue(model);
+            }
+            else
+            {
+                throw new ArgumentOutOfRangeException(nameof(model));
+            }
+        }
+
+        public override void Dispose()
+        {
+            if (!_disposed)
+            {
+                base.Dispose();
+                _connection?.Close();
+
+                _connection = null;
+                _disposed = true;
+            }
+        }
+
+        public virtual void SetPool(object pool)
+        {
+            Check.NotNull(pool, nameof(pool));
+            Check.IsType<ConnectionPool>(pool, nameof(pool));
+
+            _connectionPool = pool as ConnectionPool;
+        }
+    }
+}
diff --git a/RabbitMqConnectionTable.cs b/RabbitMqConnectionTable.cs
diff --git a/RabbitMqHelper.cs b/RabbitMqHelper.cs
diff --git a/RabbitMqOptions.cs b/RabbitMqOptions.cs
