Merge pull request #50 from ojdev/dev

Dev

Author: ojdev

.github/workflows/dotnetcore.yml

@@ -1,21 +1,32 @@
-name: .NET Core
+name: .NET Core App CI to Docker Hub
 
-on: 
+on:
   push:
-    branches:
-      - master
+    tags:
+    - v*
+
 
 jobs:
   build:
-
-    runs-on: [ubuntu-latest]
-
+  
+    runs-on: ubuntu-latest    
+    env:
+      TZ: Asia/Shanghai
+      TAG_NUMBER: $GITHUB_RUN_NUMBER
     steps:
-    - uses: actions/checkout@v1
-    - name: Setup .NET Core
+    - uses: actions/checkout@v2
+    - name: Setup .NET
       uses: actions/setup-dotnet@v1
       with:
         dotnet-version: 5.0.301
+    - name: Get tag
+      id: tag
+      uses: dawidd6/action-get-tag@v1
+      with:
+        # Optionally strip `v` prefix
+        strip_v: true
+    - name: Set VERSION variable from tag
+      run: echo ${{steps.tag.outputs.tag}} 
     - name: Build with dotnet
       run: dotnet build --configuration Release src/RabbitMQ.EventBus.AspNetCore 
     - name: Pack

RabbitMQ.EventBus.AspNetCore.Sample/Controllers/ValuesController.cs

@@ -24,6 +24,7 @@ public async Task<ActionResult<string>> Get()
             //    Body = "rabbitmq.eventbus.test=>发送消息",
             //    Time = DateTimeOffset.Now
             //}, exchange: "RabbitMQ.EventBus.Simple", routingKey: "rabbitmq.eventbus.test");
