增加可设置的预取消息数

Author: ojdev

RabbitMQ.EventBus.AspNetCore.Sample/Startup.cs

@@ -33,6 +33,7 @@ public void ConfigureServices(IServiceCollection services)
                  eventBusOption.EnableRetryOnFailure(true, 5000, TimeSpan.FromSeconds(30));
                  eventBusOption.RetryOnFailure(TimeSpan.FromSeconds(1));
                  eventBusOption.MessageTTL(2000);
+                 eventBusOption.SetBasicQos(10);
                  eventBusOption.DeadLetterExchangeConfig(config =>
                  {
                      config.Enabled = true;

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

@@ -35,6 +35,10 @@ public sealed class RabbitMQEventBusConnectionConfiguration
         /// </summary>
         public LogLevel Level { get; set; }
         /// <summary>
+        /// 每次预取的消息条数（默认：1）
+        /// </summary>
+        public ushort PrefetchCount { get; set; }
+        /// <summary>
         /// 队列名前缀（默认ClientProvidedName）
         /// </summary>
         public QueuePrefixType Prefix { get; set; }
@@ -68,6 +72,7 @@ public RabbitMQEventBusConnectionConfiguration()
             AutomaticRecoveryEnabled = true;
             ConsumerFailRetryInterval = TimeSpan.FromSeconds(1);
             DeadLetterExchange = new DeadLetterExchangeConfig();
+            PrefetchCount = 1;
         }
     }
 }

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

@@ -83,5 +83,17 @@ public void DeadLetterExchangeConfig(Action<DeadLetterExchangeConfig> config)
                 Configuration.MessageTTL = 60000;
             }
         }
+        /// <summary>
+        /// 设置预取条数
+        /// </summary>
+        /// <param name="prefetchCount"></param>
+        public void SetBasicQos(ushort prefetchCount)
+        {
+            if (prefetchCount < 1)
+            {
+                throw new ArgumentOutOfRangeException($"{nameof(prefetchCount)}必须大于0");
+            }
+            Configuration.PrefetchCount = prefetchCount;
+        }
     }
 }

src/RabbitMQ.EventBus.AspNetCore/DefaultRabbitMQEventBus.cs

@@ -133,15 +133,16 @@ public void Subscribe(Type eventType, string type = ExchangeType.Topic)
                     }
                     #endregion
                     channel.QueueBind(queue, attr.Exchange, attr.RoutingKey, null);
-                    channel.BasicQos(0, 1, false);
+                    channel.BasicQos(0, _persistentConnection.Configuration.PrefetchCount, false);
                     EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                     consumer.Received += async (model, ea) =>
                     {
-                        string body = Encoding.UTF8.GetString(ea.Body);
+                        string body = Encoding.UTF8.GetString(ea.Body.ToArray());
                         bool isAck = false;
                         try
                         {
                             await ProcessEvent(body, eventType, ea);
+                            //不确定是否需要改变Multiple是否需要改为true
                             channel.BasicAck(ea.DeliveryTag, multiple: false);
                             isAck = true;
                         }

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

@@ -12,7 +12,7 @@
     <Description>asp.net core 下使用的RabbitMQ</Description>
     <PackageLicenseUrl></PackageLicenseUrl>
     <PackageProjectUrl>https://github.com/ojdev/RabbitMQ.EventBus.AspNetCore</PackageProjectUrl>
-    <Version>2.1.3</Version>
+    <Version>2.1.4</Version>
     <PackageLicenseFile>LICENSE</PackageLicenseFile>
   </PropertyGroup>
 
@@ -22,11 +22,11 @@
 
   <ItemGroup>
     <PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.2.0" />
-    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.0" />
-    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="3.1.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.3" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="3.1.3" />
     <PackageReference Include="Polly" Version="7.2.0" />
-    <PackageReference Include="RabbitMQ.Client" Version="5.1.2" />
-    <PackageReference Include="System.Text.Json" Version="4.7.0" />
+    <PackageReference Include="RabbitMQ.Client" Version="6.0.0" />
+    <PackageReference Include="System.Text.Json" Version="4.7.1" />
   </ItemGroup>
 
   <ItemGroup>