1

src/YSAI.DAQ/YSAI.Can/YSAI.Can.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.3</Version>
+	  <Version>1.0.0.4</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -24,7 +24,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
   <!--<ItemGroup>

src/YSAI.DAQ/YSAI.Core/YSAI.Core.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
     <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-    <Version>1.0.0.19</Version>
+    <Version>1.0.0.20</Version>
     <Authors>Shun</Authors>
     <Company>YSAI</Company>
     <Product>SCADA</Product>

src/YSAI.DAQ/YSAI.Core/data/AddressDetails.cs

@@ -20,6 +20,7 @@ namespace YSAI.Core.data
         /// </summary>
         public string? SN { get; set; }
         /// <summary>
+        /// 唯一
         /// 实际的点位地址名称
         /// 是扩展数据时，可为空
         /// 不是扩展数据时，此项为PLC的地址

src/YSAI.DAQ/YSAI.Core/subscribe/SubscribeData.cs

@@ -72,8 +72,8 @@ namespace YSAI.Core.subscription
             /// <summary>
             /// 休眠时间（毫秒）
             /// </summary>
-            [Description("休眠时间")]
-            public int SleepTime { get; set; } = 500;
+            [Description("休眠时间（毫秒）")]
+            public int SleepTime { get; set; } = 1000;
             /// <summary>
             /// 数据变化抛出 false则为实时数据
             /// 只抛出变化项
@@ -88,6 +88,19 @@ namespace YSAI.Core.subscription
             /// </summary>
             [Description("未变项与变化项一同抛出")]
             public bool SameDataOut { get; set; }
+
+            /// <summary>
+            /// 任务处理数量
+            /// </summary>
+            [Description("任务处理数量")]
+            public int TaskHandleCount { get; set; } = 10;
+            /// <summary>
+            /// 任务处理完成休眠时间(毫秒)
+            /// </summary>
+            [Description("任务处理完成休眠时间(毫秒)")]
+            public int TaskHandleAccomplishSleepTime { get; set; } = 1000;
+
+
             /// <summary>
             /// 重写基类中的Equals方法
             /// </summary>

src/YSAI.DAQ/YSAI.Core/subscribe/SubscribeOperate.cs

@@ -68,9 +68,9 @@ namespace YSAI.Core.subscription
         SubscribeData.Basics basics;
 
         /// <summary>
-        /// 线程
+        /// 轮询任务处理
         /// </summary>
-        Thread thread;
+        Task PollTaskHandle;
 
         /// <summary>
         /// 别的操作进来是否继续轮询
@@ -144,106 +144,37 @@ namespace YSAI.Core.subscription
         /// <summary>
         /// 执行轮询
         /// </summary>
