合并更新

Author: 欧俊

.github/workflows/dotnetcore.yml

@@ -1,24 +1,35 @@
-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
+        dotnet-version: 3.1.100
+    - 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
-      run: dotnet pack src/RabbitMQ.EventBus.AspNetCore -c Release --include-symbols --include-source  -p:PackageVersion=5.0.$GITHUB_RUN_NUMBER -o artifacts/
+      run: dotnet pack src/RabbitMQ.EventBus.AspNetCore -c Release --include-symbols --include-source  -p:PackageVersion=3.1.$GITHUB_RUN_NUMBER -o artifacts/
     - name: Publish Symbols to NuGet
       run: dotnet nuget push artifacts/*.symbols.nupkg --api-key ${{secrets.NUGET_API_KEY}} --source https://api.nuget.org/v3/index.json

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

@@ -24,15 +24,13 @@ 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++)
+            for (int i = 0; i < 100; i++)
             {
                 _eventBus.Publish(new
                 {
-                    Body = $"rabbitmq.eventbus.test1=>发送消息/t{i}",
+                    Body = "rabbitmq.eventbus.test1=>发送消息",
                     Time = DateTimeOffset.Now,
                 }, exchange: "RabbitMQ.EventBus.Simple", routingKey: "rabbitmq.eventbus.test1");
-                await Task.Yield();
-                await Task.Delay(500);
             }
             return "Ok";
         }

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));

docs/_config.yml

@@ -0,0 +1 @@
+theme: jekyll-theme-midnight

docs/index.md

@@ -0,0 +1,37 @@
+## Welcome to GitHub Pages
+
+You can use the [editor on GitHub](https://github.com/ojdev/RabbitMQ.EventBus.AspNetCore/edit/master/docs/index.md) to maintain and preview the content for your website in Markdown files.
+
+Whenever you commit to this repository, GitHub Pages will run [Jekyll](https://jekyllrb.com/) to rebuild the pages in your site, from the content in your Markdown files.
+
+### Markdown
+
+Markdown is a lightweight and easy-to-use syntax for styling your writing. It includes conventions for
+
+```markdown
+Syntax highlighted code block
+
+# Header 1
+## Header 2
+### Header 3
+
+- Bulleted
+- List
+
+1. Numbered
+2. List
+
+**Bold** and _Italic_ and `Code` text
+
+[Link](url) and ![Image](src)
+```
+
+For more details see [GitHub Flavored Markdown](https://guides.github.com/features/mastering-markdown/).
+
+### Jekyll Themes
+
+Your Pages site will use the layout and styles from the Jekyll theme you have selected in your [repository settings](https://github.com/ojdev/RabbitMQ.EventBus.AspNetCore/settings/pages). The name of this theme is saved in the Jekyll `_config.yml` configuration file.
+
+### Support or Contact
+
+Having trouble with Pages? Check out our [documentation](https://docs.github.com/categories/github-pages-basics/) or [contact support](https://support.github.com/contact) and we’ll help you sort it out.

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,8 @@ 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 +75,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,7 +127,8 @@ 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);
-                    EventingBasicConsumer consumer = new(channel);
+                    subscribes[onlyKey] = channel;
+                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                     consumer.Received += async (model, ea) =>
                     {
                         string body = Encoding.UTF8.GetString(ea.Body.ToArray());

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