From d8fceb69f6130b6f9f958bf52a2c822fb7bd99f8 Mon Sep 17 00:00:00 2001
From: Savorboard <yangxiaodong1214@126.com>
Date: Sun, 16 Feb 2020 18:13:23 +0800
Subject: [PATCH] Replace BlockingCollection with Channel to improve
 performance. #492

---
 src/DotNetCore.CAP/DotNetCore.CAP.csproj      |  3 +-
 .../Processor/IDispatcher.Default.cs          | 40 ++++++++++++-------
 2 files changed, 27 insertions(+), 16 deletions(-)

diff --git a/src/DotNetCore.CAP/DotNetCore.CAP.csproj b/src/DotNetCore.CAP/DotNetCore.CAP.csproj
index 58dd39d..a286b04 100644
--- a/src/DotNetCore.CAP/DotNetCore.CAP.csproj
+++ b/src/DotNetCore.CAP/DotNetCore.CAP.csproj
@@ -13,7 +13,8 @@
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="3.1.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />
-    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="4.6.0" />
+    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="4.7.0" />
+    <PackageReference Include="System.Threading.Channels" Version="4.7.0" />
   </ItemGroup>
  
 </Project>
\ No newline at end of file
diff --git a/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs b/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs
index 74e4014..2139ee1 100644
--- a/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs
+++ b/src/DotNetCore.CAP/Processor/IDispatcher.Default.cs
@@ -2,14 +2,16 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using System.Collections.Concurrent;
+using System.Linq;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Internal;
 using DotNetCore.CAP.Messages;
 using DotNetCore.CAP.Persistence;
 using DotNetCore.CAP.Transport;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 
 namespace DotNetCore.CAP.Processor
 {
@@ -20,32 +22,35 @@ namespace DotNetCore.CAP.Processor
         private readonly ISubscribeDispatcher _executor;
         private readonly ILogger<Dispatcher> _logger;
 
-        private readonly BlockingCollection<MediumMessage> _publishedMessageQueue =
-            new BlockingCollection<MediumMessage>(new ConcurrentQueue<MediumMessage>());
-
-        private readonly BlockingCollection<(MediumMessage, ConsumerExecutorDescriptor)> _receivedMessageQueue =
-            new BlockingCollection<(MediumMessage, ConsumerExecutorDescriptor)>(new ConcurrentQueue<(MediumMessage, ConsumerExecutorDescriptor)>());
+        private readonly Channel<MediumMessage> _publishedChannel;
+        private readonly Channel<(MediumMessage, ConsumerExecutorDescriptor)> _receivedChannel;
 
         public Dispatcher(ILogger<Dispatcher> logger,
             IMessageSender sender,
+            IOptions<CapOptions> options,
             ISubscribeDispatcher executor)
         {
             _logger = logger;
             _sender = sender;
             _executor = executor;
 
+            _publishedChannel = Channel.CreateUnbounded<MediumMessage>(new UnboundedChannelOptions() { SingleReader = true, SingleWriter = true });
+            _receivedChannel = Channel.CreateUnbounded<(MediumMessage, ConsumerExecutorDescriptor)>();
+
             Task.Factory.StartNew(Sending, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
-            Task.Factory.StartNew(Processing, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+
+            Task.WhenAll(Enumerable.Range(0, options.Value.ConsumerThreadCount)
+                .Select(_ => Task.Factory.StartNew(Processing, _cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default)).ToArray());
         }
 
         public void EnqueueToPublish(MediumMessage message)
         {
-            _publishedMessageQueue.Add(message);
+            _publishedChannel.Writer.TryWrite(message);
         }
 
         public void EnqueueToExecute(MediumMessage message, ConsumerExecutorDescriptor descriptor)
         {
-            _receivedMessageQueue.Add((message, descriptor));
+            _receivedChannel.Writer.TryWrite((message, descriptor));
         }
 
         public void Dispose()
@@ -57,21 +62,23 @@ namespace DotNetCore.CAP.Processor
         {
             try
             {
-                while (!_publishedMessageQueue.IsCompleted)
+                while (await _publishedChannel.Reader.WaitToReadAsync(_cts.Token))
                 {
-                    if (_publishedMessageQueue.TryTake(out var message, 3000, _cts.Token))
+                    while (_publishedChannel.Reader.TryRead(out var message))
                     {
                         try
                         {
                             var result = await _sender.SendAsync(message);
                             if (!result.Succeeded)
                             {
-                                _logger.MessagePublishException(message.Origin.GetId(), result.ToString(), result.Exception);
+                                _logger.MessagePublishException(message.Origin.GetId(), result.ToString(),
+                                    result.Exception);
                             }
                         }
                         catch (Exception ex)
                         {
-                            _logger.LogError(ex, $"An exception occurred when sending a message to the MQ. Id:{message.DbId}");
+                            _logger.LogError(ex,
+                                $"An exception occurred when sending a message to the MQ. Id:{message.DbId}");
                         }
                     }
                 }
@@ -86,9 +93,12 @@ namespace DotNetCore.CAP.Processor
         {
             try
             {
-                foreach (var message in _receivedMessageQueue.GetConsumingEnumerable(_cts.Token))
+                while (await _receivedChannel.Reader.WaitToReadAsync(_cts.Token))
                 {
-                    await _executor.DispatchAsync(message.Item1, message.Item2, _cts.Token);
+                    while (_receivedChannel.Reader.TryRead(out var message))
+                    {
+                        await _executor.DispatchAsync(message.Item1, message.Item2, _cts.Token);
+                    }
                 }
             }
             catch (OperationCanceledException)