-        void Polling()
+        Task Polling()
         {
-            while (!MonitorSwitch.IsCancellationRequested)
+            //起个新线程处理
+            return Task.Factory.StartNew(() =>
             {
-                try
+                while (!MonitorSwitch.IsCancellationRequested)
                 {
-                    if (!GoOn) continue;
-                    lock (basics.Address)  //锁住不让其他操作
+                    try
                     {
-                        //数据必须大于0才进行
-                        if (basics.Address.AddressArray.Count > 0)
+                        if (!GoOn) continue;
+                        lock (basics.Address)  //锁住不让其他操作
                         {
-                            //执行委托
-                            OperateResult value = basics.Function?.Invoke(basics.Address);
-                            //状态判断
-                            if (value.State)
+                            //数据必须大于0才进行
+                            if (basics.Address.AddressArray.Count > 0)
                             {
-                                //判断数据类型
-                                switch (value.RType)
-                                {
-                                    case @enum.ResultType.KeyValue:
-                                        ConcurrentDictionary<string, AddressValue>? RData = value.RData as ConcurrentDictionary<string, AddressValue>;
-                                        //做流程处理
-                                        if (basics.DataChangeOut)
-                                        {
-                                            //实例化对象
-                                            if (UpParam == null)
-                                            {
-                                                UpParam = new ConcurrentDictionary<string, AddressValue>();
-                                            }
-                                            //空数据不做处理
-                                            if (RData != null && RData.Count > 0)
-                                            {
-                                                //获取到第一次数据，准备节点检测数据是否变化
-                                                if (!ConcurrentDictionaryEquals(UpParam, RData))
-                                                {
-                                                    //当节点数据变化，有一项数据未变化，也把这未变项与变化项一同抛出，在特殊用途中，确保此批点位数据都存在
-                                                    if (basics.SameDataOut)
-                                                    {
-                                                        //抛出差异数据
-                                                        OnEventHandler?.Invoke(this, new EventResult(true, "存在变化数据", RData, value.RType));
-                                                    }
-                                                    else
-                                                    {
-                                                        //如果不一致，就找寻里面不一样的 键、值
-                                                        ConcurrentDictionary<string, AddressValue> data = new ConcurrentDictionary<string, AddressValue>(RData.Except(UpParam).ToDictionary(x => x.Key, x => x.Value));
-                                                        //抛出差异数据
-                                                        OnEventHandler?.Invoke(this, new EventResult(true, "变化数据", data, value.RType));
-                                                    }
-                                                    //把这次新数据 赋值到历史数据中
-                                                    UpParam = RData;
-                                                }
-                                            }
-                                        }
-                                        else
-                                        {
-                                            OnEventHandler?.Invoke(this, new EventResult(true, "实时数据", value.RData, value.RType));
-                                        }
-                                        break;
-                                    case @enum.ResultType.KeyValueArray:
-                                        List<ConcurrentDictionary<string, AddressValue>>? RDataArray = value.RData as List<ConcurrentDictionary<string, AddressValue>>;
-                                        //实例化对象
-                                        if (UpParamArray == null)
-                                        {
-                                            UpParamArray = new List<ConcurrentDictionary<string, AddressValue>>();
-                                        }
-                                        //空数据不做处理
-                                        if (RDataArray != null && RDataArray.Count > 0)
-                                        {
-                                            //获取到第一次数据，准备节点检测数据是否变化
-                                            if (!ConcurrentDictionaryEquals(UpParamArray, RDataArray))
-                                            {
-                                                foreach (var item in RDataArray)
-                                                {
-                                                    //抛出差异数据
-                                                    OnEventHandler?.Invoke(this, new EventResult(true, "存在变化数据", RDataArray, value.RType));
-                                                }
-                                                //把这次新数据 赋值到历史数据中
-                                                UpParamArray = RDataArray;
-                                            }
-                                        }
-                                        break;
-                                }
-                            }
-                            else
-                            {
-                                //当状态为false 说明读取失败，写入日志
-                                OnEventHandler?.Invoke(this, new EventResult(false, $"自定义订阅轮询异常：{value.Message}"));
+                                //执行委托
+                                OperateResult value = basics.Function?.Invoke(basics.Address);
+
+                                //数据入列
+                                DataQueue.Enqueue(new QueueData() { value = value});
                             }
                         }
                     }
+                    catch (Exception ex)
+                    {
+                        OnEventHandler?.Invoke(this, new EventResult(false, $"自定义订阅轮询异常：{ex.Message}"));
+                    }
+                    Thread.Sleep(basics.SleepTime);
                 }
-                catch (Exception ex)
-                {
-                    OnEventHandler?.Invoke(this, new EventResult(false, $"自定义订阅轮询异常：{ex.Message}"));
-                }
-                Thread.Sleep(basics.SleepTime);
-            }
+            });
         }
-
         /// <summary>
         /// 释放
         /// </summary>
@@ -331,12 +262,29 @@ namespace YSAI.Core.subscription
             Depart("On");
             try
             {
-                if (thread == null)
+                if (PollTaskHandle == null)
                 {
                     MonitorSwitch = new CancellationTokenSource();
-                    thread = new Thread(Polling);
-                    thread.IsBackground = true;
-                    thread.Start();
+
+                    //轮询执行动作任务
+                    PollTaskHandle = Polling();
+
+                    //当队列为空，初始化队列
+                    if (DataQueue == null)
+                    {
+                        DataQueue = new ConcurrentQueue<QueueData>();
+                    }
+
+                    //任务为空创建任务
+                    if (TaskArray == null)
+                    {
+                        TaskArray = new ConcurrentDictionary<Guid, Task>();
+                        //创建任务
+                        for (int i = 0; i < basics.TaskHandleCount; i++)
+                        {
+                            TaskArray.TryAdd(Guid.NewGuid(), TaskHandle());
+                        }
+                    }
                     return Break("On", true);
                 }
                 else
@@ -369,7 +317,25 @@ namespace YSAI.Core.subscription
                 //立马停止轮询
                 GoOn = false;
                 //线程结束
-                thread?.Interrupt();
+                PollTaskHandle.Dispose();
+
+                //任务清空
+                if (TaskArray != null)
+                {
+                    foreach (var item in TaskArray)
+                    {
+                        item.Value.Dispose();
+                    }
+                    TaskArray.Clear();
+                    TaskArray = null;
+                }
+                //队列清空
+                if (DataQueue != null)
+                {
+                    DataQueue.Clear();
+                    DataQueue = null;
+                }
+
                 return Break("Off", true);
             }
             catch (Exception ex)
@@ -377,5 +343,127 @@ namespace YSAI.Core.subscription
                 return Break("Off", false, ex.Message, Exc: ex);
             }
         }
