Optimize Dispatcher message processing logic

Refactor the `Dispatcher` class to improve performance and reduce thread pool scheduling overhead:
- Replace `Task.Run` with direct async calls to `SendMessageAsync` for message sending.
- Ensure `batchSize` has a minimum value of 1 and preallocate the `tasks` list for efficiency.
- Add a conditional check to invoke `Task.WhenAll` only when tasks exist.
- Simplify non-parallel send logic by reusing `SendMessageAsync`.
- Introduce the `SendMessageAsync` method to encapsulate message sending, handle exceptions, and log errors.
- Use `.ConfigureAwait(false)` consistently for asynchronous calls.

These changes enhance maintainability, readability, and performance by reducing redundant code and optimizing asynchronous operations.

Author: yang-xiaodong

src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

@@ -249,39 +249,24 @@ private async ValueTask Sending()
             {
                 if (_enableParallelSend)
                 {
-                    var tasks = new List<Task>();
-                    var batchSize = _pChannelSize / 50;
+                    // Use direct async sends instead of Task.Run to reduce thread pool scheduling and allocations
+                    var batchSize = Math.Max(1, _pChannelSize / 50);
+                    var tasks = new List<Task>(batchSize);
+
                     for (var i = 0; i < batchSize && _publishedChannel.Reader.TryRead(out var message); i++)
                     {
-                        var item = message;
-                        tasks.Add(Task.Run(async () =>
-                        {
-                            try
-                            {
-                                var result = await _sender.SendAsync(item).ConfigureAwait(false);
-                                if (!result.Succeeded) _logger.MessagePublishException(item.Origin.GetId(), result.ToString(), result.Exception);
-                            }
-                            catch (Exception ex)
-                            {
-                                _logger.LogError(ex, $"An exception occurred when sending a message to the transport. Id:{message.DbId}");
-                            }
-                        }));
+                        tasks.Add(SendMessageAsync(message));
                     }
 
-                    await Task.WhenAll(tasks);
+                    if (tasks.Count > 0)
+                    {
+                        await Task.WhenAll(tasks).ConfigureAwait(false);
+                    }
                 }
                 else
                 {
                     while (_publishedChannel.Reader.TryRead(out var message))
-                        try
-                        {
-                            var result = await _sender.SendAsync(message).ConfigureAwait(false);
-                            if (!result.Succeeded) _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
-                        }
-                        catch (Exception ex)
-                        {
-                            _logger.LogError(ex, $"An exception occurred when sending a message to the transport. Id:{message.DbId}");
-                        }
+                        await SendMessageAsync(message).ConfigureAwait(false);
                 }
             }
 
@@ -292,6 +277,22 @@ private async ValueTask Sending()
         }
     }
 
+    private async Task SendMessageAsync(MediumMessage message)
+    {
+        try
+        {
+            var result = await _sender.SendAsync(message).ConfigureAwait(false);
+            if (!result.Succeeded)
+            {
+                _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "An exception occurred when sending a message to the transport. Id:{MessageId}", message.DbId);
+        }
+    }
+
     private async ValueTask Processing()
     {
         try