+
             for (int i = 0; i < 1000; i++)
             {
                 _eventBus.Publish(new

RabbitMQ.EventBus.AspNetCore.Sample/Startup.cs

@@ -27,7 +27,7 @@ public void ConfigureServices(IServiceCollection services)
             services.AddControllers();
             services.AddHealthChecks();
 
-            services.AddRabbitMQEventBus(() => "amqp://guest:guest@192.168.0.250:5672/", eventBusOptionAction: eventBusOption =>
+            services.AddRabbitMQEventBus(() => "amqp://guest:guest@localhost:5672/", eventBusOptionAction: eventBusOption =>
              {
                  eventBusOption.ClientProvidedAssembly(assemblyName);
                  eventBusOption.EnableRetryOnFailure(true, 5000, TimeSpan.FromSeconds(30));

src/RabbitMQ.EventBus.AspNetCore/Configurations/DeadLetterExchangeConfig.cs

@@ -24,11 +24,11 @@ public class DeadLetterExchangeConfig
         /// <summary>
         /// 
         /// </summary>
-        /// <param name="enabled">是否开启(默认开启)</param>
+        /// <param name="enabled">是否开启(默认关闭)</param>
         /// <param name="exchangeNamePrefix">交换机名前缀(默认为"dead-")</param>
         /// <param name="exchangeNameSuffix">交换机名后缀</param>
         /// <param name="customizeExchangeName">自定义交换机名(留空则使用原有的交换机名)</param>
-        public DeadLetterExchangeConfig(bool enabled = true, string exchangeNamePrefix = "dead-", string exchangeNameSuffix = null, string customizeExchangeName = null)
+        public DeadLetterExchangeConfig(bool enabled = false, string exchangeNamePrefix = "dead-", string exchangeNameSuffix = null, string customizeExchangeName = null)
         {
             Enabled = enabled;
             ExchangeNameSuffix = exchangeNameSuffix;

src/RabbitMQ.EventBus.AspNetCore/DefaultRabbitMQEventBus.cs

@@ -23,6 +23,7 @@ internal class DefaultRabbitMQEventBus : IRabbitMQEventBus
         private readonly ILogger<DefaultRabbitMQEventBus> _logger;
         private readonly IServiceProvider _serviceProvider;
         private readonly IEventHandlerModuleFactory _eventHandlerFactory;
+        private readonly Dictionary<string, IModel> subscribes;
         /// <summary>
         /// 
         /// </summary>
@@ -36,9 +37,10 @@ public DefaultRabbitMQEventBus(IRabbitMQPersistentConnection persistentConnectio
             _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
             _eventHandlerFactory = eventHandlerFactory ?? throw new ArgumentNullException(nameof(eventHandlerFactory));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+            subscribes = new Dictionary<string, IModel>();
+            _logger.LogInformation("消息队列准备就绪");
         }
 
-        private IModel _publishChannel;
         /// <summary>
         /// 
         /// </summary>
@@ -50,16 +52,9 @@ public DefaultRabbitMQEventBus(IRabbitMQPersistentConnection persistentConnectio
         public void Publish<TMessage>(TMessage message, string exchange, string routingKey, string type = ExchangeType.Topic)
         {
             string body = message.Serialize();
-            if (_publishChannel?.IsOpen != true)
-            {
-                if (_persistentConnection.IsConnected)
-                {
-                    _persistentConnection.TryConnect();
-                }
-                _publishChannel = _persistentConnection.ExchangeDeclare(exchange, type: type);
-                _publishChannel.BasicReturn += async (se, ex) => await Task.Delay((int)_persistentConnection.Configuration.ConsumerFailRetryInterval.TotalMilliseconds).ContinueWith(t => Publish(body, ex.Exchange, ex.RoutingKey));
-            }
-            
+            using var _publishChannel = _persistentConnection.ExchangeDeclare(exchange, type: type);
+            _publishChannel.BasicReturn += async (se, ex) => await Task.Delay((int)_persistentConnection.Configuration.ConsumerFailRetryInterval.TotalMilliseconds).ContinueWith(t => Publish(body, ex.Exchange, ex.RoutingKey));
+
             IBasicProperties properties = _publishChannel.CreateBasicProperties();
             properties.DeliveryMode = 2; // persistent
             _publishChannel.BasicPublish(exchange: exchange,
@@ -81,11 +76,9 @@ public void Subscribe(Type eventType, string type = ExchangeType.Topic)
                     string queue = attr.Queue ?? (_persistentConnection.Configuration.Prefix == QueuePrefixType.ExchangeName
                         ? $"{ attr.Exchange }.{ eventType.Name }"
                         : $"{_persistentConnection.Configuration.ClientProvidedName}.{ eventType.Name }");
-                    if (!_persistentConnection.IsConnected)
-                    {
-                        _persistentConnection.TryConnect();
-                    }
-                    IModel channel;
+
+                    var onlyKey = $"{attr.Exchange}_{queue}_{attr.RoutingKey}";
+                    subscribes.TryGetValue(onlyKey, out IModel channel);
                     #region snippet
                     var arguments = new Dictionary<string, object>();
 
@@ -135,6 +128,7 @@ public void Subscribe(Type eventType, string type = ExchangeType.Topic)
                     #endregion
                     channel.QueueBind(queue, attr.Exchange, attr.RoutingKey, null);
                     channel.BasicQos(0, _persistentConnection.Configuration.PrefetchCount, false);
+                    subscribes[onlyKey] = channel;
                     EventingBasicConsumer consumer = new(channel);
                     consumer.Received += async (model, ea) =>
                     {

src/RabbitMQ.EventBus.AspNetCore/Factories/DefaultRabbitMQPersistentConnection.cs

@@ -81,20 +81,19 @@ public bool TryConnect()
                 );
 
                 policy.Execute(() =>
-                {
-                    string connectionString = _connectionAction.Invoke();
-                    _logger.WriteLog(LogLevel.Information, $"[ConnectionString]:\t{connectionString}");
-                    _connectionFactory.Uri = new Uri(connectionString);
-                    _connection = _connectionFactory.CreateConnection(clientProvidedName: Configuration.ClientProvidedName);
-                });
+               {
+                   string connectionString = _connectionAction.Invoke();
+                   _logger.WriteLog(LogLevel.Information, $"[ConnectionString]:\t{connectionString}");
+                   _connectionFactory.Uri = new Uri(connectionString);
+                   _connection = _connectionFactory.CreateConnection(clientProvidedName: Configuration.ClientProvidedName);
+               });
 
                 if (IsConnected)
                 {
                     _connection.ConnectionShutdown += OnConnectionShutdown;
                     _connection.CallbackException += OnCallbackException;
                     _connection.ConnectionBlocked += OnConnectionBlocked;
                     _logger.WriteLog(LogLevel.Information, $"RabbitMQ persistent connection acquired a connection {_connection.Endpoint.HostName} and is subscribed to failure events");
-
                     return true;
                 }
                 else