+
+
+
+        /// <summary>
+        /// 队列数据
+        /// </summary>
+        private class QueueData
+        {
+            /// <summary>
+            /// 操作结果
+            /// </summary>
+            public OperateResult value { get; set; }
+        }
+        /// <summary>
+        /// 任务集合
+        /// </summary>
+        private ConcurrentDictionary<Guid, Task> TaskArray;
+        /// <summary>
+        /// 数据队列
+        /// </summary>
+        private ConcurrentQueue<QueueData> DataQueue;
+
+        /// <summary>
+        /// 任务处理
+        /// </summary>
+        /// <returns></returns>
+        private Task TaskHandle()
+        {
+            //起个新线程处理
+            return Task.Factory.StartNew(() =>
+            {
+                //循环
+                while (TaskArray != null)
+                {
+                    //队列数据
+                    QueueData? queueData;
+                    while (DataQueue.TryDequeue(out queueData))
+                    {
+                        if (queueData != null)
+                        {
+                            //状态判断
+                            if (queueData.value.State)
+                            {
+                                //判断数据类型
+                                switch (queueData.value.RType)
+                                {
+                                    case @enum.ResultType.KeyValue:
+                                        ConcurrentDictionary<string, AddressValue>? RData = queueData.value.RData as ConcurrentDictionary<string, AddressValue>;
+                                        //做流程处理
+                                        if (basics.DataChangeOut)
+                                        {
+                                            //实例化对象
+                                            if (UpParam == null)
+                                            {
+                                                UpParam = new ConcurrentDictionary<string, AddressValue>();
+                                            }
+                                            //空数据不做处理
+                                            if (RData != null && RData.Count > 0)
+                                            {
+                                                //获取到第一次数据，准备节点检测数据是否变化
+                                                if (!ConcurrentDictionaryEquals(UpParam, RData))
+                                                {
+                                                    //当节点数据变化，有一项数据未变化，也把这未变项与变化项一同抛出，在特殊用途中，确保此批点位数据都存在
+                                                    if (basics.SameDataOut)
+                                                    {
+                                                        //抛出差异数据
+                                                        OnEventHandler?.Invoke(this, new EventResult(true, "存在变化数据", RData, queueData.value.RType));
+                                                    }
+                                                    else
+                                                    {
+                                                        //如果不一致，就找寻里面不一样的 键、值
+                                                        ConcurrentDictionary<string, AddressValue> data = new ConcurrentDictionary<string, AddressValue>(RData.Except(UpParam).ToDictionary(x => x.Key, x => x.Value));
+                                                        //抛出差异数据
+                                                        OnEventHandler?.Invoke(this, new EventResult(true, "变化数据", data, queueData.value.RType));
+                                                    }
+                                                    //把这次新数据 赋值到历史数据中
+                                                    UpParam = RData;
+                                                }
+                                            }
+                                        }
+                                        else
+                                        {
+                                            OnEventHandler?.Invoke(this, new EventResult(true, "实时数据", queueData.value.RData, queueData.value.RType));
+                                        }
+                                        break;
+                                    case @enum.ResultType.KeyValueArray:
+                                        List<ConcurrentDictionary<string, AddressValue>>? RDataArray = queueData.value.RData as List<ConcurrentDictionary<string, AddressValue>>;
+                                        //实例化对象
+                                        if (UpParamArray == null)
+                                        {
+                                            UpParamArray = new List<ConcurrentDictionary<string, AddressValue>>();
+                                        }
+                                        //空数据不做处理
+                                        if (RDataArray != null && RDataArray.Count > 0)
+                                        {
+                                            //获取到第一次数据，准备节点检测数据是否变化
+                                            if (!ConcurrentDictionaryEquals(UpParamArray, RDataArray))
+                                            {
+                                                foreach (var item in RDataArray)
+                                                {
+                                                    //抛出差异数据
+                                                    OnEventHandler?.Invoke(this, new EventResult(true, "存在变化数据", RDataArray, queueData.value.RType));
+                                                }
+                                                //把这次新数据 赋值到历史数据中
+                                                UpParamArray = RDataArray;
+                                            }
+                                        }
+                                        break;
+                                }
+                            }
+                            else
+                            {
+                                //当状态为false 说明读取失败，写入日志
+                                OnEventHandler?.Invoke(this, new EventResult(false, $"自定义订阅轮询异常：{queueData.value.Message}"));
+                            }
+                        }
+                    }
+                    //队列里面的数据处理完休息一下
+                    Thread.Sleep(basics.TaskHandleAccomplishSleepTime);
+                }
+            });
+        }
     }
 }

src/YSAI.DAQ/YSAI.DAQ.sln

