1

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

@@ -1,23 +1,9 @@
-using Microsoft.IdentityModel.Tokens;
-using Microsoft.VisualBasic;
-using Newtonsoft.Json.Linq;
-using System;
-using System.Collections.Generic;
-using System.Dynamic;
-using System.Linq;
-using System.Reflection;
-using System.Text;
-using System.Threading.Tasks;
-using System.Xml.Linq;
+using System.Dynamic;
 using YSAI.Core.@enum;
-using YSAI.Core.@interface.only;
 using YSAI.Core.reflection;
 using YSAI.Core.script;
 using YSAI.Log;
 using YSAI.Unility;
-using static YSAI.Core.reflection.ReflectionData.Basics.DllData.NamespaceData.ClassData;
-using static YSAI.Core.reflection.ReflectionData.Basics.DllData.NamespaceData;
-using YSAI.Core.@interface.unify;
 
 namespace YSAI.Core.data
 {

src/YSAI.DAQ/YSAI.Core/data/packet/InstancePacket.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace YSAI.Core.data.packet
+{
+    /// <summary>
+    /// 实例配置的包
+    /// </summary>
+    public class InstancePacket
+    {
+        /// <summary>
+        /// 完整的SN
+        /// 命名空间 + 类名.ISn.Daq.Config.json
+        /// 命名空间 + 类名.ISn.Relay.Config.json
+        /// </summary>
+        public string? SN { get; set; }
+        /// <summary>
+        /// 匹配SN的名称
+        /// </summary>
+        public string? MatchingSnName { get; set; }
+        /// <summary>
+        /// 实例参数
+        /// </summary>
+        public object InstanceParam { get; set; }
+    }
+}

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

@@ -9,9 +9,9 @@ using YSAI.Core.@interface.unify;
 namespace YSAI.Core.@interface.only
 {
     /// <summary>
-    /// 管理员接口
+    /// 生产/消费 者接口
     /// </summary>
-    public interface IKafkaAdmin
+    public interface IKafka: IDisposable
     {
         /// <summary>
         /// 创建主题，用于生产者消费者使用
@@ -24,45 +24,24 @@ namespace YSAI.Core.@interface.only
         /// </summary>
         /// <returns>统一出参</returns>
         OperateResult CreateTopics(List<string> Topics);
-    }
-    /// <summary>
-    /// 消费者接口
-    /// </summary>
-    public interface IKafkaConsumer
-    {
+
+
+
         /// <summary>
-        /// 单步 - 消费
+        /// 订阅 - 消费
         /// </summary>
         /// <param name="Topic">主题</param>
         /// <returns>统一出参</returns>
         Task<OperateResult> ConsumeAsync(string Topic);
 
         /// <summary>
-        /// 单步 - 消费
+        /// 订阅 - 消费
         /// </summary>
         /// <param name="Topic">主题</param>
         /// <returns>统一出参</returns>
         OperateResult Consume(string Topic);
-        /// <summary>
-        /// 轮询 - 消费
-        /// </summary>
-        /// <param name="Topic">主题集合</param>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> ConsumeAsync(List<string> Topics);
 
-        /// <summary>
-        /// 轮询 - 消费
-        /// </summary>
-        /// <param name="Topic">主题集合</param>
-        /// <returns>统一出参</returns>
-        OperateResult Consume(List<string> Topics);
 
-    }
-    /// <summary>
-    /// 生产者接口
-    /// </summary>
-    public interface IKafkaProducer: IRelay
-    {
         /// <summary>
         /// 生产
         /// </summary>

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

@@ -12,7 +12,7 @@ namespace YSAI.Core.@interface.only
     /// <summary>
     /// MQTT客户端接口
     /// </summary>
-    public interface IMqttClient : IRelay, IDisposable
+    public interface IMqttClient :IDisposable
     {
         /// <summary>
         /// 发布订阅

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

@@ -11,7 +11,7 @@ namespace YSAI.Core.@interface.only
     /// <summary>
     /// RabbitMQ接口
     /// </summary>
-    public interface IRabbitMQ : IRelay, IDisposable
+    public interface IRabbitMQ :IDisposable
     {
         /// <summary>
         /// 发布 一个交换机可以发布多个消息

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

@@ -0,0 +1,45 @@
+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 Subscribe(string Topic);
+        /// <summary>
+        /// 添加订阅
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <returns>统一结果</returns>
+        Task<OperateResult> SubscribeAsync(string Topic);
+
+        /// <summary>
+        /// 移除订阅
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <returns>统一结果</returns>
+        OperateResult UnSubscribe(string Topic);
+        /// <summary>
+        /// 移除订阅
+        /// </summary>
+        /// <param name="Topic">主题</param>
+        /// <returns>统一结果</returns>
+        Task<OperateResult> UnSubscribeAsync(string Topic);
+
+        
+    }
+
+
+}

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

@@ -9,7 +9,7 @@ namespace YSAI.Core.@interface.unify
     /// <summary>
     /// 转发接口
     /// </summary>
-    public interface IRelay : IOn, IOff, IProducer, IDisposable
+    public interface IRelay : IOn, IOff, IProducer, IConsumer, IDisposable
     {
 
     }

src/YSAI.DAQ/YSAI.Core/serialport/SerialPortOperate.cs

@@ -330,8 +330,9 @@ namespace YSAI.Core.serialport
                 //打开串口
                 serialPort.Open();
                 ResponseState = true;
+
                 //起线程，获取数据
-                ThreadPool.QueueUserWorkItem(o => ResponseData());
+                Task.Factory.StartNew(()=> ResponseData());
 
                 //启动定时器，解决数据粘包问题
                 ComResponseDataTimer.Interval = 100;   //设置毫秒间隔

src/YSAI.DAQ/YSAI.DB/DBOperate.cs

@@ -740,7 +740,7 @@ namespace YSAI.DB
                 if (subscribeOperate != null)
                 {
                     OperateResult operateResult = subscribeOperate.UnSubscribe(address);
-                    return Break("Subscribe", operateResult.State, operateResult.Message);
+                    return Break("UnSubscribe", operateResult.State, operateResult.Message);
                 }
                 else
                 {

src/YSAI.DAQ/YSAI.DaqManage/DaqManageData.cs

@@ -35,7 +35,7 @@ namespace YSAI.DaqManage
             /// 配置文件名称的格式，SN与配置数据中的SN一致
             /// 库配置：命名空间 + 类名.SN.Config.json
             /// </summary>
-            public string? ConfigFileNameFormat { get; set; } = "{0}.*.Config.json";
+            public string? ConfigFileNameFormat { get; set; } = "{0}.*.Daq.Config.json";
 
             /// <summary>
             /// 配置监控格式

src/YSAI.DAQ/YSAI.Kafka/KafkaAdminData.cs

@@ -1,61 +0,0 @@
-using Confluent.Kafka;
-using Newtonsoft.Json;
-using Newtonsoft.Json.Converters;
-using System;
-using System.Collections.Generic;
-using System.ComponentModel;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace YSAI.Kafka
-{
-    public class KafkaAdminData
-    {
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        public class Basics : KafkaBaseData
-        {
-            /// <summary>
-            /// 安全协议
-            /// </summary>
-            [Description("安全协议")]
-            [JsonConverter(typeof(StringEnumConverter))]
-            public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.Plaintext;
-            /// <summary>
-            /// 新主题的分区数 默认-1
-            /// </summary>
-            [Description("主题分区数")]
-            public int NumPartitions { get; set; } = -1;
-            /// <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 (BootstrapServers == Obj.BootstrapServers && SecurityProtocol == Obj.SecurityProtocol)
-                    {
-                        return true;
-                    }
-                    else
-                    {
-                        return false;
-                    }
-                }
-            }
-        }
-    }
-}

src/YSAI.DAQ/YSAI.Kafka/KafkaAdminOperate.cs

@@ -1,123 +0,0 @@
-using Confluent.Kafka;
-using Confluent.Kafka.Admin;
-using System;
-using System.Collections;
-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.only;
-using YSAI.Core.@interface.unify;
-using YSAI.Log;
-using YSAI.Unility;
-
-namespace YSAI.Kafka
-{
-    /// <summary>
-    /// 管理员创建主题使用
-    /// </summary>
-    public sealed class KafkaAdminOperate:IBaseAbstract,IKafkaAdmin
-    {
-        private static readonly object Lock = new object();  //锁
-        private static List<KafkaAdminOperate> ThisObjList = new List<KafkaAdminOperate>(); //自身对象集合
-        /// <summary>
-        /// 单例模式
-        /// </summary>
-        /// <returns></returns>
-        public static KafkaAdminOperate Instance(KafkaAdminData.Basics basics)
-        {
-            KafkaAdminOperate? 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
-                    {
-                        KafkaAdminOperate exp2 = new KafkaAdminOperate(basics);
-                        ThisObjList.Add(exp2);
-                        return exp2;
-                    }
-                }
-            }
-            return exp;
-        }
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        private KafkaAdminData.Basics basics { get; set; }
-
-        /// <summary>
-        /// 构造函数
-        /// </summary>
-        public KafkaAdminOperate(KafkaAdminData.Basics basics)
-        {
-            this.basics = basics;
-        }
-        /// <summary>
-        /// 客户端管理员配置
-        /// </summary>
-        private AdminClientConfig adminClientConfig { get; set; }
-
-        public void Dispose()
-        {
-            GC.Collect();
-            GC.SuppressFinalize(this);
-            ThisObjList.Remove(this);
-        }
-        protected override string LogHead => "[ KafkaAdminOperate 操作 ]";
-        protected override string ClassName => "KafkaAdminOperate";
-        public Task<OperateResult> CreateTopicsAsync(List<string> Topics)
-        {
-            return Task.Run(() => CreateTopics(Topics));
-        }
-
-        public OperateResult CreateTopics(List<string> Topics)
-        {
-            //开始记录运行时间
-            Depart("CreateTopics");
-            try
-            {
-                if (adminClientConfig == null)
-                {
-                    //创建配置
-                    adminClientConfig = new AdminClientConfig() { BootstrapServers = basics.BootstrapServers, SecurityProtocol = basics.SecurityProtocol };
-                }
-                using (IAdminClient adminClient = new AdminClientBuilder(adminClientConfig).Build())
-                {
-                    //先获取kafka已存在的主题
-                    Metadata metadata = adminClient.GetMetadata(TimeSpan.FromSeconds(5));  //超时时间5秒
-                    //已存在的主题集合
-                    List<string> adnimTopics = new List<string>();
-                    foreach (var item in metadata.Topics) { adnimTopics.Add(item.Topic); }
-                    //得到主题差异集合
-                    List<string> topicDifferenceArray = Topics.Except(adnimTopics).ToList();
-                    //差异主题集合
-                    List<TopicSpecification> topicSpecifications = new List<TopicSpecification>();
-                    //循环添加主题
-                    foreach (var topic in topicDifferenceArray)
-                    {
-                        topicSpecifications.Add(new TopicSpecification()
-                        {
-                            Name = topic,
-                            NumPartitions = basics.NumPartitions
-                        });
-                    }
-                    //执行创建
-                    adminClient.CreateTopicsAsync(topicSpecifications).Wait();
-                    //返回
-                    return Break("CreateTopics", true, "主题集合创建成功");
-                }
-            }
-            catch (Exception ex)
-            {
-                return Break("CreateTopics", false, ex.Message);
-            }
-        }
-    }
-}

