1. 新增json字符串判断

2. 修改opcua当断线重连自动订阅
3. 修改传输协议的数据格式
4. 新增新的传输协议（目前存在内存飙升问题）
5. 版本更新

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

@@ -17,7 +17,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </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.58</Version>
+    <Version>1.0.0.60</Version>
     <Authors>Shun</Authors>
     <Company>YSAI</Company>
     <Product>SCADA</Product>
@@ -17,7 +17,7 @@
 		<PackageReference Include="Microsoft.ClearScript" Version="7.4.4" />
 		<PackageReference Include="System.IO.Ports" Version="7.0.0" />
 		<PackageReference Include="YSAI.Log" Version="1.0.0.9" />
-		<PackageReference Include="YSAI.Unility" Version="1.0.0.17" />
+		<PackageReference Include="YSAI.Unility" Version="1.0.0.19" />
 	</ItemGroup>
 
 </Project>

src/YSAI.DAQ/YSAI.DAQ.sln

@@ -93,7 +93,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Test.All", "YSAI.Test.
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Pack.Param", "YSAI.Pack.Param\YSAI.Pack.Param.csproj", "{36696247-658B-445E-9C05-A815B73F248A}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YSAI.Netty", "YSAI.Netty\YSAI.Netty.csproj", "{D757285D-0B7C-47C4-BC54-744B6E2A7E61}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Netty", "YSAI.Netty\YSAI.Netty.csproj", "{D757285D-0B7C-47C4-BC54-744B6E2A7E61}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution

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

@@ -13,12 +13,12 @@
   </PropertyGroup>
 
 	<ItemGroup>
-		<PackageReference Include="Dapper" Version="2.1.11" />
+		<PackageReference Include="Dapper" Version="2.1.15" />
 		<PackageReference Include="MySql.Data" Version="8.1.0" />
 		<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.58" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.60" />
 	</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.58" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.60" />
 	</ItemGroup>
 
 	<!--<ItemGroup>

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

@@ -112,10 +112,17 @@ namespace YSAI.Kafka
                         if (result != null)
                         {
                             string Topic = result.Topic;
-                            object Content = result.Message.Value;
+                            string Content = result.Message.Value;
                             string Key = result.Message.Key;
                             long Offset = result.TopicPartitionOffset.Offset.Value;
-                            OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} ) 键 ( {Key} ) 偏移量 ( {Offset} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content},\"Key\":\"{Key}\",\"Offset\":\"{Offset}\"" + "}", Core.@enum.ResultType.Dynamic));
+                            if (Content.IsJson())
+                            {
+                                OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} ) 键 ( {Key} ) 偏移量 ( {Offset} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content},\"Key\":\"{Key}\",\"Offset\":\"{Offset}\"" + "}", Core.@enum.ResultType.Dynamic));
+                            }
+                            else
+                            {
+                                OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} ) 键 ( {Key} ) 偏移量 ( {Offset} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":\"{Content}\",\"Key\":\"{Key}\",\"Offset\":\"{Offset}\"" + "}", Core.@enum.ResultType.Dynamic));
+                            }
                             //消息已处理提交偏移量
                             Consumer?.Commit(result);
                         }

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

@@ -14,7 +14,7 @@
 
   <ItemGroup>
     <PackageReference Include="Confluent.Kafka" Version="2.2.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 	
 	<!--<ItemGroup>

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

@@ -15,7 +15,7 @@
 	<ItemGroup>
     <PackageReference Include="NModbus" Version="3.0.80" />
     <PackageReference Include="NModbus.Serial" Version="3.0.80" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
 	<!--<ItemGroup>

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

@@ -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.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
 	<!--<ItemGroup>

src/YSAI.DAQ/YSAI.Mqtt/client/MqttClientOperate.cs

@@ -89,13 +89,22 @@ namespace YSAI.Mqtt.client
         /// <summary>
         /// 消息事件
         /// </summary>
-        private async Task MqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
+        private Task MqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
         {
-            string Topic = arg.ApplicationMessage.Topic;
-            object Content = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
-            string ClientID = arg.ClientId;
-            OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content},\"ClientID\":\"{ClientID}\"" + "}", Core.@enum.ResultType.Dynamic));
-            await Task.CompletedTask;
+            return Task.Run(() =>
+            {
+                string Topic = arg.ApplicationMessage.Topic;
+                string Content = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
+                string ClientID = arg.ClientId;
+                if (Content.IsJson())
+                {
+                    OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content},\"ClientID\":\"{ClientID}\"" + "}", Core.@enum.ResultType.Dynamic));
+                }
+                else
+                {
+                    OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":\"{Content}\",\"ClientID\":\"{ClientID}\"" + "}", Core.@enum.ResultType.Dynamic));
+                }
+            });
         }
 
 

src/YSAI.DAQ/YSAI.Netty/YSAI.Netty.csproj

@@ -1,9 +1,19 @@
-<Project Sdk="Microsoft.NET.Sdk">
-
-  <PropertyGroup>
-    <TargetFramework>net6.0</TargetFramework>
-    <ImplicitUsings>enable</ImplicitUsings>
-    <Nullable>enable</Nullable>
-  </PropertyGroup>
+<Project Sdk="Microsoft.NET.Sdk">
 
+	<PropertyGroup>
+		<TargetFramework>net6.0</TargetFramework>
+		<ImplicitUsings>enable</ImplicitUsings>
+		<Nullable>enable</Nullable>
+		<GeneratePackageOnBuild>True</GeneratePackageOnBuild>
+		<Version>1.0.0.1</Version>
+		<Authors>Shun</Authors>
+		<Company>YSAI</Company>
+		<Product>SCADA</Product>
+		<GenerateDocumentationFile>True</GenerateDocumentationFile>
+	</PropertyGroup>
+	<ItemGroup>
+		<PackageReference Include="DotNetty.Handlers" Version="0.7.5" />
+		<PackageReference Include="DotNetty.Transport" Version="0.7.5" />
+		<PackageReference Include="YSAI.Core" Version="1.0.0.60" />
+	</ItemGroup>
 </Project>