@@ -93,7 +93,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Tool.Core", "YSAI.Tool
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Tool", "YSAI.Tool\YSAI.Tool.csproj", "{3C333E39-520D-4183-8E3D-D0FF6628C15E}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YSAI.Document", "YSAI.Document\YSAI.Document.csproj", "{863DE76E-F597-4624-A6B4-860B97FA06D0}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Document", "YSAI.Document\YSAI.Document.csproj", "{863DE76E-F597-4624-A6B4-860B97FA06D0}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YSAI.Test.All", "YSAI.Test.All\YSAI.Test.All.csproj", "{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -245,6 +247,10 @@ Global
 		{863DE76E-F597-4624-A6B4-860B97FA06D0}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{863DE76E-F597-4624-A6B4-860B97FA06D0}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{863DE76E-F597-4624-A6B4-860B97FA06D0}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -291,6 +297,7 @@ Global
 		{E85BF19B-E671-4A1E-BC67-B545700B4BF1} = {1856E9E1-33C4-45C1-832C-854F9BE1ACC4}
 		{3C333E39-520D-4183-8E3D-D0FF6628C15E} = {1856E9E1-33C4-45C1-832C-854F9BE1ACC4}
 		{863DE76E-F597-4624-A6B4-860B97FA06D0} = {0A264424-1AD7-49FA-B813-D96498066479}
+		{3CF5DE43-8FBE-4C28-A955-FAA75BF77E0E} = {12CB0510-7B1E-4518-AA3B-412A4D323D42}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {5D5D3927-6714-40C0-84EA-44C5BA4C5E87}

src/YSAI.DAQ/YSAI.DB/YSAI.DB.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.3</Version>
+	  <Version>1.0.0.4</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -18,7 +18,7 @@
 		<PackageReference Include="System.Data.OracleClient" Version="1.0.8" />
 		<PackageReference Include="System.Data.SqlClient" Version="4.8.5" />
 		<PackageReference Include="System.Data.SQLite" Version="1.0.118" />
-		<PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.20" />
 	</ItemGroup>
 	<!--<ItemGroup>
 		<ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />

src/YSAI.DAQ/YSAI.DaqManage/YSAI.DaqManage.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 	<ItemGroup>
 		<FrameworkReference Include="Microsoft.AspNetCore.App" />
-		<PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.20" />
 	</ItemGroup>
 
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Kafka/KafkaOperate.cs

@@ -118,6 +118,10 @@ namespace YSAI.Kafka
                             //消息已处理提交偏移量
                             Consumer?.Commit(result);
                         }
+                        else
+                        {
+                            Thread.Sleep(100);
+                        }
                     }
                 }
                 catch (Exception ex)

src/YSAI.DAQ/YSAI.Kafka/YSAI.Kafka.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.4</Version>
+	  <Version>1.0.0.5</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -14,7 +14,7 @@
 
   <ItemGroup>
     <PackageReference Include="Confluent.Kafka" Version="2.2.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 	
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Lib/YSAI.Opc.Pack/YSAI.Opc.xml

@@ -1291,6 +1291,26 @@
             处理从重新连接处理程序完成的重新连接事件
             </summary>
         </member>
+        <member name="T:YSAI.Opc.ua.client.OpcUaClientOperate.QueueData">
+            <summary>
+            队列数据
+            </summary>
+        </member>
+        <member name="P:YSAI.Opc.ua.client.OpcUaClientOperate.QueueData.addressDetails">
+            <summary>
+            地址详情
+            </summary>
+        </member>
+        <member name="P:YSAI.Opc.ua.client.OpcUaClientOperate.QueueData.Value">
+            <summary>
+            这个点的值
+            </summary>
+        </member>
+        <member name="F:YSAI.Opc.ua.client.OpcUaClientOperate.DataQueue">
+            <summary>
+            数据队列
+            </summary>
+        </member>
         <member name="M:YSAI.Opc.ua.client.OpcUaClientOperate.OnMonitoredItemNotification(Opc.Ua.Client.MonitoredItem,Opc.Ua.Client.MonitoredItemNotificationEventArgs,YSAI.Core.data.AddressDetails)">
             <summary>
             当数据变化更新

src/YSAI.DAQ/YSAI.Modbus/YSAI.Modbus.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.3</Version>
+	  <Version>1.0.0.4</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -13,7 +13,7 @@
   </PropertyGroup>
 
 	<ItemGroup>
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Mqtt/YSAI.Mqtt.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.4</Version>
+	  <Version>1.0.0.5</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -15,7 +15,7 @@
   <ItemGroup>
     <PackageReference Include="MQTTnet" Version="4.3.1.873" />
     <PackageReference Include="MQTTnet.AspNetCore" Version="4.3.1.873" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Opc/YSAI.Opc.csproj

@@ -5,29 +5,15 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.3</Version>
+	  <Version>1.0.0.4</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
 	  <GenerateDocumentationFile>True</GenerateDocumentationFile>
   </PropertyGroup>
-
-  <ItemGroup>
-    <None Remove="Opc.Ua.Client.Config.xml" />
-    <None Remove="Opc.Ua.Service.Config.xml" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <Content Include="Opc.Ua.Client.Config.xml">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-    </Content>
-    <Content Include="Opc.Ua.Service.Config.xml">
-      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
-    </Content>
-  </ItemGroup>
   <ItemGroup>
     <PackageReference Include="OPCFoundation.NetStandard.Opc.Ua" Version="1.4.372.56" />
-	<PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+	<PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
 
@@ -40,5 +26,14 @@
     <Folder Include="Properties\" />
   </ItemGroup>
 
+  <ItemGroup>
+    <None Update="Opc.Ua.Client.Config.xml">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Update="Opc.Ua.Service.Config.xml">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
 
 </Project>