src/YSAI.DAQ/YSAI.Kafka/KafkaBaseData.cs

@@ -10,16 +10,7 @@ namespace YSAI.Kafka
 {
     public class KafkaBaseData
     {
-        /// <summary>
-        /// 唯一标识符
-        /// </summary>
-        [Description("唯一标识符")]
-        public string? SN { get; set; } = Guid.NewGuid().ToString();
-        /// <summary>
-        /// 服务器地址
-        /// </summary>
-        [Description("服务器地址")]
-        public string BootstrapServers { get; set; }
+       
 
 
         //public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.Plaintext;

src/YSAI.DAQ/YSAI.Kafka/KafkaConsumerOperate.cs

@@ -1,245 +0,0 @@
-
-using Confluent.Kafka;
-using Newtonsoft.Json.Linq;
-using System.Collections.Concurrent;
-using System.Xml.Linq;
-using YSAI.Core.data;
-using YSAI.Core.@interface.only;
-using YSAI.Core.@interface.unify;
-using YSAI.Log;
-using YSAI.Unility;
-
-namespace YSAI.Kafka
-{
-    public sealed class KafkaConsumerOperate : IBaseAbstract, IKafkaConsumer
-    {
-        private static readonly object Lock = new object();  //锁
-        private static List<KafkaConsumerOperate> ThisObjList = new List<KafkaConsumerOperate>(); //自身对象集合
-        /// <summary>
-        /// 单例模式
-        /// </summary>
-        /// <returns></returns>
-        public static KafkaConsumerOperate Instance(KafkaConsumerData.Basics basics)
-        {
-            KafkaConsumerOperate? 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
-                    {
-                        KafkaConsumerOperate exp2 = new KafkaConsumerOperate(basics);
-                        ThisObjList.Add(exp2);
-                        return exp2;
-                    }
-                }
-            }
-            return exp;
-        }
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        private KafkaConsumerData.Basics basics { get; set; }
-
-        /// <summary>
-        /// 构造函数
-        /// </summary>
-        public KafkaConsumerOperate(KafkaConsumerData.Basics basics)
-        {
-            this.basics = basics;
-        }
-
-
-        /// <summary>
-        /// 消费者配置
-        /// </summary>
-        private ConsumerConfig consumerConfig { get; set; }
-
-
-        /// <summary>
-        /// 订阅数据集合
-        /// KEY:组ID 
-        /// VALUE:(对象,主题名集合)
-        /// </summary>
-        private ConcurrentDictionary<string, (IConsumer<object, object> ConsumerObj, List<string> Topics)> SubscribeDataArray { get; set; }
-        /// <summary>
-        /// 轮询状态
-        /// </summary>
-        private bool PollingState = false;
-
-      
-
-        /// <summary>
-        /// 全局信息事件中状态抛出
-        /// </summary>
-        public event EventHandler<KafkaConsumerData.Event<object, object>> OnEvent
-        {
-            add { OnEventHandler += value; }
-            remove { OnEventHandler -= value; }
-        }
-        private EventHandler<KafkaConsumerData.Event<object, object>>? OnEventHandler;
-
-        /// <summary>
-        /// 轮询消费
-        /// </summary>
-        /// <returns></returns>
-        private async Task PollingConsume(IConsumer<object, object> consumer)
-        {
-            try
-            {
-                while (PollingState)
-                {
-                    //超时时间1秒
-                    ConsumeResult<object, object>? result = consumer?.Consume(TimeSpan.FromSeconds(1));
-
-                    if (result != null)
-                    {
-                        OnEventHandler?.Invoke(this, new KafkaConsumerData.Event<object, object>() { State = true, Key = result.Message.Key, Value = result.Message.Value, Topic = result.Topic, Message = $"收到主题:{result.Topic} - Key:{result.Message.Key} - Value:{result.Message.Value}" });
-                        //消息已处理提交偏移量
-                        consumer?.Commit();
-                    }
-                }
-            }
-            catch (Exception ex)
-            {
-                OnEventHandler?.Invoke(this, new KafkaConsumerData.Event<object, object>() { State = false, Message = $"轮询消费异常:{ex.Message}" });
-            }
-        }
-
-        public void Dispose()
-        {
-            GC.Collect();
-            GC.SuppressFinalize(this);
-            ThisObjList.Remove(this);
-        }
-        protected override string LogHead => "[ KafkaConsumerOperate 操作 ]";
-        protected override string ClassName => "KafkaConsumerOperate";
-        public Task<OperateResult> ConsumeAsync(string Topic)
-        {
-            return Task.Run(()=> Consume(Topic));
-        }
-        public OperateResult Consume(string Topic)
-        {
-            //开始记录运行时间
-            Depart("Consume");
-            try
-            {
-                //实例化订阅数据集合
-                if (SubscribeDataArray == null)
-                {
-                    SubscribeDataArray = new ConcurrentDictionary<string, (IConsumer<object, object>, List<string>)>();
-                }
-
-                if (consumerConfig == null)
-                {
-                    //创建配置
-                    consumerConfig = new ConsumerConfig() { BootstrapServers = basics.BootstrapServers, GroupId = basics.GroupId, AutoOffsetReset = basics.AutoOffsetReset };
-                }
-
-                //创建一个实例
-                IConsumer<object, object> consumer = new ConsumerBuilder<object, object>(consumerConfig).Build();
-
-                //订阅多个主题
-                consumer.Subscribe(Topic);
-
-                //超时时间1秒
-                ConsumeResult<object, object>? result = consumer?.Consume(TimeSpan.FromSeconds(1));
-                //状态
-                bool state = false;
-                //消息信息
-                string message = string.Empty;
-
-                if (result != null)
-                {
-                    state = true;
-                    message = $"消费成功，收到主题:{result.Topic} - Key:{result.Message.Key} - Value:{result.Message.Value}";
-                    OnEventHandler?.Invoke(this, new KafkaConsumerData.Event<object, object>() { State = state, Key = result.Message.Key, Value = result.Message.Value, Topic = result.Topic, Message = message });
-                    //消息已处理提交偏移量
-                    consumer?.Commit();
-                }
-                else
-                {
-                    state = true;
-                    message = $"消费失败，未收到消息，执行消费未返回数据";
-                    OnEventHandler?.Invoke(this, new KafkaConsumerData.Event<object, object>() { State = state, Message = message });
-                }
-                //先关闭此组的订阅
-                consumer?.Unsubscribe();
-                //关闭
-                consumer?.Close();
-                //释放掉这个实例
-                consumer?.Dispose();
-
-                return Break("Consume", state, message);
-            }
-            catch (Exception ex)
-            {
-                return Break("Consume", false, ex.Message);
-            }
-        }
-        public Task<OperateResult> ConsumeAsync(List<string> Topics)
-        {
-            return Task.Run(() => Consume(Topics));
-        }
-        public OperateResult Consume(List<string> Topics)
-        {
-            //开始记录运行时间
-            Depart("Consume");
-            try
-            {
-                //实例化订阅数据集合
-                if (SubscribeDataArray == null)
-                {
-                    SubscribeDataArray = new ConcurrentDictionary<string, (IConsumer<object, object>, List<string>)>();
-                }
-
-                if (consumerConfig == null)
-                {
-                    //创建配置
-                    consumerConfig = new ConsumerConfig() { BootstrapServers = basics.BootstrapServers, GroupId = basics.GroupId, AutoOffsetReset = basics.AutoOffsetReset };
-                }
-
-                //判断是否存在组
-                if (SubscribeDataArray.ContainsKey(basics.GroupId))
-                {
-                    //设置轮询状态
-                    PollingState = false;
-                    //先关闭此组的订阅
-                    SubscribeDataArray[basics.GroupId].ConsumerObj.Unsubscribe();
-                    //关闭
-                    SubscribeDataArray[basics.GroupId].ConsumerObj.Close();
-                    //释放掉这个实例
-                    SubscribeDataArray[basics.GroupId].ConsumerObj.Dispose();
-                }
-
-                //创建一个实例
-                IConsumer<object, object> consumer = new ConsumerBuilder<object, object>(consumerConfig).Build();
-
-                //订阅多个主题
-                consumer.Subscribe(Topics);
-
-                //设置轮询状态
-                PollingState = true;
-
-                //异步轮询
-                PollingConsume(consumer);
-
-                //设置值
-                (IConsumer<object, object>, List<string>) value = (consumer, Topics);
-
-                //存在更新value 反之添加
-                SubscribeDataArray.AddOrUpdate(basics.GroupId, value, (k, v) => value);
-
-                return Break("Consume", true, "消费轮询已执行");
-            }
-            catch (Exception ex)
-            {
-                return Break("Consume", false, ex.Message);
-            }
-        }
-    }
-}

