The options EnableConsumerPrefetch and UseDispatchingPerGroup will wo…

…rk together without interference. (#1399)
dotnetcore · Sep 8, 2023 · 7f84af5 · 7f84af5
1 parent ff76dea
commit 7f84af5
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/src/DotNetCore.CAP/CAP.Options.cs b/src/DotNetCore.CAP/CAP.Options.cs
@@ -91,18 +91,15 @@ public CapOptions()
 
     /// <summary>
     /// If true, the message will be pre fetch to memory queue for parallel execute by thread pool.
-    /// <para>Not available when <see cref="UseDispatchingPerGroup"/> true.</para>
     /// Default is false
     /// </summary>
     public bool EnableConsumerPrefetch { get; set; }
 
     /// <summary>
     /// If true then each message group will have own independent dispatching pipeline. Each pipeline use as many threads as
     /// <see cref="ConsumerThreadCount" /> value is.
-    /// <para>If true, the <see cref="EnableConsumerPrefetch"/> is not available.</para>
     /// Default is false.
     /// </summary>
-    [Obsolete("Use EnableConsumerPrefetch instead. Setting it to true means that each consumer is now executed concurrently by thread pool, regardless of whether they are in different groups.")]
     public bool UseDispatchingPerGroup { get; set; }
 
     /// <summary>

diff --git a/src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs b/src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs
@@ -119,6 +119,7 @@ public void Execute()
             ICollection<string> topics;
             try
             {
+                // ReSharper disable once ConvertToUsingDeclaration
                 using (var client = _consumerClientFactory.Create(matchGroup.Key))
                 {
                     client.OnLogCallback = WriteLog;
@@ -139,6 +140,7 @@ public void Execute()
                   {
                       try
                       {
+                          // ReSharper disable once ConvertToUsingDeclaration
                           using (var client = _consumerClientFactory.Create(matchGroup.Key))
                           {
                               _serverAddress = client.BrokerAddress;

diff --git a/src/DotNetCore.CAP/Processor/IDispatcher.PerGroup.cs b/src/DotNetCore.CAP/Processor/IDispatcher.PerGroup.cs
@@ -27,6 +27,7 @@ internal class DispatcherPerGroup : IDispatcher
     private readonly IMessageSender _sender;
     private readonly IDataStorage _storage;
     private readonly PriorityQueue<MediumMessage, DateTime> _schedulerQueue;
+    private readonly bool _enablePrefetch;
 
     private Channel<MediumMessage> _publishedChannel = default!;
     private ConcurrentDictionary<string, Channel<(MediumMessage, ConsumerExecutorDescriptor?)>> _receivedChannels = default!;
@@ -45,6 +46,7 @@ public DispatcherPerGroup(ILogger<Dispatcher> logger,
         _executor = executor;
         _schedulerQueue = new PriorityQueue<MediumMessage, DateTime>();
         _storage = storage;
+        _enablePrefetch = options.Value.EnableConsumerPrefetch;
     }
 
     public async Task Start(CancellationToken stoppingToken)
@@ -242,7 +244,15 @@ private async ValueTask Processing(string group, Channel<(MediumMessage, Consume
 
                         var item1 = message.Item1;
                         var item2 = message.Item2;
-                        _ = Task.Run(() => _executor.ExecuteAsync(item1, item2, _tasksCts.Token).ConfigureAwait(false));
+
+                        if (_enablePrefetch)
+                        {
+                            _ = Task.Run(() => _executor.ExecuteAsync(item1, item2, _tasksCts.Token).ConfigureAwait(false));
+                        }
+                        else
+                        {
+                            await _executor.ExecuteAsync(item1, item2, _tasksCts.Token).ConfigureAwait(false);
+                        }
                     }
                     catch (OperationCanceledException)
                     {