src/YSAI.DAQ/YSAI.Opc/da/client/OpcDaClientData.cs

@@ -44,6 +44,16 @@ namespace YSAI.Opc.da.client
             [Description("更新频率")]
             public int UpdateRate { get; set; } = 100;
             /// <summary>
+            /// 任务处理数量
+            /// </summary>
+            [Description("任务处理数量")]
+            public int TaskHandleCount { get; set; } = 10;
+            /// <summary>
+            /// 任务处理完成休眠时间(毫秒)
+            /// </summary>
+            [Description("任务处理完成休眠时间(毫秒)")]
+            public int TaskHandleAccomplishSleepTime { get; set; } = 1000;
+            /// <summary>
             /// 重写Equals
             /// </summary>
             /// <param name="obj"></param>

src/YSAI.DAQ/YSAI.Opc/da/client/OpcDaClientOperate.cs

@@ -1,5 +1,6 @@
 using Microsoft.AspNetCore.DataProtection.KeyManagement;
 using Newtonsoft.Json.Linq;
+using Opc.Ua;
 using OpcDaNetApi.Da;
 using System.Collections.Concurrent;
 using System.Reflection;
@@ -104,32 +105,8 @@ namespace YSAI.Opc.da.client
         /// <exception cref="NotImplementedException"></exception>
         private void Subscription_DataChanged(object subscriptionHandle, object requestHandle, ItemValueResult[] values, string GroupName)
         {
-            try
-            {
-                foreach (var item in values)
-                {
-                    AddressDetails? addressDetails = GetAddressData(GroupName, item.ItemName);
-                    if (addressDetails != null)
-                    {
-                        //设置参数
-                        ConcurrentDictionary<string, AddressValue> param = new ConcurrentDictionary<string, AddressValue>();
-
-                        //处理数据
-                        AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(addressDetails, item.Value.ToString());  //数据 
-
-                        //添加至集合
-                        param.AddOrUpdate(addressDetails.AddressName, addressValue, (k, v) => addressValue);
-
-                        //响应
-                        OnEventHandler?.Invoke(this, new EventResult(true, "[ 订阅通知 ]点位数据更新", param,Core.@enum.ResultType.KeyValue));
-                    }
-                }
-            }
-            catch (Exception ex)
-            {
-                //响应
-                OnEventHandler?.Invoke(this, new EventResult(false, $"点位数据更新异常：{ex.Message}"));
-            }
+            //数据入列
+            DataQueue.Enqueue(new QueueData() { values = values, GroupName = GroupName });
         }
         /// <summary>
         /// 添加地址数据
@@ -596,6 +573,85 @@ namespace YSAI.Opc.da.client
             return Task.Run(() => RemoveNode(NodeName, GroupName));
         }
 
+
+
+        /// <summary>
+        /// 队列数据
+        /// </summary>
+        private class QueueData
+        {
+            /// <summary>
+            /// 地址详情
+            /// </summary>
+            public ItemValueResult[] values { get; set; }
+            /// <summary>
+            /// 组名
+            /// </summary>
+            public string GroupName { get; set; }
+        }
+        /// <summary>
+        /// 任务集合
+        /// </summary>
+        private ConcurrentDictionary<Guid, Task> TaskArray;
+        /// <summary>
+        /// 数据队列
+        /// </summary>
+        private ConcurrentQueue<QueueData> DataQueue;
+
+        /// <summary>
+        /// 任务处理
+        /// </summary>
+        /// <returns></returns>
+        private Task TaskHandle()
+        {
+            //起个新线程处理
+            return Task.Factory.StartNew(() =>
+            {
+                //循环
+                while (TaskArray != null)
+                {
+                    //队列数据
+                    QueueData? queueData;
+                    while (DataQueue.TryDequeue(out queueData))
+                    {
+                        if (queueData != null)
+                        {
+                            try
+                            {
+                                foreach (var item in queueData.values)
+                                {
+                                    AddressDetails? addressDetails = GetAddressData(queueData.GroupName, item.ItemName);
+                                    if (addressDetails != null)
+                                    {
+                                        //设置参数
+                                        ConcurrentDictionary<string, AddressValue> param = new ConcurrentDictionary<string, AddressValue>();
+
+                                        //处理数据
+                                        AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(addressDetails, item.Value.ToString());  //数据 
+
+                                        //添加至集合
+                                        param.AddOrUpdate(addressDetails.AddressName, addressValue, (k, v) => addressValue);
+
+                                        //响应
+                                        OnEventHandler?.Invoke(this, new EventResult(true, "[ 订阅通知 ]点位数据更新", param, Core.@enum.ResultType.KeyValue));
+                                    }
+                                }
+                            }
+                            catch (Exception ex)
+                            {
+                                //响应
+                                OnEventHandler?.Invoke(this, new EventResult(false, $"点位数据更新异常：{ex.Message}"));
+                            }
+                        }
+                    }
+                    //队列里面的数据处理完休息一下
+                    Thread.Sleep(basics.TaskHandleAccomplishSleepTime);
+                }
+            });
+        }
+
+
+
         public Task<OperateResult> OnAsync()
         {
             return Task.Run(()=> On());
@@ -636,6 +692,24 @@ namespace YSAI.Opc.da.client
                                 //判断连接状态
                                 if (opcDaClient.IsConnected)
                                 {
+
+                                    //当队列为空，初始化队列
+                                    if (DataQueue == null)
+                                    {
+                                        DataQueue = new ConcurrentQueue<QueueData>();
+                                    }
+
+                                    //任务为空创建任务
+                                    if (TaskArray == null)
+                                    {
+                                        TaskArray = new ConcurrentDictionary<Guid, Task>();
+                                        //创建任务
+                                        for (int i = 0; i < basics.TaskHandleCount; i++)
+                                        {
+                                            TaskArray.TryAdd(Guid.NewGuid(), TaskHandle());
+                                        }
+                                    }
+
                                     return Break("On", true);
                                 }
                             }