src/YSAI.DAQ/YSAI.Kafka/KafkaConsumerData.cs

@@ -10,18 +10,37 @@ using System.Threading.Tasks;
 
 namespace YSAI.Kafka
 {
-    public class KafkaConsumerData
+    public class KafkaData
     {
         /// <summary>
         /// 基础数据
         /// </summary>
-        public class Basics : KafkaBaseData
+        public class Basics
         {
             /// <summary>
-            /// 组ID
+            /// 唯一标识符
             /// </summary>
-            [Description("组ID")]
-            public string GroupId { get; set; }
+            [Description("唯一标识符")]
+            public string? SN { get; set; } = Guid.NewGuid().ToString();
+
+            /// <summary>
+            /// 服务器地址
+            /// </summary>
+            [Description("服务器地址")]
+            public string? BootstrapServers { get; set; }
+
+            /// <summary>
+            /// 最多等待时间（毫秒）
+            /// </summary>
+            [Description("等待时间")]
+            public int WaitTime { get; set; } = 1000;
+
+            /// <summary>
+            /// 安全协议
+            /// </summary>
+            [Description("安全协议")]
+            [JsonConverter(typeof(StringEnumConverter))]
+            public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.Plaintext;
 
             /// <summary>
             /// 自动偏移复位
@@ -29,6 +48,7 @@ namespace YSAI.Kafka
             [Description("自动偏移复位")]
             [JsonConverter(typeof(StringEnumConverter))]
             public AutoOffsetReset AutoOffsetReset { get; set; } = AutoOffsetReset.Earliest;
+
             /// <summary>
             /// 重写基类中的Equals方法
             /// </summary>
@@ -47,7 +67,11 @@ namespace YSAI.Kafka
                 }
                 else
                 {
-                    if (BootstrapServers == Obj.BootstrapServers && GroupId == Obj.GroupId)
+                    if (SN == Obj.SN &&
+                        BootstrapServers == Obj.BootstrapServers &&
+                        WaitTime == Obj.WaitTime &&
+                        SecurityProtocol == Obj.SecurityProtocol &&
+                         AutoOffsetReset == Obj.AutoOffsetReset)
                     {
                         return true;
                     }
@@ -58,10 +82,11 @@ namespace YSAI.Kafka
                 }
             }
         }
