内部消息队列实现

Data/OSS.Tools.DataFlow/DataFlowFactory.cs

@@ -1,4 +1,5 @@
-using System;
+using OSS.Tools.DataFlow.Inter;
+using System;
 using System.Threading.Tasks;
 
 namespace OSS.Tools.DataFlow

Data/OSS.Tools.DataFlow/DefaultDataFlow.cs

@@ -1,4 +1,5 @@
 using System.Threading.Tasks;
+using OSS.Tools.DataFlow.Inter;
 
 namespace OSS.Tools.DataFlow
 {
@@ -9,6 +10,7 @@ namespace OSS.Tools.DataFlow
     public class DefaultDataFlow<TData> : IDataPublisher<TData>
     {
         private IDataSubscriber<TData> _subscriber;
+
         /// <summary>
         ///  构造函数
         /// </summary>
@@ -26,7 +28,7 @@ namespace OSS.Tools.DataFlow
         public Task<bool> Publish(TData data)
         {
             Task.Factory.StartNew((obj) => { _subscriber?.Subscribe((TData) obj); }, data);
-            return Task.FromResult(true);
+            return InterUtils.TrueTask;
         }
     }
 }

Data/OSS.Tools.DataFlow/IDataSubscriber.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 
 namespace OSS.Tools.DataFlow
 {
@@ -17,18 +16,5 @@ namespace OSS.Tools.DataFlow
         Task<bool> Subscribe(TData data);
     }
 
-    internal class InterDataSubscriber<TData> : IDataSubscriber<TData>
-    {
-        private readonly Func<TData, Task<bool>> _subscriber;
-
-        internal InterDataSubscriber(Func<TData, Task<bool>> subscribeFunc)
-        {
-            _subscriber = subscribeFunc ?? throw new ArgumentNullException(nameof(subscribeFunc),"订阅者方法不能为空！");
-        }
-
-        public Task<bool> Subscribe(TData data)
-        {
-            return _subscriber.Invoke(data);
-        }
-    }
+  
 }

Data/OSS.Tools.DataFlow/Inter/InterDataSubscriber.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Threading.Tasks;
+
+namespace OSS.Tools.DataFlow.Inter
+{
+
+    internal class InterDataSubscriber<TData> : IDataSubscriber<TData>
+    {
+        private readonly Func<TData, Task<bool>> _subscriber;
+
+        internal InterDataSubscriber(Func<TData, Task<bool>> subscribeFunc)
+        {
+            _subscriber = subscribeFunc ?? throw new ArgumentNullException(nameof(subscribeFunc), "订阅者方法不能为空！");
+        }
+
+        public Task<bool> Subscribe(TData data)
+        {
+            return _subscriber.Invoke(data);
+        }
+
+        internal Task<bool> Subscribe(object data)
+        {
+            return Subscribe((TData)data);
+        }
+    }
+}

Data/OSS.Tools.DataFlow/Inter/InterQueueHub.cs

@@ -0,0 +1,77 @@
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace OSS.Tools.DataFlow.Inter
+{
+    internal static class InterQueueHub
+    {
+        private static readonly object                     _syncObj   = new object();
+        private static readonly ConcurrentQueue<InterData> _dataQueue = new ConcurrentQueue<InterData>();
+
+        private static readonly ConcurrentDictionary<string, InterDataSubscriber<InterData>> _keySubscriberMaps =
+            new ConcurrentDictionary<string, InterDataSubscriber<InterData>>();
+
+        public static Task<bool> Publish(InterData data)
+        {
+            _dataQueue.Enqueue(data);
+            CheckConsumer();
+            return InterUtils.TrueTask;
+        }
+        
+        private static void CheckConsumer()
+        {
+            GuardThread();
+            autoEvent.Set();
+        }
+
+        private static Thread _consumerThread;
+        private static void GuardThread()
+        {
+            if (_consumerThread == null || !_consumerThread.IsAlive)
+            {
+                lock (_syncObj)
+                {
+                    if (_consumerThread == null || !_consumerThread.IsAlive)
+                    {
+                        _consumerThread              = new Thread(ConsumerData);
+                        _consumerThread.IsBackground = true;
+                        _consumerThread.Start();
+                    }
+                }
+            }
+        }
+
+        private static AutoResetEvent autoEvent = new AutoResetEvent(true);
+
+        private static void ConsumerData()
+        {
+            while (true)
+            {
+                autoEvent.WaitOne();
+                while (!_dataQueue.IsEmpty)
+                {
+                    NoticeSubscriber();
+                }
+            }
+
+        }
+
+        private static void NoticeSubscriber()
+        {
+            if (_dataQueue.TryDequeue(out InterData data))
+            {
+                if (_keySubscriberMaps.TryGetValue(data.data_flow_key, out var subscriber))
+                {
+                    subscriber.Subscribe(data);
+                }
+            }
+        }
+    }
+
+
+    internal class InterData
+    {
+        public string data_flow_key { get; set; }
+    }
+}

Data/OSS.Tools.DataFlow/Inter/InterUtils.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace OSS.Tools.DataFlow.Inter
+{
+    public static class InterUtils
+    {
+        public static Task<bool> TrueTask => Task.FromResult(true);
+    }
+}