@@ -667,6 +741,25 @@ namespace YSAI.Opc.da.client
             {
                 if (opcDaClient != null && opcDaClient.IsConnected)
                 {
+
+                    //任务清空
+                    if (TaskArray != null)
+                    {
+                        foreach (var item in TaskArray)
+                        {
+                            item.Value.Dispose();
+                        }
+                        TaskArray.Clear();
+                        TaskArray = null;
+                    }
+                    //队列清空
+                    if (DataQueue != null)
+                    {
+                        DataQueue.Clear();
+                        DataQueue = null;
+                    }
+
+
                     opcDaClient.Dispose();
                     opcDaClient = null;
                 }

src/YSAI.DAQ/YSAI.Opc/ua/client/OpcUaClientData.cs

@@ -74,7 +74,18 @@ namespace YSAI.Opc.ua.client
             /// 发布时间间隔
             /// </summary>
             [Description("发布时间间隔")]
-            public byte PublishingInterval { get; set; } = 20;
+            public int PublishingInterval { get; set; } = 1000;
+
+            /// <summary>
+            /// 任务处理数量
+            /// </summary>
+            [Description("任务处理数量")]
+            public int TaskHandleCount { get; set; } = 10;
+            /// <summary>
+            /// 任务处理完成休眠时间(毫秒)
+            /// </summary>
+            [Description("任务处理完成休眠时间(毫秒)")]
+            public int TaskHandleAccomplishSleepTime { get; set; } = 1000;
 
             /// <summary>
             /// 重写基类中的Equals方法

src/YSAI.DAQ/YSAI.Opc/ua/client/OpcUaClientOperate.cs

@@ -16,7 +16,9 @@ using YSAI.Core.virtualAddress;
 using YSAI.Log;
 using YSAI.Opc.ua.client.unility;
 using YSAI.Unility;
+using static Opc.Ua.Client.SessionReconnectHandler;
 using static YSAI.Opc.ua.client.OpcUaClientData;
+using static YSAI.Opc.ua.client.OpcUaClientOperate;
 
 namespace YSAI.Opc.ua.client
 {
@@ -183,6 +185,7 @@ namespace YSAI.Opc.ua.client
                     OnEventHandler?.Invoke(this, new EventResult(false, Steps.事件对象不匹配.ToString()));
                     return;
                 }
+
                 clientSession = sessionReconnectHandler.Session;
                 sessionReconnectHandler.Dispose();
                 sessionReconnectHandler = null;
@@ -193,6 +196,10 @@ namespace YSAI.Opc.ua.client
                 OnEventHandler?.Invoke(this, new EventResult(false, $"处理重新连接处理程序完成的重新连接事件异常:{ex.Message}"));
             }
         }
+
+        
+        
+
         /// <summary>
         /// 当数据变化更新
         /// </summary>
@@ -201,24 +208,8 @@ namespace YSAI.Opc.ua.client
         /// <exception cref="Exception"></exception>
         private void OnMonitoredItemNotification(MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e, AddressDetails addressDetails)
         {
-            try
-            {
-                //对象转换
-                MonitoredItemNotification? notification = e.NotificationValue as MonitoredItemNotification;
-                //设置参数
-                ConcurrentDictionary<string, AddressValue> param = new ConcurrentDictionary<string, AddressValue>();
-                //处理数据
-                AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(addressDetails, notification.Value.ToString());  //数据 
-
-                param.AddOrUpdate(addressDetails.AddressName, addressValue, (k, v) => addressValue);
-
-                //响应
-                OnEventHandler?.Invoke(this, new EventResult(true, $"[ {Steps.订阅通知} ]点位数据更新", param,Core.@enum.ResultType.KeyValue));
-            }
-            catch (Exception ex)
-            {
-                OnEventHandler?.Invoke(this, new EventResult(false,$"[ {Steps.订阅通知} ]异常:{ex.Message}"));
-            }
+            //数据入列
+            DataQueue.Enqueue(new QueueData() { addressDetails = addressDetails, NotificationValue = e.NotificationValue });
         }
 
 