+
         /// <summary>
-        /// 事件响应数据
+        /// 消费订阅得到的数据
         /// </summary>
-        public class Event<K, V>
+        public class ConsumeData<K, V>
         {
             /// <summary>
             /// 消息
@@ -88,6 +113,5 @@ namespace YSAI.Kafka
             /// </summary>
             public DateTime Time { get; set; } = DateTime.Now.ToLocalTime();
         }
-    
     }
 }

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

@@ -0,0 +1,400 @@
+using Confluent.Kafka;
+using Confluent.Kafka.Admin;
+using Microsoft.IdentityModel.Tokens;
+using Newtonsoft.Json.Linq;
+using System.Collections.Concurrent;
+using System.Data;
+using YSAI.Core.data;
+using YSAI.Core.@interface.only;
+using YSAI.Core.@interface.unify;
+using YSAI.Log;
+using YSAI.Unility;
+
+namespace YSAI.Kafka
+{
+    public sealed class KafkaOperate : IBaseAbstract, IKafka, IRelay
+    {
+        protected override string LogHead => "[ KafkaOperate 操作 ]";
+        protected override string ClassName => "KafkaOperate";
+        private static readonly object Lock = new object();  //锁
+        private static List<KafkaOperate> ThisObjList = new List<KafkaOperate>(); //自身对象集合
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static KafkaOperate Instance(KafkaData.Basics basics)
+        {
+            KafkaOperate? 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
+                    {
+                        KafkaOperate exp2 = new KafkaOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private KafkaData.Basics basics { get; set; }
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        public KafkaOperate(KafkaData.Basics basics)
+        {
+            this.basics = basics;
+
+            if (producerConfig == null)
+            {
+                //创建配置
+                producerConfig = new ProducerConfig()
+                {
+                    ClientId = Guid.NewGuid().ToString(),
+                    BootstrapServers = basics.BootstrapServers
+                    //,
+                    //SecurityProtocol = basics.SecurityProtocol,
+                    //SaslMechanism = basics.SaslMechanism,
+                    //SaslKerberosServiceName = basics.SaslKerberosServiceName,
+                    //SaslKerberosKeytab = basics.SaslKerberosKeytab,
+                    //SaslKerberosPrincipal = basics.SaslKerberosPrincipal
+                };
+            }
+
+            if (consumerConfig == null)
+            {
+                //创建配置
+                consumerConfig = new ConsumerConfig() 
+                { 
+                    BootstrapServers = basics.BootstrapServers, 
+                    GroupId = Guid.NewGuid().ToString(), 
+                    AutoOffsetReset = basics.AutoOffsetReset
+                };
+            }
+
+            if (adminClientConfig == null)
+            {
+                //创建配置
+                adminClientConfig = new AdminClientConfig() 
+                { 
+                    BootstrapServers = basics.BootstrapServers, 
+                    SecurityProtocol = basics.SecurityProtocol 
+                };
+            }
+        }
+      
+
+        /// <summary>
+        /// 生产者配置
+        /// </summary>
+        private ProducerConfig producerConfig { get; set; }
+        /// <summary>
+        /// 消费者配置
+        /// </summary>
+        private ConsumerConfig consumerConfig { get; set; }
+        /// <summary>
+        /// 客户端管理员配置
+        /// </summary>
+        private AdminClientConfig adminClientConfig { get; set; }
+
+        public void Dispose()
+        {
+            GC.Collect();
+            GC.SuppressFinalize(this);
+            ThisObjList.Remove(this);
+        }
+
+        /// <summary>
+        /// 主题集合
+        /// </summary>
+        private List<string> TopicArray = new List<string>();
+        /// <summary>
+        /// 订阅的消费者
+        /// </summary>
+        private IConsumer<object, object>? Consumer = null;
+        /// <summary>
+        /// 轮询状态
+        /// </summary>
+        private bool PollingState = false;
+
+        /// <summary>
+        /// 轮询消费
+        /// </summary>
+        /// <returns></returns>
+        private Task Polling()
+        {
+            return Task.Run(() =>
+            {
+                try
+                {
+                    while (PollingState)
+                    {
+                        //超时时间1秒
+                        ConsumeResult<object, object>? result = Consumer?.Consume(new TimeSpan(0, 0, 0, 0, basics.WaitTime));
+
+                        if (result != null)
+                        {
+                            OnEventHandler?.Invoke(this, new EventResult(true, $"消费成功，收到主题:{result.Topic} - Key:{result.Message.Key} - Value:{result.Message.Value}", new KafkaData.ConsumeData<object, object>() { State = true, Key = result.Message.Key, Value = result.Message.Value, Topic = result.Topic, Message = $"收到主题:{result.Topic} - Key:{result.Message.Key} - Value:{result.Message.Value}" }, Core.@enum.ResultType.Dynamic));
+                            //消息已处理提交偏移量
+                            Consumer?.Commit();
+                        }
+                    }
+                }
+                catch (Exception ex)
+                {
+                    OnEventHandler?.Invoke(this, new EventResult(false, $"轮询消费异常:{ex.Message}"));
+                }
+            });
+        }
+
+        public Task<OperateResult> CreateTopicsAsync(List<string> Topics)
+        {
+            return Task.Run(() => CreateTopics(Topics));
+        }
+
+        public OperateResult CreateTopics(List<string> Topics)
+        {
+            //开始记录运行时间
+            Depart("CreateTopics");
+            try
+            {
+                using (IAdminClient adminClient = new AdminClientBuilder(adminClientConfig).Build())
+                {
+                    //先获取kafka已存在的主题
+                    Metadata metadata = adminClient.GetMetadata(TimeSpan.FromSeconds(5));  //超时时间5秒
+                    //已存在的主题集合
+                    List<string> adnimTopics = new List<string>();
+                    foreach (var item in metadata.Topics) { adnimTopics.Add(item.Topic); }
+                    //得到主题差异集合
+                    List<string> topicDifferenceArray = Topics.Except(adnimTopics).ToList();
+                    //差异主题集合
+                    List<TopicSpecification> topicSpecifications = new List<TopicSpecification>();
+                    //循环添加主题
+                    foreach (var topic in topicDifferenceArray)
+                    {
+                        topicSpecifications.Add(new TopicSpecification()
+                        {
+                            Name = topic
+                        });
+                    }
+                    //执行创建
+                    adminClient.CreateTopicsAsync(topicSpecifications).Wait();
+                    //返回
+                    return Break("CreateTopics", true, "主题集合创建成功");
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("CreateTopics", false, ex.Message);
+            }
+        }
+
+        public OperateResult Produce(string Topic, object Key, object Content)
+        {
+            //开始记录运行时间
+            Depart("Produce");
+            try
+            {
+                //失败的消息
+                List<string> MessageFail = new List<string>();
+              
+                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 =>
+                    {
+                        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)
+                    {
+                        result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败");
+                    }
+                    if (!result.State)
+                    {
+                        MessageFail.Add(result.Message);
+                    }
+                }
+                if (MessageFail.Count > 0)
+                {
+                    return Break("Produce", false, $"批量发送失败，存在发送失败数据：{MessageFail.ToJson()}");
+                }
+                else
+                {
+                    return Break("Produce", true, $"批量发送成功");
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Produce", false, ex.Message);
+            }
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, object Key, object Content)
+        {
+            return Task.Run(() => Produce(Topic, Key, Content));
+        }
+
+        public OperateResult Produce(string Topic, string Content)
+        {
+            Depart("Produce");
+            OperateResult operateResult = Produce(Topic, null, Content);
+            return Break("Produce", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> ProduceAsync(string Topic, string Content)
+        {
+            return Task.Run(() => Produce(Topic, Content));
+        }
+
+        public Task<OperateResult> OnAsync()
+        {
+            return Task.Run(() => On());
+        }
+
+        public OperateResult On()
+        {
+            return Break(Depart("On"), true);
+        }
+
+        public Task<OperateResult> OffAsync()
+        {
+            return Task.Run(() => Off());
+        }
+
+        public OperateResult Off()
+        {
+            return Break(Depart("Off"), true);
+        }
+
+        public Task<OperateResult> ConsumeAsync(string Topic)
+        {
+            return Task.Run(() => Consume(Topic));
+        }
+
+        public OperateResult Consume(string Topic)
+        {
+            //开始记录运行时间
+            Depart("Consume");
+            try
+            {
+                //创建一个实例
+                if (Consumer == null)
+                {
+                    Consumer = new ConsumerBuilder<object, object>(consumerConfig).Build();
+                }
+
+                if (TopicArray.Contains(Topic))
+                {
+                    return Break("Consume", false, $"{Topic} 此主题已订阅");
+                }
+                else
+                {
+                    //添加主题
+                    TopicArray.Add(Topic);
+
+                    //订阅多个主题，会自动取消订阅之前的主题
+                    Consumer.Subscribe(TopicArray);
+
+                    //当订阅状态为false 调用轮询函数
+                    if (!PollingState)
+                    {
+                        //设置轮询状态
+                        PollingState = true;
+
+                        //异步轮询
+                        Polling();
+                    }
+
+                    return Break("Consume", true);
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Consume", false, ex.Message);
+            }
+        }
+
+        public OperateResult Subscribe(string Topic)
+        {
+            Depart("Subscribe");
+            OperateResult operateResult = Consume(Topic);
+            return Break("Subscribe", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> SubscribeAsync(string Topic)
+        {
+            return Task.Run(()=> Subscribe(Topic));
+        }
+
+        public OperateResult UnSubscribe(string Topic)
+        {
+            Depart("UnSubscribe");
+            try
+            {
+                if (!TopicArray.Contains(Topic))
+                {
+                    return Break("UnSubscribe", false, $"{Topic} 此主题不存在");
+                }
+                else
+                {
+                    //移除这个主题
+                    TopicArray.Remove(Topic);
+
+                    //主题数量判断
+                    if (TopicArray.Count > 0)
+                    {
+                        //大于零直接修改订阅
+                        //订阅多个主题，会自动取消订阅之前的主题
+                        Consumer.Subscribe(TopicArray);
+                    }
+                    else
+                    {
+                        //小于或等于零，直接关闭订阅
+                        //设置轮询状态
+                        PollingState = false;
+                        //先关订阅
+                        Consumer.Unsubscribe();
+                        //关闭
+                        Consumer.Close();
+                        //释放掉这个实例
+                        Consumer.Dispose();
+                        //置空
+                        Consumer = null;
+                    }
+
+                    return Break("UnSubscribe", true);
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("UnSubscribe", false, ex.Message);
+            }
+        }
+
+        public Task<OperateResult> UnSubscribeAsync(string Topic)
+        {
+            return Task.Run(() => UnSubscribe(Topic));
+        }
+    }
+}

src/YSAI.DAQ/YSAI.Kafka/KafkaProducerData.cs

@@ -1,60 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.ComponentModel;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace YSAI.Kafka
-{
-    public class KafkaProducerData
-    {
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        public class Basics:KafkaBaseData
-        {
-            /// <summary>
-            /// 客户端ID
-            /// </summary>
-            [Description("客户端ID")]
-            public string ClientId { get; set; }
-            /// <summary>
-            /// 最多等待时间
-            /// </summary>
-            [Description("等待时间")]
-            public int WaitTime { get; set; } = 1000;
-
-           
-
-            /// <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 (BootstrapServers == Obj.BootstrapServers&& ClientId==Obj.ClientId)
-                    {
-                        return true;
-                    }
-                    else
-                    {
-                        return false;
-                    }
-                }
-            }
-        }
-    }
-}

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

@@ -1,169 +0,0 @@
-using Confluent.Kafka;
-using Newtonsoft.Json.Linq;
-using System.Data;
-using YSAI.Core.data;
-using YSAI.Core.@interface.only;
-using YSAI.Core.@interface.unify;
-using YSAI.Log;
-using YSAI.Unility;
-
-namespace YSAI.Kafka
-{
-    public sealed class KafkaProducerOperate : IBaseAbstract, IKafkaProducer, IRelay
-    {
-        private static readonly object Lock = new object();  //锁
-        private static List<KafkaProducerOperate> ThisObjList = new List<KafkaProducerOperate>(); //自身对象集合
-        /// <summary>
-        /// 单例模式
-        /// </summary>
-        /// <returns></returns>
-        public static KafkaProducerOperate Instance(KafkaProducerData.Basics basics)
-        {
-            KafkaProducerOperate? 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
-                    {
-                        KafkaProducerOperate exp2 = new KafkaProducerOperate(basics);
-                        ThisObjList.Add(exp2);
-                        return exp2;
-                    }
-                }
-            }
-            return exp;
-        }
-        /// <summary>
-        /// 基础数据
-        /// </summary>
-        private KafkaProducerData.Basics basics { get; set; }
-
-        /// <summary>
-        /// 构造函数
-        /// </summary>
-        public KafkaProducerOperate(KafkaProducerData.Basics basics)
-        {
-            this.basics = basics;
-        }
-
-        /// <summary>
-        /// 生产者配置
-        /// </summary>
-        private ProducerConfig producerConfig { get; set; }
-
-        public void Dispose()
-        {
-            GC.Collect();
-            GC.SuppressFinalize(this);
-            ThisObjList.Remove(this);
-        }
-        protected override string LogHead => "[ KafkaProducerOperate 操作 ]";
-        protected override string ClassName => "KafkaProducerOperate";
-        public OperateResult Produce(string Topic, object Key, object Content)
-        {
-            //开始记录运行时间
-            Depart("Produce");
-            try
-            {
-                //失败的消息
-                List<string> MessageFail = new List<string>();
-                if (producerConfig == null)
-                {
-                    //创建配置
-                    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())
-                {
-                    //返回结果
-                    OperateResult? result = null;
-                    //发布结果
-                    producer.Produce(Topic, new Message<object, object>() { Key = Key, Value = Content }, 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)
-                    {
-                        result = Break("Produce", false, $"主题 [ {Topic} ] 发送失败");
-                    }
-                    if (!result.State)
-                    {
-                        MessageFail.Add(result.Message);
-                    }
-                }
-                if (MessageFail.Count > 0)
-                {
-                    return Break("Produce", false, $"批量发送失败，存在发送失败数据：{MessageFail.ToJson()}");
-                }
-                else
-                {
-                    return Break("Produce", true, $"批量发送成功");
-                }
-            }
-            catch (Exception ex)
-            {
-                return Break("Produce", false, ex.Message);
-            }
-        }
-
-        public Task<OperateResult> ProduceAsync(string Topic, object Key, object Content)
-        {
-            return Task.Run(() => Produce(Topic, Key, Content));
-        }
-
-        public OperateResult Produce(string Topic, string Content)
-        {
-            Depart("Produce");
-            OperateResult operateResult = Produce(Topic, null, Content);
-            return Break("Produce", operateResult.State, operateResult.Message);
-        }
-
-        public Task<OperateResult> ProduceAsync(string Topic, string Content)
-        {
-            return Task.Run(() => Produce(Topic, Content));
-        }
-
-        public Task<OperateResult> OnAsync()
-        {
-            return Task.Run(() => On());
-        }
-
-        public OperateResult On()
-        {
-            return Break(Depart("On"), true);
-        }
-
-        public Task<OperateResult> OffAsync()
-        {
-            return Task.Run(() => Off());
-        }
-
-        public OperateResult Off()
-        {
-            return Break(Depart("Off"), true);
-        }
-    }
-}

src/YSAI.DAQ/YSAI.Manage/Controllers/OperateController.cs

@@ -60,7 +60,7 @@ namespace YSAI.Manage.Controllers
                 case LibType.S7:
                     return new S7ClientData.Basics();
                 case LibType.Kafka:
-                    return new KafkaProducerData.Basics();
+                    return new KafkaData.Basics();
                 case LibType.Mqtt:
                     return new MqttClientData.Basics();
                 case LibType.RabbitMQ:
@@ -95,7 +95,7 @@ namespace YSAI.Manage.Controllers
                 case LibType.S7:
                     return new OperateResult(true, string.Empty, 0.1, RData: ReflexTool.GetClassAllPropertyData<S7ClientData.Basics>(), RType: ResultType.Json);
                 case LibType.Kafka:
-                    return new OperateResult(true, string.Empty, 0.1, RData: ReflexTool.GetClassAllPropertyData<KafkaProducerData.Basics>(), RType: ResultType.Json);
+                    return new OperateResult(true, string.Empty, 0.1, RData: ReflexTool.GetClassAllPropertyData<KafkaData.Basics>(), RType: ResultType.Json);
                 case LibType.Mqtt:
                     return new OperateResult(true, string.Empty, 0.1, RData: ReflexTool.GetClassAllPropertyData<MqttClientData.Basics>(), RType: ResultType.Json);
                 case LibType.RabbitMQ:

src/YSAI.DAQ/YSAI.Manage/Properties/PublishProfiles/FolderProfile.pubxml

@@ -4,7 +4,7 @@ https://go.microsoft.com/fwlink/?LinkID=208121.
 -->
 <Project>
   <PropertyGroup>
-    <DeleteExistingFiles>false</DeleteExistingFiles>
+    <DeleteExistingFiles>true</DeleteExistingFiles>
     <ExcludeApp_Data>false</ExcludeApp_Data>
     <LaunchSiteAfterPublish>true</LaunchSiteAfterPublish>
     <LastUsedBuildConfiguration>Release</LastUsedBuildConfiguration>
@@ -13,5 +13,9 @@ https://go.microsoft.com/fwlink/?LinkID=208121.
     <PublishUrl>bin\Release\net6.0\publish\</PublishUrl>
     <WebPublishMethod>FileSystem</WebPublishMethod>
     <_TargetId>Folder</_TargetId>
+    <SiteUrlToLaunchAfterPublish />
+    <TargetFramework>net6.0</TargetFramework>
+    <ProjectGuid>6e9667c0-303c-472d-949b-60f812e6c659</ProjectGuid>
+    <SelfContained>false</SelfContained>
   </PropertyGroup>
 </Project>

src/YSAI.DAQ/YSAI.Modbus/client/ModbusClientOperate.cs

@@ -916,7 +916,7 @@ namespace YSAI.Modbus.client
                 if (subscribeOperate != null)
                 {
                     OperateResult operateResult = subscribeOperate.UnSubscribe(address);
-                    return Break("Subscribe", operateResult.State, operateResult.Message);
+                    return Break("UnSubscribe", operateResult.State, operateResult.Message);
                 }
                 else
                 {

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

@@ -276,5 +276,29 @@ namespace YSAI.Mqtt.client
         {
             return Task.Run(() => Produce(Topic, Content));
         }
+
+        public OperateResult Subscribe(string Topic)
+        {
+            Depart("Subscribe");
+            OperateResult operateResult = AddSubscribe(Topic);
+            return Break("Subscribe", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> SubscribeAsync(string Topic)
+        {
+            return Task.Run(() => Subscribe(Topic));
+        }
+
+        public OperateResult UnSubscribe(string Topic)
+        {
+            Depart("UnSubscribe");
+            OperateResult operateResult = RemoveSubscribe(Topic);
+            return Break("UnSubscribe", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> UnSubscribeAsync(string Topic)
+        {
+            return Task.Run(() => UnSubscribe(Topic));
+        }
     }
 }

src/YSAI.DAQ/YSAI.Opc/da/http/OpcDaHttpOperate.cs

@@ -409,7 +409,7 @@ namespace YSAI.Opc.da.http
                 if (subscribeOperate != null)
                 {
                     OperateResult operateResult = subscribeOperate.UnSubscribe(address);
-                    return Break("Subscribe", operateResult.State, operateResult.Message);
+                    return Break("UnSubscribe", operateResult.State, operateResult.Message);
                 }
                 else
                 {

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

@@ -302,5 +302,36 @@ namespace YSAI.RabbitMQ
         {
             return Task.Run(() => Produce(Topic, Content));
         }
+
+        public OperateResult Subscribe(string Topic)
+        {
+            Depart("Subscribe");
+            OperateResult operateResult = Consume(Topic);
+            return Break("Subscribe", operateResult.State, operateResult.Message);
+        }
+
+        public Task<OperateResult> SubscribeAsync(string Topic)
+        {
+            return Task.Run(() => Subscribe(Topic));
+        }
+
+        public OperateResult UnSubscribe(string Topic)
+        {
+            Depart("UnSubscribe");
+            try
+            {
+                Channels[basics.ExChangeName].QueueDelete(Topic);
+                return Break("UnSubscribe", true);
+            }
+            catch (Exception ex)
+            {
+                return Break("UnSubscribe", false, ex.Message);
+            }
+        }
+
+        public Task<OperateResult> UnSubscribeAsync(string Topic)
+        {
+            return Task.Run(() => UnSubscribe(Topic));
+        }
     }
 }

src/YSAI.DAQ/YSAI.S7/client/S7ClientOperate.cs

@@ -339,7 +339,7 @@ namespace YSAI.S7.client
                 if (subscribeOperate != null)
                 {
                     OperateResult operateResult = subscribeOperate.UnSubscribe(address);
-                    return Break("Subscribe", operateResult.State, operateResult.Message);
+                    return Break("UnSubscribe", operateResult.State, operateResult.Message);
                 }
                 else
                 {

src/YSAI.DAQ/YSAI.Test/TestAll.cs

@@ -481,9 +481,9 @@ namespace YSAI.Test
                                 {
                                     new Basics.DllData.NamespaceData.ClassData()
                                     {
-                                        ClassName="KafkaProducerOperate",
-                                        ConstructorParam=new object[]{ new KafkaProducerData.Basics { BootstrapServers = "127.0.0.1:8083", ClientId = "ysai", SN = "YSAI" } },
-                                        SN="YSAI.Kafka.KafkaProducerOperate[Instance]",
+                                        ClassName="KafkaOperate",
+                                        ConstructorParam=new object[]{ new KafkaData.Basics { BootstrapServers = "127.0.0.1:8083", SN = "YSAI" } },
+                                        SN="YSAI.Kafka.KafkaOperate[Instance]",
                                         MethodDatas=new List<Basics.DllData.NamespaceData.ClassData.MethodData>()
                                         {
                                             new Basics.DllData.NamespaceData.ClassData.MethodData()