src/YSAI.DAQ/YSAI.Netty/client/NettyClientData.cs

@@ -0,0 +1,49 @@
+using System;
+using System.Collections.Generic;
+using System.ComponentModel;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Unility;
+
+namespace YSAI.Netty.client
+{
+    public class NettyClientData
+    {
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        public class Basics
+        {
+            /// <summary>
+            /// 唯一标识符
+            /// </summary>
+            [Description("唯一标识符")]
+            public string? SN { get; set; } = Guid.NewGuid().ToNString();
+
+            /// <summary>
+            /// 主机
+            /// </summary>
+            [Description("主机")]
+            public string Host { get; set; }
+
+            /// <summary>
+            /// 端口
+            /// </summary>
+            [Description("端口")]
+            public int Port { get; set; }
+
+            /// <summary>
+            /// SSL文件路径
+            /// </summary>
+            [Description("SSL文件路径")]
+            public string? SslFilePath { get; set; }
+
+            /// <summary>
+            /// SSL文件密码
+            /// </summary>
+            [Description("SSL文件密码")]
+            public string? SslFilePassword { get; set; }
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Netty/client/NettyClientOperate.cs

@@ -0,0 +1,421 @@
+using DotNetty.Codecs;
+using DotNetty.Handlers.Tls;
+using DotNetty.Transport.Bootstrapping;
+using DotNetty.Transport.Channels;
+using DotNetty.Transport.Channels.Sockets;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using System.Collections.Concurrent;
+using System.Net;
+using System.Net.Security;
+using System.Security.Cryptography.X509Certificates;
+using System.Text;
+using System.Threading.Channels;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using YSAI.Log;
+using YSAI.Unility;
+using static YSAI.Netty.client.NettyClientOperate.SecureChatClientHandler;
+
+namespace YSAI.Netty.client
+{
+    public class NettyClientOperate : IBaseAbstract, IRelay
+    {
+        protected override string TAG => "NettyClientOperate";
+
+        /// <summary>
+        /// 锁
+        /// </summary>
+        private static readonly object Lock = new object();
+        /// <summary>
+        /// 自身对象集合
+        /// </summary>
+        private static List<NettyClientOperate> ThisObjList = new List<NettyClientOperate>();
+
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static NettyClientOperate Instance(NettyClientData.Basics basics)
+        {
+            if (ThisObjList.Count >= MaxInstanceCount)
+            {
+                throw new Exception(ExceedMaxInstanceCountTips);
+            }
+            NettyClientOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Comparer(basics).result);
+            if (exp == null)
+            {
+                lock (Lock)
+                {
+                    if (ThisObjList.Count(c => c.basics.Comparer(basics).result) > 0)
+                    {
+                        return ThisObjList.First(c => c.basics.Comparer(basics).result);
+                    }
+                    else
+                    {
+                        NettyClientOperate exp2 = new NettyClientOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private NettyClientData.Basics basics;
+
+        /// <summary>
+        /// 通信库
+        /// </summary>
+        private Bootstrap Communication;
+        /// <summary>
+        /// 组
+        /// </summary>
+        private MultithreadEventLoopGroup Group;
+        /// <summary>
+        /// 通道
+        /// </summary>
+        private IChannel Channel;
+
+        /// <summary>
+        /// 订阅的主题项容器
+        /// </summary>
+        private ConcurrentDictionary<string, DateTime> SubIoc = new ConcurrentDictionary<string, DateTime>();
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        public NettyClientOperate(NettyClientData.Basics basics)
+        {
+            this.basics = basics;
+        }
+        /// <summary>
+        /// 释放
+        /// </summary>
+        /// <exception cref="NotImplementedException"></exception>
+        public void Dispose()
+        {
+            Off();
+            GC.Collect();
+            GC.SuppressFinalize(this);
+            ThisObjList.Remove(this);
+        }
+
+        public OperateResult GetStatus()
+        {
+            if (Channel == null)
+            {
+                return Break("GetStatus", false, "未连接");
+            }
+            return Break("GetStatus", true, "已连接");
+        }
+
+        public Task<OperateResult> GetStatusAsync()
+        {
+            return Task.Run(() => GetStatus());
+        }
+
+        public OperateResult Off()
+        {
+            Depart("Off");
+            try
+            {
+                if (Channel == null)
+                {
+                    return Break("Off", false, "未连接");
+                }
+                Communication = null;
+                Group?.ShutdownGracefullyAsync().Wait();
+                Channel.CloseAsync().Wait();
+                Channel.DisconnectAsync().Wait();
+                Channel = null;
+                return Break("Off", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("Off", false, ex.Message, Exception: ex);
+            }
+        }
+
+        public Task<OperateResult> OffAsync()
+        {
+            return Task.Run(() => Off());
+        }
+
+        /// <summary>
+        /// 内部消息
+        /// </summary>
+        private class M
+        {
+            /// <summary>
+            /// 主题
+            /// </summary>
+            public string T { get; set; }
+            /// <summary>
+            /// 内容
+            /// </summary>
+            public string C {  get; set; }
+        }
+
+        /// <summary>
+        /// 内部消息接收
+        /// </summary>
+        /// <param name="sender"></param>
+        /// <param name="e"></param>
+        protected void OnMe(object? sender, EventResult e)
+        {
+            if (e.State)
+            {
+                //收到数据
+                if (SubIoc.Count > 0)
+                {
+                    Info? info = e.RData as Info;
+                    if (info != null && !string.IsNullOrEmpty(info.Msg))
+                    {
+                        string Con = info.Msg;
+                        string? Terminal = info.Context.Channel.RemoteAddress.ToString();
+                        if (Con.IsJson())
+                        {
+                            M? msg = Con.JsonToObject<M>();
+                            if (SubIoc.ContainsKey(msg.T))
+                            {
+                                string Topic = msg.T;
+                                string Content = msg.C;
+
+                                if (Content.IsJson())
+                                {
+                                    OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content},\"Terminal\":\"{Terminal}\"" + "}", Core.@enum.ResultType.Dynamic));
+                                }
+                                else
+                                {
+                                    OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":\"{Content}\",\"Terminal\":\"{Terminal}\"" + "}", Core.@enum.ResultType.Dynamic));
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            else
+            {
+                //出现异常
+                OnEventHandler(this, new EventResult(false, $"订阅异常：{e.Message}"));
+                Off();
+            }
+        }
+
+        /// <summary>
+        /// 消息接收处理
+        /// </summary>
+        public class SecureChatClientHandler : SimpleChannelInboundHandler<string>
+        {
+            /// <summary>
+            /// 消息事件
+            /// </summary>
+            Action<object?, EventResult> MessageEvent;
+            /// <summary>
+            /// 构造函数
+            /// </summary>
+            /// <param name="MessageEvent">息事件</param>
+            public SecureChatClientHandler(Action<object?, EventResult> MessageEvent)
+            {
+                this.MessageEvent = MessageEvent;
+            }
+
+            /// <summary>
+            /// 信息
+            /// </summary>
+            public class Info
+            {
+                /// <summary>
+                /// 通道处理程序上下文
+                /// </summary>
+                public IChannelHandlerContext Context { get; set; }
+                /// <summary>
+                /// 信息
+                /// </summary>
+                public string Msg { get; set; }
+            }
+
+            /// <summary>
+            /// 通道消息
+            /// </summary>
+            /// <param name="contex">通道处理程序上下文</param>
+            /// <param name="msg">消息</param>
+            protected override void ChannelRead0(IChannelHandlerContext contex, string msg)
+            {
+                //消息抛出
+                MessageEvent?.Invoke(this, new EventResult(true, msg, new Info { Context = contex, Msg = msg }, Core.@enum.ResultType.Dynamic));
+            }
+            /// <summary>
+            /// 出现了异常
+            /// </summary>
+            /// <param name="contex">通道处理程序上下文</param>
+            /// <param name="e">异常信息</param>
+            public override void ExceptionCaught(IChannelHandlerContext contex, Exception ex)
+            {
+                //消息抛出
+                MessageEvent?.Invoke(this, new EventResult(false, ex.Message));
+
+                //通道关闭
+                contex.CloseAsync();
+            }
+        }
+
+        public OperateResult On()
+        {
+            Depart("On");
+            try
+            {
+                if (Channel != null)
+                {
+                    return Break("On", false, "已连接");
+                }
+                //证书
+                X509Certificate2? Cert = null;
+                //证书目标主机
+                string? CertTargetHost = string.Empty;
+                //如果当证书不为空
+                if (!string.IsNullOrWhiteSpace(basics.SslFilePath) && !string.IsNullOrWhiteSpace(basics.SslFilePassword))
+                {
+                    Cert = new X509Certificate2(basics.SslFilePath, basics.SslFilePassword);
+                    CertTargetHost = Cert.GetNameInfo(X509NameType.DnsName, false);
+                }
+
+                //实例化通信库
+                Communication = new Bootstrap();
+                //实例化组
+                Group = new MultithreadEventLoopGroup();
+                //走TCP通道
+                Communication
+                .Group(Group)  //创建一个组
+                .Channel<TcpSocketChannel>()  //创建一个通道
+                .Option(ChannelOption.TcpNodelay, true)   //TCP节点布局
+                .Handler(new ActionChannelInitializer<ISocketChannel>(channel => { //处理
+
+                    IChannelPipeline pipeline = channel.Pipeline;
+                    if (Cert != null)
+                    {
+                        pipeline.AddLast(new TlsHandler(stream => new SslStream(stream, true, (sender, certificate, chain, errors) => true), new ClientTlsSettings(CertTargetHost)));
+                    }
+                    pipeline.AddLast(new DelimiterBasedFrameDecoder(8192, Delimiters.LineDelimiter()));
+                    //这里就是启动订阅
+                    pipeline.AddLast(new StringEncoder(), new StringDecoder(), new SecureChatClientHandler(OnMe));
+
+                }));
+
+                //连接
+                Channel = Communication.ConnectAsync(new IPEndPoint(IPAddress.Parse(basics.Host), basics.Port)).Result;
+
+                return Break("On", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("On", false, ex.Message, Exception: ex);
+            }
+        }
+
+        public Task<OperateResult> OnAsync()
+        {
+            return Task.Run(() => On());
+        }
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            Depart("Produce");
+            try
+            {
+                if (Channel == null)
+                {
+                    return Break("Produce", false, "未连接");
+                }
+
+                //内部消息组织
+                string msg = string.Empty;
+                if (Content.IsJson())
+                {
+                    msg = "{" + string.Format("\"T\":\"{0}\",\"C\":{1}", Topic, Content) + "}\r\n";
+                }
+                else
+                {
+                    msg = "{" + string.Format("\"T\":\"{0}\",\"C\":\"{1}\"", Topic, Content) + "}\r\n";
+                }
+                Channel.WriteAndFlushAsync(msg);
+                //延时
+                //TimeTool.DelayUs(0.2);
+                return Break("Produce", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("Produce", false, ex.Message, Exception: ex);
+            }
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            return Task.Run(() => Produce(Topic, Content));
+        }
+
+        public OperateResult Subscribe(string Topic)
+        {
+            Depart("Subscribe");
+            try
+            {
+                if (Channel == null)
+                {
+                    return Break("Subscribe", false, "未连接");
+                }
+                if (SubIoc.ContainsKey(Topic))
+                {
+                    return Break("Subscribe", false, "主题已存在");
+                }
+                else
+                {
+                    SubIoc.AddOrUpdate(Topic, DateTime.Now, (k, v) => DateTime.Now);
+                    return Break("Subscribe", true);
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Subscribe", false, ex.Message, Exception: ex);
+            }
+        }
+
+        public Task<OperateResult> SubscribeAsync(string Topic)
+        {
+            return Task.Run(() => Subscribe(Topic));
+        }
+
+        public OperateResult UnSubscribe(string Topic)
+        {
+            Depart("UnSubscribe");
+            try
+            {
+                if (Channel == null)
+                {
+                    return Break("UnSubscribe", false, "未连接");
+                }
+                if (SubIoc.ContainsKey(Topic))
+                {
+                    SubIoc.Remove(Topic, out _);
+                    return Break("UnSubscribe", true);
+                }
+                else
+                {
+                    return Break("UnSubscribe", false, "主题不存在");
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("UnSubscribe", false, ex.Message, Exception: ex);
+            }
+        }
+
+        public Task<OperateResult> UnSubscribeAsync(string Topic)
+        {
+            return Task.Run(() => UnSubscribe(Topic));
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Netty/service/NettyServiceData.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using System.ComponentModel;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Unility;
+
+namespace YSAI.Netty.service
+{
+    public class NettyServiceData
+    {
+
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        public class Basics
+        {
+            /// <summary>
+            /// 唯一标识符
+            /// </summary>
+            [Description("唯一标识符")]
+            public string? SN { get; set; } = Guid.NewGuid().ToNString();
+
+            /// <summary>
+            /// 端口
+            /// </summary>
+            [Description("端口")]
+            public int Port { get; set; }
+
+            /// <summary>
+            /// SSL文件路径
+            /// </summary>
+            [Description("SSL文件路径")]
+            public string? SslFilePath { get; set; }
+
+            /// <summary>
+            /// SSL文件密码
+            /// </summary>
+            [Description("SSL文件密码")]
+            public string? SslFilePassword { get; set; }
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Netty/service/NettyServiceOperate.cs

@@ -0,0 +1,303 @@
+using DotNetty.Codecs;
+using DotNetty.Handlers.Logging;
+using DotNetty.Handlers.Tls;
+using DotNetty.Transport.Bootstrapping;
+using DotNetty.Transport.Channels;
+using DotNetty.Transport.Channels.Groups;
+using DotNetty.Transport.Channels.Sockets;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.NetworkInformation;
+using System.Security.Cryptography.X509Certificates;
+using System.Text;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+using YSAI.Core.@interface;
+using YSAI.Netty.client;
+using YSAI.Unility;
+
+namespace YSAI.Netty.service
+{
+    public class NettyServiceOperate: IBaseAbstract, IDisposable
+    {
+        protected override string TAG => "NettyServiceOperate";
+
+        /// <summary>
+        /// 锁
+        /// </summary>
+        private static readonly object Lock = new object();
+        /// <summary>
+        /// 自身对象集合
+        /// </summary>
+        private static List<NettyServiceOperate> ThisObjList = new List<NettyServiceOperate>();
+
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static NettyServiceOperate Instance(NettyServiceData.Basics basics)
+        {
+            if (ThisObjList.Count >= MaxInstanceCount)
+            {
+                throw new Exception(ExceedMaxInstanceCountTips);
+            }
+            NettyServiceOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Comparer(basics).result);
+            if (exp == null)
+            {
+                lock (Lock)
+                {
+                    if (ThisObjList.Count(c => c.basics.Comparer(basics).result) > 0)
+                    {
+                        return ThisObjList.First(c => c.basics.Comparer(basics).result);
+                    }
+                    else
+                    {
+                        NettyServiceOperate exp2 = new NettyServiceOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private NettyServiceData.Basics basics;
+
+        /// <summary>
+        /// 客户端组
+        /// </summary>
+        private MultithreadEventLoopGroup ClientGroup;
+
+        /// <summary>
+        /// 客户端消息组
+        /// </summary>
+        private MultithreadEventLoopGroup ClientMessageGroup;
+
+        /// <summary>
+        /// 通信库
+        /// </summary>
+        private ServerBootstrap Communication;
+
+        /// <summary>
+        /// 通道
+        /// </summary>
+        private IChannel Channel;
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        public NettyServiceOperate(NettyServiceData.Basics basics)
+        {
+            this.basics = basics;
+        }
+
+        /// <summary>
+        /// 处理程序
+        /// </summary>
+        public class SecureChatServerHandler : SimpleChannelInboundHandler<string>
+        {
+            /// <summary>
+            /// 消息事件
+            /// </summary>
+            Action<object?, EventResult> MessageEvent;
+            /// <summary>
+            /// 组
+            /// </summary>
+            static volatile IChannelGroup group;
+
+            /// <summary>
+            /// 构造函数
+            /// </summary>
+            /// <param name="MessageEvent">消息事件</param>
+            public SecureChatServerHandler(Action<object?, EventResult> MessageEvent)
+            {
+                this.MessageEvent = MessageEvent;
+            }
+            public override void ChannelActive(IChannelHandlerContext contex)
+            {
+                IChannelGroup g = group;
+                if (g == null)
+                {
+                    lock (this)
+                    {
+                        if (group == null)
+                        {
+                            g = group = new DefaultChannelGroup(contex.Executor);
+                        }
+                    }
+                }
+                g.Add(contex.Channel);
+            }
+
+            class EveryOneBut : IChannelMatcher
+            {
+                readonly IChannelId id;
+
+                public EveryOneBut(IChannelId id)
+                {
+                    this.id = id;
+                }
+
+                public bool Matches(IChannel channel) => channel.Id != this.id;
+            }
+
+            protected override void ChannelRead0(IChannelHandlerContext contex, string msg)
+            {
+                //数据转发
+                string broadcast = string.Format("{0}\n", msg);
+                string response = string.Format("{0}\n", msg);
+                group.WriteAndFlushAsync(broadcast, new EveryOneBut(contex.Channel.Id));
+                //contex.WriteAndFlushAsync(response);
+                if (string.Equals("bye", msg, StringComparison.OrdinalIgnoreCase))
+                {
+                    contex.CloseAsync();
+                }
+            }
+
+            public override void ChannelReadComplete(IChannelHandlerContext ctx) => ctx.Flush();
+
+            /// <summary>
+            /// 出现了异常
+            /// </summary>
+            /// <param name="contex">通道处理程序上下文</param>
+            /// <param name="e">异常信息</param>
+            public override void ExceptionCaught(IChannelHandlerContext contex, Exception ex)
+            {
+                //消息抛出
+                MessageEvent?.Invoke(this, new EventResult(false, ex.Message));
+
+                //通道关闭
+                contex.CloseAsync();
+            }
+
+            public override bool IsSharable => true;
+        }
+
+        /// <summary>
+        /// 释放
+        /// </summary>
+        public void Dispose()
+        {
+            Off();
+            GC.Collect();
+            GC.SuppressFinalize(this);
+            ThisObjList.Remove(this);
+        }
+
+        /// <summary>
+        /// 打开
+        /// </summary>
+        /// <returns>统一泛型结果</returns>
+        public OperateResult On()
+        {
+            //开始记录运行时间
+            Depart("On");
+            try
+            {
+                if (Channel != null)
+                {
+                    return Break("On", false, "已连接");
+                }
+                //证书
+                X509Certificate2? Cert = null;
+                //证书目标主机
+                string? CertTargetHost = string.Empty;
+                //如果当证书不为空
+                if (!string.IsNullOrWhiteSpace(basics.SslFilePath) && !string.IsNullOrWhiteSpace(basics.SslFilePassword))
+                {
+                    Cert = new X509Certificate2(basics.SslFilePath, basics.SslFilePassword);
+                    CertTargetHost = Cert.GetNameInfo(X509NameType.DnsName, false);
+                }
+
+                //实例化通信库
+                Communication = new ServerBootstrap();
+                //实例化组
+                ClientGroup = new MultithreadEventLoopGroup(1);
+                ClientMessageGroup = new MultithreadEventLoopGroup();
+
+                //实例化编码格式
+                var STRING_ENCODER = new StringEncoder();
+                var STRING_DECODER = new StringDecoder();
+                //实例化处理程序
+                var SERVER_HANDLER = new SecureChatServerHandler(OnEventHandler);
+
+                //走TCP通道
+                Communication
+                .Group(ClientGroup, ClientMessageGroup)  //创建一个组
+                .Channel<TcpServerSocketChannel>()  //创建一个通道
+                .Option(ChannelOption.SoBacklog, 128)   //TCP节点布局
+                .Handler(new LoggingHandler(LogLevel.INFO)) //日志
+                .ChildHandler(new ActionChannelInitializer<ISocketChannel>(channel => { //处理
+                    IChannelPipeline pipeline = channel.Pipeline;
+                    if (Cert != null)
+                    {
+                        pipeline.AddLast(TlsHandler.Server(Cert));
+                    }
+                    pipeline.AddLast(new DelimiterBasedFrameDecoder(8192, Delimiters.LineDelimiter()));
+                    pipeline.AddLast(STRING_ENCODER, STRING_DECODER, SERVER_HANDLER);
+                }));
+
+                //创建
+                Channel = Communication.BindAsync(basics.Port).Result;
+
+                return Break("On", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("On", false, ex.Message, Exception: ex);
+            }
+        }
+
+        /// <summary>
+        /// 异步打开
+        /// </summary>
+        /// <returns>统一泛型结果</returns>
+        public Task<OperateResult> OnAsync()
+        {
+            return Task.Run(() => On());
+        }
+
+        /// <summary>
+        /// 关闭
+        /// </summary>
+        /// <returns>统一泛型结果</returns>
+        public OperateResult Off()
+        {
+            Depart("Off");
+            try
+            {
+                if (Channel == null)
+                {
+                    return Break("Off", false, "未连接");
+                }
+                Communication = null;
+                ClientGroup?.ShutdownGracefullyAsync().Wait();
+                ClientMessageGroup?.ShutdownGracefullyAsync().Wait();
+                Channel.CloseAsync().Wait();
+                Channel.DisconnectAsync().Wait();
+                Channel = null;
+                return Break("Off", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("Off", false, ex.Message, Exception: ex);
+            }
+        }
+
+        /// <summary>
+        /// 异步关闭
+        /// </summary>
+        /// <returns>统一泛型结果</returns>
+        public Task<OperateResult> OffAsync()
+        {
+            return Task.Run(() => Off());
+        }
+    }
+}

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

@@ -13,7 +13,7 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="OPCFoundation.NetStandard.Opc.Ua" Version="1.4.372.56" />
-	<PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+	<PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
 

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

@@ -64,7 +64,7 @@ namespace YSAI.Opc.ua.client
             /// 取样时间间隔
             /// </summary>
             [Description("取样时间间隔")]
-            public int SamplingInterval { get; set; } = 1000;
+            public int SamplingInterval { get; set; } = 100;
 
             /// <summary>
             /// 发布时间间隔

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

@@ -2,6 +2,7 @@
 using Opc.Ua.Client;
 using Opc.Ua.Configuration;
 using System.Collections.Concurrent;
+using System.Reflection.Metadata;
 using System.Security.Cryptography.X509Certificates;
 using YSAI.Core.data;
 using YSAI.Core.@enum;
@@ -10,6 +11,7 @@ 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;
 
 namespace YSAI.Opc.ua.client
@@ -136,21 +138,24 @@ namespace YSAI.Opc.ua.client
             {
                 if (!object.ReferenceEquals(session, this.clientSession))
                 {
-                    OnEventHandler(this, new EventResult(false, Steps.事件对象不匹配.ToString()));
                     return;
                 }
                 if (ServiceResult.IsBad(e.Status))
                 {
                     if (reconnectPeriod <= 0)
                     {
+                        OnEventHandler(this, new EventResult(false, $"[ {Steps.异常断开} ]连接错误  ({e.Status})"));
                         return;
                     }
+
+                    OnEventHandler(this, new EventResult(false, $"[ {Steps.异常断开} ]重连 {ConnectedCounet++} 次 ..."));
+
                     IsConnected = false;
+
                     if (sessionReconnectHandler == null)
                     {
                         sessionReconnectHandler = new SessionReconnectHandler();
-                        sessionReconnectHandler.BeginReconnect(session, reconnectPeriod * 1000, Server_ReconnectComplete);
-                        OnEventHandler(this, new EventResult(false, $"[ {Steps.异常断开} ]启动重连"));
+                        var state = sessionReconnectHandler.BeginReconnect(session, reconnectPeriod * 1000, Server_ReconnectComplete);
                     }
                     return;
                 }
@@ -180,14 +185,61 @@ namespace YSAI.Opc.ua.client
             {
                 if (!ReferenceEquals(sender, sessionReconnectHandler))
                 {
-                    OnEventHandler(this, new EventResult(false, Steps.事件对象不匹配.ToString()));
                     return;
                 }
+                // 仅在需要重新连接时应用会话
+                if (sessionReconnectHandler.Session != null)
+                {
+                    if (!ReferenceEquals(clientSession, sessionReconnectHandler.Session))
+                    {
+                        var session = clientSession;
+                        session.KeepAlive -= Session_KeepAlive;
+                        clientSession = sessionReconnectHandler.Session as Session;
+                        clientSession.KeepAlive += Session_KeepAlive;
+                        Utils.SilentDispose(session);
+
+                        //说明存在订阅
+                        if (allSubscriptions != null && allSubscriptions.Count > 0)
+                        {
+                            //先移除订阅
+                            if (!clientSession.RemoveSubscriptions(allSubscriptions.Values.ToArray()))
+                            {
+                                OnEventHandler(this, new EventResult(false, Steps.订阅通知.ToString(), $"设备重连成功，移除会话中的订阅项失败"));
+                            }
+                            else
+                            {
+                                foreach (var item in allSubscriptions)
+                                {
+                                    //重新添加订阅
+                                    if (!clientSession.AddSubscription(item.Value))
+                                    {
+                                        OnEventHandler(this, new EventResult(false, Steps.订阅通知.ToString(), $"设备重连成功，重新添加订阅失败，TAG = {item.Key}"));
+                                    }
+                                    else
+                                    {
+                                        //在服务器上创建订阅并添加所有监视项
+                                        item.Value.Create();
+
+                                        //将任何更改用于订阅
+                                        item.Value.ApplyChanges();
 
-                clientSession = sessionReconnectHandler.Session;
-                sessionReconnectHandler.Dispose();
-                sessionReconnectHandler = null;
-                OnEventHandler(this, new EventResult(true, Steps.重连成功.ToString()));
+                                        //初始重连次数
+                                        ConnectedCounet = 1;
+
+                                        OnEventHandler(this, new EventResult(true, Steps.重连成功.ToString()));
+                                    }
+                                }
+                            }
+                        }
+                        else
+                        {
+                            //初始重连次数
+                            ConnectedCounet = 1;
+
+                            OnEventHandler(this, new EventResult(true, Steps.重连成功.ToString()));
+                        }
+                    }
+                }
             }
             catch (Exception ex)
             {
@@ -857,7 +909,9 @@ namespace YSAI.Opc.ua.client
                             MonitoredItem monitoredItem = new MonitoredItem(allSubscriptions[Tag].DefaultItem);
                             monitoredItem.StartNodeId = new NodeId(item.AddressName);  //标识要监视的节点的浏览路径的开始节点
                             monitoredItem.AttributeId = Attributes.Value;  //要监视的属性
-                            monitoredItem.SamplingInterval = basics.SamplingInterval;  //采样间隔
+                            monitoredItem.SamplingInterval = basics.SamplingInterval;  //多少秒采集一次数据
+                            monitoredItem.DiscardOldest = true;  //当队列满时清空较早的队列数据
+                            monitoredItem.QueueSize = 1000;  //队列大小
                             monitoredItem.Notification += delegate (MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e) { OnMonitoredItemNotification(monitoredItem, e, item); };  //重写事件添加一个参数
                             monitoredItems?.Add(monitoredItem);  //添加至集合
                         }
@@ -875,6 +929,9 @@ namespace YSAI.Opc.ua.client
 
                             //在服务器上创建订阅并添加所有监视项
                             allSubscriptions[Tag].Create();
+
+                            //将任何更改用于订阅
+                            allSubscriptions[Tag].ApplyChanges();
                         }
                     }
                     else
@@ -882,13 +939,14 @@ namespace YSAI.Opc.ua.client
                         //此标签不存在，直接创建
                         //订阅
                         Subscription subscription = new Subscription(clientSession?.DefaultSubscription);
+                        subscription.Handle = this;  //分配给订阅的本地句柄
                         subscription.PublishingEnabled = true;//是否启用发布
                         subscription.PublishingInterval = basics.PublishingInterval; //出版间隔
-                        subscription.KeepAliveCount = uint.MaxValue;  //保活计数
-                        subscription.LifetimeCount = uint.MaxValue;  //寿命计数
-                        subscription.MaxNotificationsPerPublish = uint.MaxValue;  //每个发布请求的最大通知数
-                        subscription.Priority = 100;  //分配给订阅的优先级
+                        subscription.KeepAliveCount = 10;  //保活计数
+                        subscription.LifetimeCount = 100;  //寿命计数
+                        subscription.MaxNotificationsPerPublish = (uint)basics.PublishingInterval;  //每个发布请求的最大通知数
                         subscription.DisplayName = Tag;  //订阅的显示名称
+                        subscription.TimestampsToReturn = TimestampsToReturn.Both;  //与通知消息一起返回的时间戳
                         subscription.StateChanged += delegate (Subscription subscription, SubscriptionStateChangedEventArgs e) { Subscription_StateChanged(subscription, e, Tag); };  //订阅状态
                        
                         //监控项集合                                                                                                                                                    
@@ -902,7 +960,9 @@ namespace YSAI.Opc.ua.client
                             MonitoredItem monitoredItem = new MonitoredItem(subscription.DefaultItem);
                             monitoredItem.StartNodeId = new NodeId(item.AddressName);  //标识要监视的节点的浏览路径的开始节点
                             monitoredItem.AttributeId = Attributes.Value;  //要监视的属性
-                            monitoredItem.SamplingInterval = basics.SamplingInterval;  //采样间隔
+                            monitoredItem.SamplingInterval = basics.SamplingInterval;  //多少秒采集一次数据
+                            monitoredItem.DiscardOldest = true;  //当队列满时清空较早的队列数据
+                            monitoredItem.QueueSize = 1000;  //队列大小
                             monitoredItem.Notification += delegate (MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e) { OnMonitoredItemNotification(monitoredItem, e, item); };  //重写事件添加一个参数
                             monitoredItems?.Add(monitoredItem);  //添加至集合
                         }
@@ -914,10 +974,12 @@ namespace YSAI.Opc.ua.client
                         {
                             return Break("AddSubscribe", false, "会话添加订阅失败");
                         }
-
                         //在服务器上创建订阅并添加所有监视项
                         subscription.Create();
 
+                        //将任何更改用于订阅
+                        subscription.ApplyChanges();
+
                         //把此订阅添加到集合，方便后续移除订阅(当存在此键则更新值，不存在则添加)
                         allSubscriptions?.AddOrUpdate(Tag, subscription, (k, v) => subscription);
                     }
@@ -974,7 +1036,9 @@ namespace YSAI.Opc.ua.client
         {
             string Message = EnumParse(e.Status.ToString());
             string MessageJson = "{" + $"\"Uri\":\"{basics.ServerUrl}\",\"CustomName\":\"{basics.CustomName}\",\"StatusCode\":\"{e.Status}\",\"StatusMessage\":\"{Message}\"" + "}";
+#if DEBUG
             OnEventHandler(this, new EventResult(true, Message, MessageJson));
+#endif
         }
 
         /// <summary>

src/YSAI.DAQ/YSAI.Opc/ua/service/OpcUaServiceOperate.cs

@@ -82,10 +82,6 @@ namespace YSAI.Opc.ua.service
         /// 最后活动时间
         /// </summary>
         private DateTime LastEventTime;
-        /// <summary>
-        /// 状态
-        /// </summary>
-        private Task Status;
         #endregion
 
         #region 公共属性
@@ -320,13 +316,12 @@ namespace YSAI.Opc.ua.service
                 //关闭
                 using (Service _server = service)
                 {
-                    // 停止状态线程
-                    service.Dispose();
-                    service = null;
-                    Status.Wait();
                     // 停止服务并处理
                     _server.Stop();
                     _server.Dispose();
+
+                    // 停止状态线程
+                    service = null;
                 }
                 IsStart = false;
                 return Break("Off", true);

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQOperate.cs

@@ -302,8 +302,15 @@ namespace YSAI.RabbitMQ
                 Channels[e.Exchange].BasicAck(e.DeliveryTag, false);
             }
             string Topic = e.RoutingKey;
-            object Content = Encoding.UTF8.GetString(e.Body.ToArray());
-            OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content}" + "}", Core.@enum.ResultType.Dynamic));
+            string Content = Encoding.UTF8.GetString(e.Body.ToArray());
+            if (Content.IsJson())
+            {
+                OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":{Content}" + "}", Core.@enum.ResultType.Dynamic));
+            }
+            else
+            {
+                OnEventHandler(this, new EventResult(true, $"{TAG} 接收到主题 ( {Topic} ) 内容 ( {Content} )", "{" + $"\"Topic\":\"{Topic}\",\"Content\":\"{Content}\"" + "}", Core.@enum.ResultType.Dynamic));
+            }
         }
         public Task<OperateResult> OnAsync()
         {

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

@@ -14,7 +14,7 @@
 
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="6.6.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
   <ItemGroup>

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

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

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

@@ -14,7 +14,7 @@
 
 	<ItemGroup>
     <PackageReference Include="S7netplus" Version="0.20.0" />
-    <PackageReference Include="YSAI.Core" Version="1.0.0.58" />
+    <PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
 	<!--<ItemGroup>

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

@@ -1,4 +1,53 @@
-
+using YSAI.Core.data;
+using YSAI.Log;
+using YSAI.Netty.client;
+using YSAI.Netty.service;
+using YSAI.Unility;
+
+//NettyServiceOperate nettyServiceOperate = NettyServiceOperate.Instance(new NettyServiceData.Basics 
+//{
+//    Port = 8007,
+//    SN = "1"
+//});
+
+//Console.WriteLine(nettyServiceOperate.On().ToJson());
+
+
+//转发协议
+NettyClientOperate operate1 = NettyClientOperate.Instance(new NettyClientData.Basics
+{
+    Host = "127.0.0.1",
+    Port = 8007,
+    SN = "1"
+});
+
+NettyClientOperate operate2 = NettyClientOperate.Instance(new NettyClientData.Basics
+{
+    Host = "127.0.0.1",
+    Port = 8007,
+    SN = "2"
+});
+
+//打开
+OperateResult result = operate1.On();
+LogHelper.Info(result.ToJson().JsonFormatting());
+result = operate2.On();
+LogHelper.Info(result.ToJson().JsonFormatting());
+
+//消费
+operate1.OnEvent += delegate (object? sender, EventResult e)
+{
+    LogHelper.Info(e.ToJson().JsonFormatting());
+};
+result = operate1.Subscribe("测试");
+LogHelper.Info(result.ToJson().JsonFormatting());
+
+while (true)
+{
+    //生产
+    operate2.Produce("测试", new Random().NextDouble().ToString());
+}
+
 
 //using YSAI.Core.subscribe.core;
 //using YSAI.Unility;
@@ -364,51 +413,51 @@
 
 
 
-using System.Collections.Concurrent;
-using YSAI.Core.data;
-using YSAI.Log;
-using YSAI.Opc.ua.client;
-using YSAI.Unility;
+//using System.Collections.Concurrent;
+//using YSAI.Core.data;
+//using YSAI.Log;
+//using YSAI.Opc.ua.client;
+//using YSAI.Unility;
 
 
-Address address = JsonTool.StringToJsonEntity<Address>(FileTool.FileToString("C:\\Users\\Shun\\Desktop\\[6032]Node_Address 202310120854271486.json"));
+//Address address = JsonTool.StringToJsonEntity<Address>(FileTool.FileToString("C:\\Users\\Shun\\Desktop\\[6032]Node_Address 202310120854271486.json"));
 
 
-OpcUaClientOperate opcUaClientOperate = OpcUaClientOperate.Instance(new OpcUaClientData.Basics
-{
-    ServerUrl = "opc.tcp://127.0.0.1:8866/Opc.Ua.Service",
-    Password = "ysai",
-    UserName = "ysai",
-    CustomName = "YSAI 性能测试",
-});
-Console.WriteLine(opcUaClientOperate.On().ToJson().JsonFormatting());
-opcUaClientOperate.OnEvent += OpcUaClientOperate_OnEvent;
+//OpcUaClientOperate opcUaClientOperate = OpcUaClientOperate.Instance(new OpcUaClientData.Basics
+//{
+//    ServerUrl = "opc.tcp://127.0.0.1:8866/Opc.Ua.Service",
+//    Password = "ysai",
+//    UserName = "ysai",
+//    CustomName = "YSAI 性能测试",
+//});
+//Console.WriteLine(opcUaClientOperate.On().ToJson().JsonFormatting());
+//opcUaClientOperate.OnEvent += OpcUaClientOperate_OnEvent;
 
-while (true)
-{
-    Console.ReadLine();
-    OperateResult operateResult = opcUaClientOperate.Subscribe(address);
-    Console.WriteLine(operateResult.ToJson().JsonFormatting());
-}
+//while (true)
+//{
+//    Console.ReadLine();
+//    OperateResult operateResult = opcUaClientOperate.Subscribe(address);
+//    Console.WriteLine(operateResult.ToJson().JsonFormatting());
+//}
 
-void OpcUaClientOperate_OnEvent(object? sender, EventResult e)
-{
-    switch (e.RType)
-    {
-        case YSAI.Core.@enum.ResultType.KeyValue:
-            ConcurrentDictionary<string, AddressValue> pairs = e.RData as ConcurrentDictionary<string, AddressValue>;
-            foreach (var item in pairs)
-            {
-                String str = String.Format("{0,-100}{1,-100}", item.Key, item.Value.Value);
-
-                LogHelper.Verbose(str);
-            }
-            break;
-        default:
-            Console.WriteLine(e.Message);
-            break;
-    }
-}
+//void OpcUaClientOperate_OnEvent(object? sender, EventResult e)
+//{
+//    switch (e.RType)
+//    {
+//        case YSAI.Core.@enum.ResultType.KeyValue:
+//            ConcurrentDictionary<string, AddressValue> pairs = e.RData as ConcurrentDictionary<string, AddressValue>;
+//            foreach (var item in pairs)
+//            {
+//                String str = String.Format("{0,-100}{1,-100}", item.Key, item.Value.Value);
+
+//                LogHelper.Verbose(str);
+//            }
+//            break;
+//        default:
+//            Console.WriteLine(e.Message);
+//            break;
+//    }
+//}
 
 
 

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

@@ -13,6 +13,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\YSAI.Netty\YSAI.Netty.csproj" />
     <ProjectReference Include="..\YSAI.Opc\YSAI.Opc.csproj" />
   </ItemGroup>
 

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.58" />
+	<PackageReference Include="YSAI.Core" Version="1.0.0.60" />
   </ItemGroup>
 
   <ItemGroup>

src/YSAI.DAQ/YSAI.Unility/ExtensionTool.cs

@@ -1,4 +1,5 @@
 using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
 using System.Net;
 using System.Security.Cryptography;
 using System.Text;
@@ -1349,6 +1350,24 @@ namespace YSAI.Unility
             return value;
         }
 
+        /// <summary>
+        /// 判断是不是有效JSON 
+        /// </summary>
+        /// <param name="json">Json 字符串</param>
+        /// <returns></returns>
+        public static bool IsJson(this string json)
+        {
+            JToken jToken = JToken.Parse(json);
+            if (jToken.Type != JTokenType.Object)
+            {
+                return false;
+            }
+            else
+            {
+                return true;
+            }
+        }
+
         /// <summary>
         /// JSON格式化
         /// </summary>

src/YSAI.DAQ/YSAI.Unility/YSAI.Unility.csproj

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