@@ -812,8 +803,8 @@ namespace YSAI.Opc.ua.client
                             Subscription subscription = new Subscription(clientSession?.DefaultSubscription);
                             subscription.PublishingEnabled = true;//是否启用发布
                             subscription.PublishingInterval = basics.PublishingInterval; //出版间隔
-                            subscription.KeepAliveCount = 5;  //存活数量
-                            subscription.LifetimeCount = 10;  //生命周期
+                            subscription.KeepAliveCount = uint.MaxValue;  //保活计数
+                            subscription.LifetimeCount = uint.MaxValue;  //寿命计数
                             subscription.MaxNotificationsPerPublish = uint.MaxValue;  //每个发布请求的最大通知数
                             subscription.Priority = 1;  //分配给订阅的优先级
                             subscription.DisplayName = Tag;  //订阅的显示名称
@@ -983,6 +974,74 @@ namespace YSAI.Opc.ua.client
             return Task.Run(() => On());
         }
 
+        /// <summary>
+        /// 队列数据
+        /// </summary>
+        private class QueueData
+        {
+            /// <summary>
+            /// 地址详情
+            /// </summary>
+            public AddressDetails addressDetails { get; set; }
+            /// <summary>
+            /// 通知的数据
+            /// </summary>
+            public IEncodeable NotificationValue { get; set; }
+        }
+        /// <summary>
+        /// 任务集合
+        /// </summary>
+        private ConcurrentDictionary<Guid, Task> TaskArray;
+        /// <summary>
+        /// 数据队列
+        /// </summary>
+        private ConcurrentQueue<QueueData> DataQueue;
+
+        /// <summary>
+        /// 任务处理
+        /// </summary>
+        /// <returns></returns>
+        private Task TaskHandle()
+        {
+            //起个新线程处理
+            return Task.Factory.StartNew(() =>
+            {
+                //循环
+                while (TaskArray != null)
+                {
+                    //队列数据
+                    QueueData? queueData;
+                    while (DataQueue.TryDequeue(out queueData))
+                    {
+                        if (queueData != null)
+                        {
+                            try
+                            {
+                                //对象转换
+                                MonitoredItemNotification? notification = queueData.NotificationValue as MonitoredItemNotification;
+                                //设置参数
+                                ConcurrentDictionary<string, AddressValue> param = new ConcurrentDictionary<string, AddressValue>();
+                                //处理数据
+                                AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(queueData.addressDetails, notification.Value.ToString());  //数据 
+
+                                param.AddOrUpdate(queueData.addressDetails.AddressName, addressValue, (k, v) => addressValue);
+
+                                //响应
+                                OnEventHandler?.Invoke(this, new EventResult(true, $"[ {Steps.订阅通知} ]点位数据更新", param, Core.@enum.ResultType.KeyValue));
+                            }
+                            catch (Exception ex)
+                            {
+                                OnEventHandler?.Invoke(this, new EventResult(false, $"[ {Steps.订阅通知} ]异常:{ex.Message}"));
+                            }
+                        }
+                    }
+                    //队列里面的数据处理完休息一下
+                    Thread.Sleep(basics.TaskHandleAccomplishSleepTime);
+                }
+            });
+        }
+
+
         public OperateResult On()
         {
             //开始记录运行时间
@@ -1049,6 +1108,13 @@ namespace YSAI.Opc.ua.client
                 AC.ApplicationName = basics.Name;
                 AC.ApplicationType = ApplicationType.Client;
                 AC.CertificateValidator = certificateValidator;
+                AC.ServerConfiguration=new ServerConfiguration
+                {
+                    MaxSubscriptionCount = int.MaxValue,
+                    MaxMessageQueueSize = int.MaxValue,
+                    MaxNotificationQueueSize = int.MaxValue,
+                    MaxPublishRequestCount = int.MaxValue
+                };
 
                 //查找与当前设置最匹配的端点
                 EndpointDescription endpointDescription = CoreClientUtils.SelectEndpoint(basics.ServerUrl, false);
@@ -1076,6 +1142,24 @@ namespace YSAI.Opc.ua.client
                 //当从服务器到达一个存活状态或检测到一个错误时将被引发
                 clientSession.KeepAlive += new KeepAliveEventHandler(Session_KeepAlive);
 
+                
+                //当队列为空，初始化队列
+                if (DataQueue == null)
+                {
+                    DataQueue = new ConcurrentQueue<QueueData>();
+                }
+
+                //任务为空创建任务
+                if (TaskArray == null)
+                {
+                    TaskArray = new ConcurrentDictionary<Guid, Task>();
+                    //创建任务
+                    for (int i = 0; i < basics.TaskHandleCount; i++)
+                    {
+                        TaskArray.TryAdd(Guid.NewGuid(), TaskHandle());
+                    }
+                }
+
                 //设置连接状态
                 IsConnected = true;
 
@@ -1115,6 +1199,24 @@ namespace YSAI.Opc.ua.client
                     clientSession.Dispose();
                     clientSession = null;
                 }
