1

readme.txt

@@ -27,6 +27,7 @@ RabbitMQ Publisher/Consumer
 4.支持多点转发，多点解析
 5.支持快速对接，与配置
 6.支持虚拟点位
+7.所有采集协议支持订阅模式
 
 
 doc 中有演示视频

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

@@ -162,7 +162,7 @@ namespace YSAI.Core.data
                     //RabbitMQ转发
                     if (addressDetails.RabbitMQRelay != null)
                     {
-                        OperateResult? operateResult = reflectionOperate.ExecuteMethod(addressDetails.RabbitMQRelay.ReflectionSN, new object[] { addressDetails.RabbitMQRelay.MessageHead, value, addressDetails.RabbitMQRelay.Type, addressDetails.RabbitMQRelay.Durable, addressDetails.RabbitMQRelay.Exclusive, addressDetails.RabbitMQRelay.AutoDelete }) as OperateResult;
+                        OperateResult? operateResult = reflectionOperate.ExecuteMethod(addressDetails.RabbitMQRelay.ReflectionSN, new object[] { addressDetails.RabbitMQRelay.Topic, value, addressDetails.RabbitMQRelay.Type, addressDetails.RabbitMQRelay.Durable, addressDetails.RabbitMQRelay.Exclusive, addressDetails.RabbitMQRelay.AutoDelete }) as OperateResult;
                         if (!operateResult.State)
                         {
                             DynamicObj.Param = addressDetails.RabbitMQRelay;

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

@@ -79,7 +79,7 @@ namespace YSAI.Core.data
             /// <summary>
             /// 消息头
             /// </summary>
-            public string MessageHead { get; set; }
+            public string Topic { get; set; }
             /// <summary>
             /// 通道类型
             /// direct  //用于AMQP直接交换的交换类型。

src/YSAI.DAQ/YSAI.Core/interface/only/IKafka.cs

@@ -65,24 +65,20 @@ namespace YSAI.Core.@interface.only
         /// <summary>
         /// 生产
         /// </summary>
-        /// <typeparam name="K">键类型</typeparam>
-        /// <typeparam name="V">值类型</typeparam>
         /// <param name="Topic">主题</param>
         /// <param name="Key">键</param>
-        /// <param name="Value">值</param>
+        /// <param name="Content">值</param>
         /// <returns>统一出参</returns>
-        OperateResult Produce(string Topic, object Key, object Value);
+        OperateResult Produce(string Topic, object Key, object Content);
 
         /// <summary>
         /// 生产
         /// </summary>
-        /// <typeparam name="K">键类型</typeparam>
-        /// <typeparam name="V">值类型</typeparam>
         /// <param name="Topic">主题</param>
         /// <param name="Key">键</param>
-        /// <param name="Value">值</param>
+        /// <param name="Content">值</param>
         /// <returns>统一出参</returns>
-        Task<OperateResult> ProduceAsync(string Topic, object Key, object Value);
+        Task<OperateResult> ProduceAsync(string Topic, object Key, object Content);
 
     }
 }

src/YSAI.DAQ/YSAI.Core/interface/only/IMqtt.cs

@@ -18,7 +18,7 @@ namespace YSAI.Core.@interface.only
         /// 发布订阅
         /// </summary>
         /// <param name="Topic">主题</param>
-        /// <param name="Value">内容</param>
+        /// <param name="Content">内容</param>
         /// <param name="QoSLevel">
         /// 消息等级
         /// QoS 0 是最低级别，基本上等同于 Fire and Forget 模式，发送者发送完数据之后，不关心消息是否已经投递到了接收者那边。
@@ -29,12 +29,12 @@ namespace YSAI.Core.@interface.only
         /// </param>
         /// <param name="Retain">消息是否保留</param>
         /// <returns>统一结果</returns>
-        Task<OperateResult> PublishSubscribeAsync(string Topic, string Value, int QoSLevel = 0, bool Retain = false);
+        Task<OperateResult> PublishSubscribeAsync(string Topic, string Content, int QoSLevel = 0, bool Retain = false);
         /// <summary>
         /// 发布订阅
         /// </summary>
         /// <param name="Topic">主题</param>
-        /// <param name="Value">内容</param>
+        /// <param name="Content">内容</param>
         /// <param name="QoSLevel">
         /// 消息等级
         /// QoS 0 是最低级别，基本上等同于 Fire and Forget 模式，发送者发送完数据之后，不关心消息是否已经投递到了接收者那边。
@@ -45,7 +45,7 @@ namespace YSAI.Core.@interface.only
         /// </param>
         /// <param name="Retain">消息是否保留</param>
         /// <returns>统一结果</returns>
-        OperateResult PublishSubscribe(string Topic, string Value, int QoSLevel = 0, bool Retain = false);
+        OperateResult PublishSubscribe(string Topic, string Content, int QoSLevel = 0, bool Retain = false);
 
         /// <summary>
         /// 添加订阅

src/YSAI.DAQ/YSAI.Core/interface/only/IRabbitMQ.cs

