Merge pull request #49 from ojdev/dev

只保留一个Channel

Author: ojdev

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

@@ -1,5 +1,6 @@
 using Microsoft.AspNetCore.Mvc;
 using System;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.EventBus.AspNetCore.Simple.Controllers
 {
@@ -16,18 +17,23 @@ public ValuesController(IRabbitMQEventBus eventBus)
 
         // GET api/values
         [HttpGet]
-        public ActionResult<string> Get()
+        public async Task<ActionResult<string>> Get()
         {
             //_eventBus.Publish(new
             //{
             //    Body = "rabbitmq.eventbus.test=>发送消息",
             //    Time = DateTimeOffset.Now
             //}, exchange: "RabbitMQ.EventBus.Simple", routingKey: "rabbitmq.eventbus.test");
-            _eventBus.Publish(new
+            for (int i = 0; i < 1000; i++)
             {
-                Body = "rabbitmq.eventbus.test1=>发送消息",
-                Time = DateTimeOffset.Now,
-            }, exchange: "RabbitMQ.EventBus.Simple", routingKey: "rabbitmq.eventbus.test1");
+                _eventBus.Publish(new
+                {
+                    Body = $"rabbitmq.eventbus.test1=>发送消息/t{i}",
+                    Time = DateTimeOffset.Now,
+                }, exchange: "RabbitMQ.EventBus.Simple", routingKey: "rabbitmq.eventbus.test1");
+                await Task.Yield();
+                await Task.Delay(500);
+            }
             return "Ok";
         }
 

src/RabbitMQ.EventBus.AspNetCore/DefaultRabbitMQEventBus.cs

@@ -59,6 +59,7 @@ public void Publish<TMessage>(TMessage message, string exchange, string routingK
                 _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,

src/RabbitMQ.EventBus.AspNetCore/Extensions/ServiceCollectionExtensions.cs

@@ -33,6 +33,7 @@ public static IServiceCollection AddRabbitMQEventBus(this IServiceCollection ser
                 ILogger<DefaultRabbitMQPersistentConnection> logger = options.GetRequiredService<ILogger<DefaultRabbitMQPersistentConnection>>();
                 var connection = new DefaultRabbitMQPersistentConnection(configuration, connectionAction, logger);
                 connection.TryConnect();
+                Console.WriteLine("创建一次连接");
                 return connection;
             });
             services.TryAddSingleton<IEventHandlerModuleFactory, EventHandlerModuleFactory>();

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

@@ -20,7 +20,7 @@ internal sealed class DefaultRabbitMQPersistentConnection : IRabbitMQPersistentC
         Func<string> _connectionAction;
         private bool _disposed;
         private readonly object sync_root = new object();
-
+        private IModel _model;
         public string Endpoint => _connection?.Endpoint.ToString();
         public DefaultRabbitMQPersistentConnection(RabbitMQEventBusConnectionConfiguration configuration, Func<string> connectionAction, ILogger<DefaultRabbitMQPersistentConnection> logger)
         {
@@ -42,7 +42,12 @@ public IModel CreateModel()
             {
                 throw new InvalidOperationException("No RabbitMQ connections are available to perform this action");
             }
-            return _connection.CreateModel();
+            if ((_model?.IsOpen ?? false) == false)
+            {
+                _model = _connection.CreateModel();
+                Console.WriteLine("创建一个Channel");
+            }
+            return _model;
         }
 
         public void Dispose()

src/RabbitMQ.EventBus.AspNetCore/RabbitMQ.EventBus.AspNetCore.csproj

@@ -26,7 +26,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="5.0.0" />
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
     <PackageReference Include="Polly" Version="7.2.2" />
-    <PackageReference Include="RabbitMQ.Client" Version="6.2.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
   </ItemGroup>
 
   <ItemGroup>