+
+                //任务清空
+                if (TaskArray != null)
+                {
+                    foreach (var item in TaskArray)
+                    {
+                        item.Value.Dispose();
+                    }
+                    TaskArray.Clear();
+                    TaskArray = null;
+                }
+                //队列清空
+                if (DataQueue != null)
+                {
+                    DataQueue.Clear();
+                    DataQueue = null;
+                }
+
                 IsConnected = false;
 
                 return Break("Off", true);

src/YSAI.DAQ/YSAI.RabbitMQ/YSAI.RabbitMQ.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.4</Version>
+	  <Version>1.0.0.5</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -14,7 +14,7 @@
 
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="6.5.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 	<!--<ItemGroup>
 		<ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />

src/YSAI.DAQ/YSAI.RelayManage/YSAI.RelayManage.csproj

@@ -9,7 +9,7 @@
 	<ItemGroup>
 		<PackageReference Include="RabbitMQ.Client" Version="6.5.0" />
 		<PackageReference Include="Confluent.Kafka" Version="2.2.0" />
-		<PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.20" />
 		<FrameworkReference Include="Microsoft.AspNetCore.App" />
 	</ItemGroup>
 

src/YSAI.DAQ/YSAI.S7/YSAI.S7.csproj

@@ -5,7 +5,7 @@
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
 	  <GeneratePackageOnBuild>True</GeneratePackageOnBuild>
-	  <Version>1.0.0.3</Version>
+	  <Version>1.0.0.4</Version>
 	  <Authors>Shun</Authors>
 	  <Company>YSAI</Company>
 	  <Product>SCADA</Product>
@@ -13,7 +13,7 @@
   </PropertyGroup>
 
 	<ItemGroup>
-    <PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Test.All/Program.cs

@@ -0,0 +1,89 @@
+using Newtonsoft.Json.Linq;
+using System.Collections.Concurrent;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using YSAI.Core.subscription;
+using YSAI.Unility;
+
+namespace YSAI.Test.All
+{
+    internal class Program
+    {
+        static void Main(string[] args)
+        {
+            //地址参数
+            Address address = new Address()
+            {
+                SN = Guid.NewGuid().ToString(),
+                AddressArray = new List<AddressDetails>()
+            };
+            for (int i = 0; i < 1000000; i++)
+            {
+                address.AddressArray.Add(new AddressDetails
+                {
+                    SN = $"测试{i}号采集",
+                    AddressName = $"TEST{i}"
+                });
+            }
+           
+
+
+
+            TEST tEST = new TEST();
+
+            SubscribeOperate subscribeOperate = SubscribeOperate.Instance(new SubscribeData.Basics()
+            {
+                Address = address,
+                Function = tEST.Read,
+            });
+            subscribeOperate.OnEvent += SubscribeOperate_OnEvent;
+            OperateResult operateResult = subscribeOperate.On();
+
+
+
+            while (true)
+            {
+                Thread.Sleep(1000);
+            }
+        }
+
+        private static void SubscribeOperate_OnEvent(object? sender, EventResult e)
+        {
+
+            ConcurrentDictionary<string, AddressValue> param =  e.RData as ConcurrentDictionary<string, AddressValue>;
+            foreach (var item in param)
+            {
+                Console.WriteLine(item.Value.AddressName);
+                Console.WriteLine(item.Value.Value);
+                Console.WriteLine("---------------------");
+            }
+        }
+
+        public class TEST: IBaseAbstract
+        {
+            protected override string LogHead => "[ TEST 操作 ]";
+            protected override string ClassName => "TEST";
+            public OperateResult Read(Address address)
+            {
+                Depart("Read");
+                //节点数据
+                ConcurrentDictionary<string, AddressValue> param = new ConcurrentDictionary<string, AddressValue>();
+                foreach (var item in address.AddressArray) 
+                {
+                    //数据处理
+                    AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(item, new Random().NextDouble().ToString());
+
+                    //AddressValue addressValue = YSAI.Core.data.AddressHandler.ExecuteDispose(item, "1");
+
+                    //数据添加
+                    param.AddOrUpdate(item.AddressName, addressValue, (k, v) => addressValue);
+                }
+                return Break("Read", true, RData: param, RType: Core.@enum.ResultType.KeyValue);
+            }
+
+
+
+
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Test.All/YSAI.Test.All.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net6.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
+  </ItemGroup>
+
+</Project>

src/YSAI.DAQ/YSAI.Test/YSAI.Test.csproj

@@ -17,7 +17,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-	<PackageReference Include="YSAI.Core" Version="1.0.0.19" />
+	<PackageReference Include="YSAI.Core" Version="1.0.0.20" />
   </ItemGroup>
 
   <ItemGroup>