@@ -9,15 +9,15 @@ using YSAI.Core.@interface.unify;
 namespace YSAI.Core.@interface.only
 {
     /// <summary>
-    /// 发布者接口
+    /// RabbitMQ接口
     /// </summary>
-    public interface IRabbitMQPublisher : IOn, IOff, IDisposable
+    public interface IRabbitMQ : IOn, IOff, IDisposable
     {
         /// <summary>
         /// 发布 一个交换机可以发布多个消息
         /// </summary>
-        /// <param name="MessageHead">消息头</param>
-        /// <param name="MessageContent">消息内容</param>
+        /// <param name="Topic">消息头</param>
+        /// <param name="Content">消息内容</param>
         /// <param name="Type">
         /// 通道类型
         /// direct  //用于AMQP直接交换的交换类型。
@@ -29,13 +29,13 @@ namespace YSAI.Core.@interface.only
         /// <param name="Exclusive">独有的</param>
         /// <param name="AutoDelete">自动删除</param>
         /// <returns>统一出参</returns>
-        OperateResult Publish(string MessageHead, string MessageContent, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
+        OperateResult Publish(string Topic, string Content, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
 
         /// <summary>
         /// 发布 一个交换机可以发布多个消息
         /// </summary>
-        /// <param name="MessageHead">消息头</param>
-        /// <param name="MessageContent">消息内容</param>
+        /// <param name="Topic">消息头</param>
+        /// <param name="Content">消息内容</param>
         /// <param name="Type">
         /// 通道类型
         /// direct  //用于AMQP直接交换的交换类型。
@@ -47,18 +47,12 @@ namespace YSAI.Core.@interface.only
         /// <param name="Exclusive">独有的</param>
         /// <param name="AutoDelete">自动删除</param>
         /// <returns>统一出参</returns>
-        Task<OperateResult> PublishAsync(string MessageHead, string MessageContent, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
-    }
+        Task<OperateResult> PublishAsync(string Topic, string Content, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
 
-    /// <summary>
-    /// 消费者接口
-    /// </summary>
-    public interface IRabbitMQConsumer : IOn, IOff, IDisposable
-    {
         /// <summary>
         /// 消费 一个交换机可以消费多个消息
         /// </summary>
-        /// <param name="MessageHead">消息头</param>
+        /// <param name="Topic">消息头</param>
         /// <param name="Type">
         /// 通道类型
         /// direct  //用于AMQP直接交换的交换类型。
@@ -71,12 +65,12 @@ namespace YSAI.Core.@interface.only
         /// <param name="Exclusive">独有的</param>
         /// <param name="AutoDelete">自动删除</param>
         /// <returns>统一出参</returns>
-        OperateResult Consume(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
+        OperateResult Consume(string Topic, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
 
         /// <summary>
         /// 消费 一个交换机可以消费多个消息
         /// </summary>
-        /// <param name="MessageHead">消息头</param>
+        /// <param name="Topic">消息头</param>
         /// <param name="Type">
         /// 通道类型
         /// direct  //用于AMQP直接交换的交换类型。
@@ -89,6 +83,6 @@ namespace YSAI.Core.@interface.only
         /// <param name="Exclusive">独有的</param>
         /// <param name="AutoDelete">自动删除</param>
         /// <returns>统一出参</returns>
-        Task<OperateResult> ConsumeAsync(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
+        Task<OperateResult> ConsumeAsync(string Topic, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false);
     }
 }

src/YSAI.DAQ/YSAI.Core/interface/unify/IConsumer.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+
+namespace YSAI.Core.@interface.unify
+{
+    /// <summary>
+    /// 消费者接口
+    /// </summary>
+    public interface IConsumer
+    {
+        /// <summary>
+        /// 消费
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <returns></returns>
+        OperateResult Consume(string Topic);
+
+        /// <summary>
+        /// 消费
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <returns></returns>
+        Task<OperateResult> ConsumeAsync(string Topic);
+    }
+}

src/YSAI.DAQ/YSAI.Core/interface/unify/IProducer.cs

@@ -0,0 +1,31 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+
+namespace YSAI.Core.@interface.unify
+{
+    /// <summary>
+    /// 生产者接口
+    /// </summary>
+    public interface IProducer
+    {
+        /// <summary>
+        /// 生产
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <param name="Content">内容</param>
+        /// <returns></returns>
+        OperateResult Produce(string Topic, string Content);
+
+        /// <summary>
+        /// 生产
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <param name="Content">内容</param>
+        /// <returns></returns>
+        Task<OperateResult> ProduceAsync(string Topic, string Content);
+    }
+}

src/YSAI.DAQ/YSAI.Core/interface/unify/IRelay.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace YSAI.Core.@interface.unify
+{
+    /// <summary>
+    /// 转发接口
+    /// </summary>
+    public interface IRelay :IProducer, IConsumer, IDisposable
+    {
+
+    }
+}

src/YSAI.DAQ/YSAI.DAQ.sln

@@ -23,7 +23,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Kafka", "YSAI.Kafka\YS
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.DB", "YSAI.DB\YSAI.DB.csproj", "{439C729E-C66D-492B-A481-0C7F73F04E76}"
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "lib", "lib", "{0A264424-1AD7-49FA-B813-D96498066479}"
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "daq", "daq", "{0A264424-1AD7-49FA-B813-D96498066479}"
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Test", "YSAI.Test\YSAI.Test.csproj", "{26BEA521-56B3-4BE8-9F5C-62BCA2323ABD}"
 EndProject
@@ -59,6 +59,16 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Tool.Windows", "YSAI.T
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "YSAI.Manage.Windows", "YSAI.Manage.Windows\YSAI.Manage.Windows.csproj", "{2221CE78-FA24-40E3-8453-9C66E32F9C0C}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "relay", "relay", "{9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "YSAI.Relay", "YSAI.Relay\YSAI.Relay.csproj", "{E43BECDA-DE1F-446A-96E7-F110977597F2}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "core", "core", "{D60224CF-7F12-453B-851E-B5C01F9D2BBE}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "test", "test", "{12CB0510-7B1E-4518-AA3B-412A4D323D42}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "manage", "manage", "{ECAD410C-2895-4836-BCA7-D4EF340E778E}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -169,19 +179,31 @@ Global
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{2221CE78-FA24-40E3-8453-9C66E32F9C0C}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E43BECDA-DE1F-446A-96E7-F110977597F2}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
+		{859336BA-B7A8-45E7-BE35-72B1BB7D01B7} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
+		{3D1D4ACF-EFD0-497D-891D-527BDE7CA9A4} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
+		{C8EEBF30-365E-4F22-A58C-C68FB58E4256} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
 		{4A24E34B-5BBF-4D1E-AEE2-B22FF6E6A7D9} = {0A264424-1AD7-49FA-B813-D96498066479}
-		{E197B574-70BD-48C4-9301-DF9963233A30} = {0A264424-1AD7-49FA-B813-D96498066479}
+		{E197B574-70BD-48C4-9301-DF9963233A30} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
 		{FEA383F4-9CDD-422F-9068-200A66DF7C81} = {0A264424-1AD7-49FA-B813-D96498066479}
-		{1EE30939-69BA-4F20-96B3-D4082C937924} = {0A264424-1AD7-49FA-B813-D96498066479}
+		{1EE30939-69BA-4F20-96B3-D4082C937924} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
 		{FEFEFB3E-30EE-429A-B39D-F39CF1338E38} = {0A264424-1AD7-49FA-B813-D96498066479}
-		{2B2ED6D8-B498-4589-ADAB-4599B9EFFC70} = {0A264424-1AD7-49FA-B813-D96498066479}
+		{2B2ED6D8-B498-4589-ADAB-4599B9EFFC70} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
 		{439C729E-C66D-492B-A481-0C7F73F04E76} = {0A264424-1AD7-49FA-B813-D96498066479}
-		{8CE7E64C-7A6A-4581-A9B3-C05214986B4F} = {0A264424-1AD7-49FA-B813-D96498066479}
+		{26BEA521-56B3-4BE8-9F5C-62BCA2323ABD} = {12CB0510-7B1E-4518-AA3B-412A4D323D42}
+		{6F261956-DB97-43E4-A982-FBA2FB78AE15} = {12CB0510-7B1E-4518-AA3B-412A4D323D42}
+		{8CE7E64C-7A6A-4581-A9B3-C05214986B4F} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
+		{6E9667C0-303C-472D-949B-60F812E6C659} = {ECAD410C-2895-4836-BCA7-D4EF340E778E}
+		{57F9FF57-ADD3-40B0-AE20-C2F88A2784A0} = {ECAD410C-2895-4836-BCA7-D4EF340E778E}
+		{6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
 		{8B8BF505-7A00-4868-939F-EB5056083E49} = {0A264424-1AD7-49FA-B813-D96498066479}
 		{CCF267B2-ABB4-4B42-8351-103D59FA4B3B} = {6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6}
 		{2FC39615-7195-4A0D-9959-E49630AB41D6} = {6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6}
@@ -190,6 +212,10 @@ Global
 		{D7FFC9F4-1248-469D-BAF9-AD675BF83A92} = {6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6}
 		{6A255A65-3198-44AA-A281-62F721D49CEF} = {6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6}
 		{0EA9373C-B55A-4400-82AC-7681AA996229} = {6F526F44-A5B0-49D5-8E04-DBFDB5F8E6C6}
+		{257F1474-B220-4C61-88C6-5B83BEF7B3A7} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
+		{49133ADB-D3BF-4682-AA5A-CC1CF1917D45} = {D60224CF-7F12-453B-851E-B5C01F9D2BBE}
+		{2221CE78-FA24-40E3-8453-9C66E32F9C0C} = {ECAD410C-2895-4836-BCA7-D4EF340E778E}
+		{E43BECDA-DE1F-446A-96E7-F110977597F2} = {9D8EDBBA-7A97-4D84-9B12-7FCC2F834046}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {5D5D3927-6714-40C0-84EA-44C5BA4C5E87}

src/YSAI.DAQ/YSAI.Kafka/KafkaProducerOperate.cs

@@ -9,7 +9,7 @@ using YSAI.Unility;
 
 namespace YSAI.Kafka
 {
-    public sealed class KafkaProducerOperate: IBaseAbstract, IKafkaProducer
+    public sealed class KafkaProducerOperate : IBaseAbstract, IKafkaProducer
     {
         private static readonly object Lock = new object();  //锁
         private static List<KafkaProducerOperate> ThisObjList = new List<KafkaProducerOperate>(); //自身对象集合
@@ -64,7 +64,7 @@ namespace YSAI.Kafka
         }
         protected override string LogHead => "[ KafkaProducerOperate 操作 ]";
         protected override string ClassName => "KafkaProducerOperate";
-        public OperateResult Produce(string Topic, object Key, object Value)
+        public OperateResult Produce(string Topic, object Key, object Content)
         {
             //开始记录运行时间
             Depart("Produce");
@@ -72,48 +72,48 @@ namespace YSAI.Kafka
             {
                 //失败的消息
                 List<string> MessageFail = new List<string>();
-                    if (producerConfig == null)
+                if (producerConfig == null)
+                {
+                    //创建配置
+                    producerConfig = new ProducerConfig()
                     {
-                        //创建配置
-                        producerConfig = new ProducerConfig()
-                        {
-                            ClientId = basics.ClientId,
-                            BootstrapServers = basics.BootstrapServers
-                            //,
-                            //SecurityProtocol = basics.SecurityProtocol,
-                            //SaslMechanism = basics.SaslMechanism,
-                            //SaslKerberosServiceName = basics.SaslKerberosServiceName,
-                            //SaslKerberosKeytab = basics.SaslKerberosKeytab,
-                            //SaslKerberosPrincipal = basics.SaslKerberosPrincipal
-                        };
-                    }
-                    using (IProducer<object, object> producer = new ProducerBuilder<object, object>(producerConfig).Build())
+                        ClientId = basics.ClientId,
+                        BootstrapServers = basics.BootstrapServers
+                        //,
+                        //SecurityProtocol = basics.SecurityProtocol,
+                        //SaslMechanism = basics.SaslMechanism,
+                        //SaslKerberosServiceName = basics.SaslKerberosServiceName,
+                        //SaslKerberosKeytab = basics.SaslKerberosKeytab,
+                        //SaslKerberosPrincipal = basics.SaslKerberosPrincipal
+                    };
+                }
+                using (IProducer<object, object> producer = new ProducerBuilder<object, object>(producerConfig).Build())
+                {
+                    //返回结果
+                    OperateResult? result = null;
+                    //发布结果
+                    producer.Produce(Topic, new Message<object, object>() { Key = Key, Value = Content }, Result =>
                     {
-                        //返回结果
-                        OperateResult? result = null;
-                        //发布结果
-                        producer.Produce(Topic, new Message<object, object>() { Key = Key, Value = Value }, Result =>
-                        {
-                            if (Result.Error.Code != ErrorCode.NoError)
-                            {
-                                result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败：{Result.Error.Reason}");
-                            }
-                            else
-                            {
-                                result = Break("Produce", true, $"主题 [ {Topic} ] 发送成功，偏移量：{Result.TopicPartitionOffset}");
-                            }
-                        });
-                        //最多等待时间
-                        producer.Flush(new TimeSpan(0,0,0,0, basics.WaitTime));
-                        if (result == null)
+                        if (Result.Error.Code != ErrorCode.NoError)
                         {
-                            result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败");
+                            result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败：{Result.Error.Reason}");
                         }
-                        if (!result.State)
+                        else
                         {
-                            MessageFail.Add(result.Message);
+                            result = Break("Produce", true, $"主题 [ {Topic} ] 发送成功，偏移量：{Result.TopicPartitionOffset}");
                         }
+                    });
+                    //最多等待时间
+                    producer.Flush(new TimeSpan(0, 0, 0, 0, basics.WaitTime));
+                    if (result == null)
+                    {
+                        result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败");
+                    }
+                    if (!result.State)
+                    {
+                        MessageFail.Add(result.Message);
                     }
+                }
                 if (MessageFail.Count > 0)
                 {
                     return Break("Produce", false, $"批量发送失败，存在发送失败数据：{MessageFail.ToJson()}");
@@ -128,10 +128,10 @@ namespace YSAI.Kafka
                 return Break("Produce", false, ex.Message);
             }
         }
-        
-        public Task<OperateResult> ProduceAsync(string Topic, object Key, object Value)
+
+        public Task<OperateResult> ProduceAsync(string Topic, object Key, object Content)
         {
-            return Task.Run(() => Produce(Topic,Key,Value));
+            return Task.Run(() => Produce(Topic, Key, Content));
         }
     }
 }

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

@@ -13,8 +13,6 @@
 
   <ItemGroup>
     <ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
-    <ProjectReference Include="..\YSAI.Log\YSAI.Log.csproj" />
-    <ProjectReference Include="..\YSAI.Unility\YSAI.Unility.csproj" />
   </ItemGroup>
 
 </Project>

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

@@ -119,12 +119,12 @@ namespace YSAI.Mqtt.client
 
         protected override string ClassName => "MqttClientOperate";
 
-        public Task<OperateResult> PublishSubscribeAsync(string Topic, string Value, int QoSLevel = 0, bool Retain = false)
+        public Task<OperateResult> PublishSubscribeAsync(string Topic, string Content, int QoSLevel = 0, bool Retain = false)
         {
-            return Task.Run(()=> PublishSubscribe(Topic, Value, QoSLevel, Retain));
+            return Task.Run(()=> PublishSubscribe(Topic, Content, QoSLevel, Retain));
         }
 
-        public OperateResult PublishSubscribe(string Topic, string Value, int QoSLevel = 0, bool Retain = false)
+        public OperateResult PublishSubscribe(string Topic, string Content, int QoSLevel = 0, bool Retain = false)
         {
             //开始记录运行时间
             Depart("PublishSubscribe");
@@ -134,7 +134,7 @@ namespace YSAI.Mqtt.client
                 //主题
                 mqttApplicationMessage.Topic = Topic;
                 //内容
-                mqttApplicationMessage.Payload = Encoding.UTF8.GetBytes(Value);
+                mqttApplicationMessage.Payload = Encoding.UTF8.GetBytes(Content);
                 //消息等级
                 mqttApplicationMessage.QualityOfServiceLevel = (MqttQualityOfServiceLevel)QoSLevel;
                 //消息是否保留

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

@@ -36,7 +36,8 @@ namespace YSAI.Opc.da.client
             /// 接口版本
             /// </summary>
             [Description("接口版本")]
-            public Specification SpecificationVer { get; set; }
+            [JsonConverter(typeof(StringEnumConverter))]
+            public ApiVerType ApiVerType { get; set; }
             /// <summary>
             /// 更新频率
             /// </summary>
@@ -53,7 +54,7 @@ namespace YSAI.Opc.da.client
                 if (obj == null) return false;
                 if (obj.SN == this.SN &&
                     obj.ServiceName == this.ServiceName &&
-                    obj.SpecificationVer == this.SpecificationVer)
+                    obj.ApiVerType == this.ApiVerType)
                 {
                     return true;
                 }
@@ -64,6 +65,27 @@ namespace YSAI.Opc.da.client
             }
         }
 
+        /// <summary>
+        /// 版本类型
+        /// </summary>
+        public enum ApiVerType
+        {
+            /// <summary>
+            /// 1.x api
+            /// </summary>
+            [Description("1.x api")]
+            COM_DA_10,
+            /// <summary>
+            /// 2.x api
+            /// </summary>
+            [Description("2.x api")]
+            COM_DA_20,
+            /// <summary>
+            /// 3.x api
+            /// </summary>
+            [Description("3.x api")]
+            COM_DA_30
+        }
 
         /// <summary>
         /// 事件响应数据带参

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

@@ -555,7 +555,20 @@ namespace YSAI.Opc.da.client
             {
                 if (opcDaClient == null)
                 {
-                    OpcDaNetApi.Server[]? servers = GetServersArray(basics.SpecificationVer);
+                    OpcDaNetApi.Specification specification = OpcDaNetApi.Specification.COM_DA_20;
+                    switch (basics.ApiVerType)
+                    {
+                        case OpcDaClientData.ApiVerType.COM_DA_10:
+                            specification = OpcDaNetApi.Specification.COM_DA_10;
+                            break;
+                        case OpcDaClientData.ApiVerType.COM_DA_20:
+                            specification = OpcDaNetApi.Specification.COM_DA_20;
+                            break;
+                        case OpcDaClientData.ApiVerType.COM_DA_30:
+                            specification = OpcDaNetApi.Specification.COM_DA_30;
+                            break;
+                    }
+                    OpcDaNetApi.Server[]? servers = GetServersArray(specification);
                     if (servers != null)
                     {
                         foreach (var item in servers)
@@ -565,7 +578,7 @@ namespace YSAI.Opc.da.client
                                 //opcda客户端对象
                                 opcDaClient = item as OpcDaNetApi.Da.Server;
                                 //连接
-                                opcDaClient.Connect();
+                                opcDaClient?.Connect();
                                 //判断连接状态
                                 if (opcDaClient.IsConnected)
                                 {

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQConsumerOperate.cs

@@ -15,23 +15,23 @@ using YSAI.Unility;
 namespace YSAI.RabbitMQ
 {
     /// <summary>
-    /// 消费者操作
+    /// 发布者操作
     /// </summary>
-    public sealed class RabbitMQConsumerOperate : IBaseAbstract<RabbitMQData.Event>, IRabbitMQConsumer
+    public sealed class RabbitMQOperate : IBaseAbstract<RabbitMQData.Event>, IRabbitMQ
     {
-        protected override string LogHead => "[ RabbitMQConsumerOperate 操作 ]";
-        protected override string ClassName => "RabbitMQConsumerOperate";
+        protected override string LogHead => "[ RabbitMQOperate 操作 ]";
+        protected override string ClassName => "RabbitMQOperate";
 
         private static readonly object Lock = new object();  //锁
-        private static List<RabbitMQConsumerOperate> ThisObjList = new List<RabbitMQConsumerOperate>(); //自身对象集合
+        private static List<RabbitMQOperate> ThisObjList = new List<RabbitMQOperate>(); //自身对象集合
 
         /// <summary>
         /// 单例模式
         /// </summary>
         /// <returns></returns>
-        public static RabbitMQConsumerOperate Instance(RabbitMQData.Basics basics)
+        public static RabbitMQOperate Instance(RabbitMQData.Basics basics)
         {
-            RabbitMQConsumerOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
+            RabbitMQOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
             if (exp == null)
             {
                 lock (Lock)
@@ -42,7 +42,7 @@ namespace YSAI.RabbitMQ
                     }
                     else
                     {
-                        RabbitMQConsumerOperate exp2 = new RabbitMQConsumerOperate(basics);
+                        RabbitMQOperate exp2 = new RabbitMQOperate(basics);
                         ThisObjList.Add(exp2);
                         return exp2;
                     }
@@ -50,6 +50,7 @@ namespace YSAI.RabbitMQ
             }
             return exp;
         }
+
         /// <summary>
         /// 基础数据
         /// </summary>
@@ -58,7 +59,7 @@ namespace YSAI.RabbitMQ
         /// <summary>
         /// 构造函数
         /// </summary>
-        public RabbitMQConsumerOperate(RabbitMQData.Basics basics)
+        public RabbitMQOperate(RabbitMQData.Basics basics)
         {
             this.basics = basics;
         }
@@ -76,19 +77,10 @@ namespace YSAI.RabbitMQ
         /// 队列集合
         /// </summary>
         private ConcurrentDictionary<string, string> Queues = new ConcurrentDictionary<string, string>();
-
         /// <summary>
-        /// [消费]接收发布者消息
+        /// 队列与基础数据
         /// </summary>
-        private void Consumer_Received(object? sender, BasicDeliverEventArgs e, bool AutoAck = false)
-        {
-            OnEventHandler?.Invoke(this, new RabbitMQData.Event() { MessageContent = Encoding.UTF8.GetString(e.Body.ToArray()), MessageHead = e.RoutingKey });
-            if (!AutoAck)
-            {
-                //当自动确认为false，得手动确认消息
-                Channels[e.RoutingKey].BasicAck(e.DeliveryTag, false);
-            }
-        }
+        private ConcurrentDictionary<string, (QueueDeclareOk QueueDeclareOk, IBasicProperties BasicProperties)> QueueAndBasicProperties = new ConcurrentDictionary<string, (QueueDeclareOk QueueDeclareOk, IBasicProperties BasicProperties)>();
 
         public Task<OperateResult> OnAsync()
         {
@@ -153,8 +145,8 @@ namespace YSAI.RabbitMQ
                     connection.Dispose();
                     //清空通道集合
                     Channels.Clear();
-                    //队列清空
-                    Queues.Clear();
+                    //清空基础数据
+                    QueueAndBasicProperties.Clear();
                     return Break("Off", true);
                 }
             }
@@ -163,7 +155,6 @@ namespace YSAI.RabbitMQ
                 return Break("Off", false, "打开RabbitMQ连接异常：" + ex.Message);
             }
         }
-
         public void Dispose()
         {
             Off();
@@ -172,7 +163,58 @@ namespace YSAI.RabbitMQ
             ThisObjList.Remove(this);
         }
 
-        public OperateResult Consume(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        public OperateResult Publish(string Topic, string Content, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        {
+            //开始记录运行时间
+            Depart("Publish");
+            try
+            {
+                if (connection == null || !connection.IsOpen)
+                {
+                    return Break("Publish", false, "RabbitMQ连接未打开");
+                }
+                else
+                {
+                    //判断是否存在此通道
+                    if (!Channels.ContainsKey(basics.ExChangeName))
+                    {
+                        IModel channel = connection.CreateModel();
+                        //创建交换机
+                        channel.ExchangeDeclare(basics.ExChangeName, Type, Durable, AutoDelete);
+                        //添加通道
+                        Channels.TryAdd(basics.ExChangeName, channel);
+                    }
+                    //判断队列与消息头是否存在
+                    if (!QueueAndBasicProperties.ContainsKey(Topic))
+                    {
+                        //创建队列
+                        QueueDeclareOk queueDeclareOk = Channels[basics.ExChangeName].QueueDeclare(Topic, Durable, Exclusive, AutoDelete);
+                        //绑定队列
+                        Channels[basics.ExChangeName].QueueBind(Topic, basics.ExChangeName, Topic);
+                        //消息持久化
+                        IBasicProperties basicProperties = Channels[basics.ExChangeName].CreateBasicProperties();
+                        basicProperties.Persistent = true;
+                        //添加消息头
+                        QueueAndBasicProperties.TryAdd(Topic, (queueDeclareOk, basicProperties));
+                    }
+                    // 发布消息
+                    Channels[basics.ExChangeName].BasicPublish(basics.ExChangeName, Topic, QueueAndBasicProperties[Topic].BasicProperties, Encoding.UTF8.GetBytes(Content));
+                    //返回
+                    return Break("Publish", true);
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Publish", false, "RabbitMQ发布异常：" + ex.Message);
+            }
+        }
+
+        public Task<OperateResult> PublishAsync(string Topic, string Content, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        {
+            return Task.Run(() => Publish(Topic, Content, Type, Durable, Exclusive, AutoDelete));
+        }
+
+        public OperateResult Consume(string Topic, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
         {
             //开始记录运行时间
             Depart("Consume");
@@ -194,12 +236,12 @@ namespace YSAI.RabbitMQ
                         Channels.TryAdd(basics.ExChangeName, channel);
                     }
 
-                    if (!Queues.ContainsKey(MessageHead))
+                    if (!Queues.ContainsKey(Topic))
                     {
                         //创建队列
-                        Channels[basics.ExChangeName].QueueDeclare(MessageHead, Durable, Exclusive, AutoDelete);
+                        Channels[basics.ExChangeName].QueueDeclare(Topic, Durable, Exclusive, AutoDelete);
                         //绑定队列
-                        Channels[basics.ExChangeName].QueueBind(MessageHead, basics.ExChangeName, MessageHead);
+                        Channels[basics.ExChangeName].QueueBind(Topic, basics.ExChangeName, Topic);
 
                         EventingBasicConsumer consumer = new(Channels[basics.ExChangeName]);
                         consumer.Received += delegate (object? sender, BasicDeliverEventArgs e)
@@ -207,7 +249,7 @@ namespace YSAI.RabbitMQ
                             Consumer_Received(sender, e, AutoAck);
                         };
                         // 开启消费者与通道、队列关联
-                        Channels[basics.ExChangeName].BasicConsume(MessageHead, AutoAck, consumer);
+                        Channels[basics.ExChangeName].BasicConsume(Topic, AutoAck, consumer);
                         //返回
                         return Break("Consume", true);
                     }
@@ -227,9 +269,22 @@ namespace YSAI.RabbitMQ
             }
         }
 
-        public Task<OperateResult> ConsumeAsync(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        public Task<OperateResult> ConsumeAsync(string Topic, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        {
+            return Task.Run(() => Consume(Topic, Type, AutoAck, Durable, Exclusive, AutoDelete));
+        }
+
+        /// <summary>
+        /// [消费]接收发布者消息
+        /// </summary>
+        private void Consumer_Received(object? sender, BasicDeliverEventArgs e, bool AutoAck = false)
         {
-            return Task.Run(() => Consume(MessageHead, Type, AutoAck, Durable, Exclusive, AutoDelete));
+            OnEventHandler?.Invoke(this, new RabbitMQData.Event() { MessageContent = Encoding.UTF8.GetString(e.Body.ToArray()), MessageHead = e.RoutingKey });
+            if (!AutoAck)
+            {
+                //当自动确认为false，得手动确认消息
+                Channels[e.RoutingKey].BasicAck(e.DeliveryTag, false);
+            }
         }
     }
 }

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQPublisherOperate.cs

@@ -1,212 +0,0 @@
-using RabbitMQ.Client;
-using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Channels;
-using System.Threading.Tasks;
-using YSAI.Core.data;
-using YSAI.Core.@interface.only;
-using YSAI.Core.@interface.unify;
-using YSAI.Unility;
-
-namespace YSAI.RabbitMQ
-{
-    /// <summary>
-    /// 发布者操作
-    /// </summary>
-    public sealed class RabbitMQPublisherOperate : IBaseAbstract, IRabbitMQPublisher
-    {
-        protected override string LogHead => "[ RabbitMQPublisherOperate 操作 ]";
-        protected override string ClassName => "RabbitMQPublisherOperate";
-
-        private static readonly object Lock = new object();  //锁
-        private static List<RabbitMQPublisherOperate> ThisObjList = new List<RabbitMQPublisherOperate>(); //自身对象集合
-
-        /// <summary>
-        /// 单例模式
-        /// </summary>
-        /// <returns></returns>
-        public static RabbitMQPublisherOperate Instance(RabbitMQData.Basics basics)
-        {
-            RabbitMQPublisherOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
-            if (exp == null)
-            {
-                lock (Lock)
-                {
-                    if (ThisObjList.Count(c => c.basics.Equals(basics)) > 0)
-                    {
-                        return ThisObjList.First(c => c.basics.Equals(basics));
-                    }
-                    else
-                    {
-                        RabbitMQPublisherOperate exp2 = new RabbitMQPublisherOperate(basics);
-                        ThisObjList.Add(exp2);
-                        return exp2;
-                    }
-                }
-            }
-            return exp;
-        }
-
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        private RabbitMQData.Basics basics { get; set; }
-
-        /// <summary>
-        /// 构造函数
-        /// </summary>
-        public RabbitMQPublisherOperate(RabbitMQData.Basics basics)
-        {
-            this.basics = basics;
-        }
-
-        /// <summary>
-        /// 连接对象
-        /// </summary>
-        private IConnection connection { get; set; }
-
-        /// <summary>
-        /// 通道集合
-        /// </summary>
-        private ConcurrentDictionary<string, IModel> Channels = new ConcurrentDictionary<string, IModel>();
-        /// <summary>
-        /// 队列与基础数据
-        /// </summary>
-        private ConcurrentDictionary<string, (QueueDeclareOk QueueDeclareOk, IBasicProperties BasicProperties)> QueueAndBasicProperties = new ConcurrentDictionary<string, (QueueDeclareOk QueueDeclareOk, IBasicProperties BasicProperties)>();
-
-        public Task<OperateResult> OnAsync()
-        {
-            return Task.Run(() => On());
-        }
-
-        public OperateResult On()
-        {
-            //开始记录运行时间
-            Depart("On");
-            try
-            {
-                if (connection == null || !connection.IsOpen)
-                {
-                    //连接工厂实例
-                    ConnectionFactory factory = new();
-                    //设置服务端参数
-                    factory.HostName = basics.HostName;
-                    factory.Port = basics.Port;
-                    factory.UserName = basics.UserName;
-                    factory.Password = basics.Password;
-                    //创建连接
-                    connection = factory.CreateConnection();
-                    //判断是否连接
-                    if (connection.IsOpen)
-                    {
-                        return Break("On", true);
-                    }
-                    return Break("On", false, "RabbitMQ连接打开失败");
-                }
-                else
-                {
-                    return Break("On", false, "RabbitMQ连接已打开");
-                }
-            }
-            catch (Exception ex)
-            {
-                return Break("On", false, "打开RabbitMQ连接异常：" + ex.Message);
-            }
-        }
-
-        public Task<OperateResult> OffAsync()
-        {
-            return Task.Run(() => Off());
-        }
-
-        public OperateResult Off()
-        {
-            //开始记录运行时间
-            Depart("Off");
-            try
-            {
-                if (connection == null || !connection.IsOpen)
-                {
-                    return Break("Off", false, "RabbitMQ连接未打开");
-                }
-                else
-                {
-                    //关闭
-                    connection.Close();
-                    //释放
-                    connection.Dispose();
-                    //清空通道集合
-                    Channels.Clear();
-                    //清空基础数据
-                    QueueAndBasicProperties.Clear();
-                    return Break("Off", true);
-                }
-            }
-            catch (Exception ex)
-            {
-                return Break("Off", false, "打开RabbitMQ连接异常：" + ex.Message);
-            }
-        }
-        public void Dispose()
-        {
-            Off();
-            GC.Collect();
-            GC.SuppressFinalize(this);
-            ThisObjList.Remove(this);
-        }
-
-        public OperateResult Publish(string MessageHead, string MessageContent, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
-        {
-            //开始记录运行时间
-            Depart("Publish");
-            try
-            {
-                if (connection == null || !connection.IsOpen)
-                {
-                    return Break("Publish", false, "RabbitMQ连接未打开");
-                }
-                else
-                {
-                    //判断是否存在此通道
-                    if (!Channels.ContainsKey(basics.ExChangeName))
-                    {
-                        IModel channel = connection.CreateModel();
-                        //创建交换机
-                        channel.ExchangeDeclare(basics.ExChangeName, Type, Durable, AutoDelete);
-                        //添加通道
-                        Channels.TryAdd(basics.ExChangeName, channel);
-                    }
-                    //判断队列与消息头是否存在
-                    if (!QueueAndBasicProperties.ContainsKey(MessageHead))
-                    {
-                        //创建队列
-                        QueueDeclareOk queueDeclareOk = Channels[basics.ExChangeName].QueueDeclare(MessageHead, Durable, Exclusive, AutoDelete);
-                        //绑定队列
-                        Channels[basics.ExChangeName].QueueBind(MessageHead, basics.ExChangeName, MessageHead);
-                        //消息持久化
-                        IBasicProperties basicProperties = Channels[basics.ExChangeName].CreateBasicProperties();
-                        basicProperties.Persistent = true;
-                        //添加消息头
-                        QueueAndBasicProperties.TryAdd(MessageHead, (queueDeclareOk, basicProperties));
-                    }
-                    // 发布消息
-                    Channels[basics.ExChangeName].BasicPublish(basics.ExChangeName, MessageHead, QueueAndBasicProperties[MessageHead].BasicProperties, Encoding.UTF8.GetBytes(MessageContent));
-                    //返回
-                    return Break("Publish", true);
-                }
-            }
-            catch (Exception ex)
-            {
-                return Break("Publish", false, "RabbitMQ发布异常：" + ex.Message);
-            }
-        }
-
-        public Task<OperateResult> PublishAsync(string MessageHead, string MessageContent, string Type = "topic", bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
-        {
-            return Task.Run(() => Publish(MessageHead, MessageContent, Type, Durable, Exclusive, AutoDelete));
-        }
-    }
-}

src/YSAI.DAQ/YSAI.Relay/RelayData.cs

@@ -0,0 +1,66 @@
+using System.ComponentModel;
+using System.Linq;
+using System.Text.Json.Serialization;
+using YSAI.Core.@interface.only;
+using YSAI.Kafka;
+using YSAI.Mqtt.client;
+using YSAI.RabbitMQ;
+
+namespace YSAI.Relay
+{
+    public class RelayData
+    {
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        public class Basics
+        {
+            /// <summary>
+            /// MQTT参数集合
+            /// </summary>
+            public List<MqttClientData.Basics>? MqttClientDataArray { get; set; }
+            /// <summary>
+            /// Kafka生产者参数集合
+            /// </summary>
+            public List<KafkaProducerData.Basics>? KafkaProducerDataArray { get; set; }
+            /// <summary>
+            /// Kafka消费者参数集合
+            /// </summary>
+            public List<KafkaConsumerData.Basics>? KafkaConsumerDataArray { get; set; }
+            /// <summary>
+            /// RabbitMQ 参数集合
+            /// </summary>
+            public List<RabbitMQData.Basics>? RabbitMQDataArray { get; set; }
+            /// <summary>
+            /// 重写基类中的Equals方法
+            /// </summary>
+            /// <param name="obj"></param>
+            /// <returns></returns>
+            public override bool Equals(object obj)
+            {
+                if (obj == null)
+                {
+                    return false;
+                }
+                Basics? Obj = obj as Basics;
+                if (Obj == null)
+                {
+                    return false;
+                }
+                else
+                {
+                    if (MqttClientDataArray.SequenceEqual(Obj.MqttClientDataArray) &&
+                        KafkaProducerDataArray.SequenceEqual(Obj.KafkaProducerDataArray) &&
+                        RabbitMQDataArray.SequenceEqual(Obj.RabbitMQDataArray))
+                    {
+                        return true;
+                    }
+                    else
+                    {
+                        return false;
+                    }
+                }
+            }
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Relay/RelayOperate.cs

@@ -0,0 +1,115 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+using YSAI.Core.@interface.unify;
+using YSAI.Kafka;
+using YSAI.Mqtt.client;
+using YSAI.RabbitMQ;
+
+namespace YSAI.Relay
+{
+    /// <summary>
+    /// 转发操作
+    /// </summary>
+    public class RelayOperate:IBaseAbstract,IRelay
+    {
+        protected override string LogHead => "[ RelayOperate 操作 ]";
+        protected override string ClassName => "RelayOperate";
+
+        private static readonly object Lock = new object();  //锁
+        private static List<RelayOperate> ThisObjList = new List<RelayOperate>(); //自身对象集合
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static RelayOperate Instance(RelayData.Basics basics)
+        {
+            RelayOperate? exp = ThisObjList.FirstOrDefault(c => c.basics.Equals(basics));
+            if (exp == null)
+            {
+                lock (Lock)
+                {
+                    if (ThisObjList.Count(c => c.basics.Equals(basics)) > 0)
+                    {
+                        return ThisObjList.First(c => c.basics.Equals(basics));
+                    }
+                    else
+                    {
+                        RelayOperate exp2 = new RelayOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+        /// <summary>
+        /// kafka消费者
+        /// </summary>
+        public ConcurrentDictionary<string, KafkaConsumerOperate> KafkaConsumerArray;
+        /// <summary>
+        /// kafka生产者
+        /// </summary>
+        public ConcurrentDictionary<string, KafkaProducerOperate> KafkaProducerArray;
+        /// <summary>
+        /// MQTT
+        /// </summary>
+        public ConcurrentDictionary<string, MqttClientOperate> MqttClientArray;
+        /// <summary>
+        /// RabbitMQ
+        /// </summary>
+        public ConcurrentDictionary<string, RabbitMQOperate> RabbitMQArray;
+
+
+
+
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        /// <param name="basics"></param>
+        public RelayOperate(RelayData.Basics basics)
+        {
+            this.basics = basics;
+            //当进入构造函数直接初始化
+
+        }
+
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private RelayData.Basics basics;
+
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            throw new NotImplementedException();
+        }
+
+        public OperateResult Consume(string Topic)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task<OperateResult> ConsumeAsync(string Topic)
+        {
+            throw new NotImplementedException();
+        }
+
+        public void Dispose()
+        {
+            throw new NotImplementedException();
+        }
+
+
+    }
+}

src/YSAI.DAQ/YSAI.Relay/YSAI.Relay.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <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/TestAll.cs

@@ -173,8 +173,7 @@ namespace YSAI.Test
             servers = OpcDaClientOperate.GetServersArray(Specification.COM_DA_30);
             Console.WriteLine();
 
-
-            using (OpcDaClientOperate opcDaClientOperate = OpcDaClientOperate.Instance(new OpcDaClientData.Basics() { SN = Guid.NewGuid().ToString(), ServiceName = "Knight.OPC.Server.Demo", SpecificationVer = Specification.COM_DA_20 }))
+            using (OpcDaClientOperate opcDaClientOperate = OpcDaClientOperate.Instance(new OpcDaClientData.Basics() { SN = Guid.NewGuid().ToString(), ServiceName = "Knight.OPC.Server.Demo", ApiVerType = OpcDaClientData.ApiVerType.COM_DA_20 }))
             {
                 OperateResult operateResult = opcDaClientOperate.On();
                 Console.WriteLine(operateResult.Message);
@@ -557,7 +556,7 @@ namespace YSAI.Test
             AddressManage addressManage = new AddressManage();
             addressManage.DType = Core.@enum.DaqType.OpcDa;
             addressManage.SN = "OPCDA²É¼¯";
-            addressManage.InstanceParam = new OpcDaClientData.Basics() { SN = "OPCDA²É¼¯", ServiceName = "Knight.OPC.Server.Demo", SpecificationVer = Specification.COM_DA_20 };
+            addressManage.InstanceParam = new OpcDaClientData.Basics() { SN = "OPCDA²É¼¯", ServiceName = "Knight.OPC.Server.Demo", ApiVerType = OpcDaClientData.ApiVerType.COM_DA_20 };
             addressManage.AddressArray = new List<AddressDetails>()
             {
                 new AddressDetails()