Supports disabling parallel message sending and modifies the default …

…behavior to serial sending. (#1480)
dotnetcore · Feb 20, 2024 · a920d22 · a920d22
1 parent 4d12dc3
commit a920d22
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 8 deletions.
diff --git a/src/DotNetCore.CAP/CAP.Options.cs b/src/DotNetCore.CAP/CAP.Options.cs
@@ -23,6 +23,7 @@ public CapOptions()
         FailedRetryInterval = 60;
         FailedRetryCount = 50;
         ConsumerThreadCount = 1;
+        EnablePublishParallelSend = false;
         EnableConsumerPrefetch = false;
         Extensions = new List<ICapOptionsExtension>();
         Version = "v1";
@@ -96,6 +97,12 @@ public CapOptions()
     /// </summary>
     public bool EnableConsumerPrefetch { get; set; }
 
+    /// <summary>
+    /// If true, the message send task will be parallel execute by .net thread pool.
+    /// Default is false
+    /// </summary>
+    public bool EnablePublishParallelSend { 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.

diff --git a/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs b/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs
@@ -27,6 +27,7 @@ public class Dispatcher : IDispatcher
     private readonly IDataStorage _storage;
     private readonly PriorityQueue<MediumMessage, long> _schedulerQueue;
     private readonly bool _enablePrefetch;
+    private readonly bool _enableParallelSend;
 
     private Channel<MediumMessage> _publishedChannel = default!;
     private Channel<(MediumMessage, ConsumerExecutorDescriptor?)> _receivedChannel = default!;
@@ -45,6 +46,7 @@ public Dispatcher(ILogger<Dispatcher> logger,
         _schedulerQueue = new PriorityQueue<MediumMessage, long>();
         _storage = storage;
         _enablePrefetch = options.Value.EnableConsumerPrefetch;
+        _enableParallelSend = options.Value.EnablePublishParallelSend;
     }
 
     public async Task Start(CancellationToken stoppingToken)
@@ -209,12 +211,19 @@ private async ValueTask Sending()
                     try
                     {
                         var item = message;
-                        _ = Task.Run(async () =>
+                        if (_enableParallelSend)
+                        {
+                            _ = Task.Run(async () =>
+                            {
+                                var result = await _sender.SendAsync(item).ConfigureAwait(false);
+                                if (!result.Succeeded) _logger.MessagePublishException(item.Origin.GetId(), result.ToString(), result.Exception);
+                            });
+                        }
+                        else
                         {
                             var result = await _sender.SendAsync(item).ConfigureAwait(false);
-                            if (!result.Succeeded)
-                                _logger.MessagePublishException(item.Origin.GetId(), result.ToString(), result.Exception);
-                        });
+                            if (!result.Succeeded) _logger.MessagePublishException(item.Origin.GetId(), result.ToString(), result.Exception);
+                        }
                     }
                     catch (Exception ex)
                     {

diff --git a/src/DotNetCore.CAP/Processor/IDispatcher.PerGroup.cs b/src/DotNetCore.CAP/Processor/IDispatcher.PerGroup.cs
@@ -28,6 +28,7 @@ internal class DispatcherPerGroup : IDispatcher
     private readonly IDataStorage _storage;
     private readonly PriorityQueue<MediumMessage, DateTime> _schedulerQueue;
     private readonly bool _enablePrefetch;
+    private readonly bool _enableParallelSend;
 
     private Channel<MediumMessage> _publishedChannel = default!;
     private ConcurrentDictionary<string, Channel<(MediumMessage, ConsumerExecutorDescriptor?)>> _receivedChannels = default!;
@@ -47,6 +48,7 @@ public DispatcherPerGroup(ILogger<Dispatcher> logger,
         _schedulerQueue = new PriorityQueue<MediumMessage, DateTime>();
         _storage = storage;
         _enablePrefetch = options.Value.EnableConsumerPrefetch;
+        _enableParallelSend = options.Value.EnablePublishParallelSend;
     }
 
     public async Task Start(CancellationToken stoppingToken)
@@ -214,12 +216,19 @@ private async ValueTask Sending()
                 while (_publishedChannel.Reader.TryRead(out var message))
                     try
                     {
-                        _ = Task.Run(async () =>
+                        if (_enableParallelSend)
+                        {
+                            _ = Task.Run(async () =>
+                            {
+                                var result = await _sender.SendAsync(message).ConfigureAwait(false);
+                                if (!result.Succeeded) _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
+                            });
+                        }
+                        else
                         {
                             var result = await _sender.SendAsync(message).ConfigureAwait(false);
-                            if (!result.Succeeded)
-                                _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
-                        });
+                            if (!result.Succeeded) _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
+                        }
                     }
                     catch (Exception ex)
                     {