1

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

@@ -8,7 +8,7 @@ using YSAI.Core.@interface.unify;
 namespace YSAI.Core.@interface.only
 {
     /// <summary>
-    /// 数据库采集接口
+    /// 数据库接口
     /// </summary>
     public interface IDB : IRead, IDisposable { }
 }

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

@@ -3,11 +3,12 @@ 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.Core.data;
 
 namespace YSAI.Core.@interface.only
 {
+
     /// <summary>
     /// MQTT客户端接口
     /// </summary>
@@ -74,7 +75,7 @@ namespace YSAI.Core.@interface.only
         /// 注意：QoS 级别越高，流程越复杂，系统资源消耗越大
         /// </param>
         /// <returns>统一结果</returns>
-        OperateResult AddSubscribe(string Topic, int QoSLevel=0);
+        OperateResult AddSubscribe(string Topic, int QoSLevel = 0);
 
         /// <summary>
         /// 移除订阅
@@ -90,4 +91,14 @@ namespace YSAI.Core.@interface.only
         /// <returns>统一结果</returns>
         OperateResult RemoveSubscribe(string Topic);
     }
+
+    /// <summary>
+    /// MQTT服务端接口
+    /// </summary>
+    public interface IMqttService : IInit, IOn, IOff, IDisposable { }
+
+    /// <summary>
+    /// MQTTwebsocket 接口
+    /// </summary>
+    public interface IMqttWebSocketService : IOn, IDisposable { }
 }

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

@@ -1,14 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.@interface.unify;
-
-namespace YSAI.Core.@interface.only
-{
-    /// <summary>
-    /// MQTT服务端接口
-    /// </summary>
-    public interface IMqttService : IInit, IOn, IOff, IDisposable { }
-}

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

@@ -1,14 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.@interface.unify;
-
-namespace YSAI.Core.@interface.only
-{
-    /// <summary>
-    /// MQTTwebsocket 接口
-    /// </summary>
-    public interface IMqttWebSocketService : IOn, IDisposable { }
-}

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

@@ -3,8 +3,9 @@ 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.Core.data;
+using System.Collections.Concurrent;
 
 namespace YSAI.Core.@interface.only
 {
@@ -91,4 +92,76 @@ namespace YSAI.Core.@interface.only
         /// <returns>统一出参</returns>
         Task<OperateResult> RemoveNodeAsync(string NodeName, string? GroupName = null);
     }
+
+    /// <summary>
+    /// OPCDAHTTP 接口
+    /// </summary>
+    public interface IOpcDaHttp : IRead, IDisposable
+    {
+        /// <summary>
+        /// 请求
+        /// </summary>
+        /// <param name="request">请求方法</param>
+        /// <param name="Param">参数</param>
+        /// <returns>统一出参</returns>
+        OperateResult Request(int requestApi, object? Param = null);
+
+        /// <summary>
+        /// 请求
+        /// </summary>
+        /// <param name="request">请求方法</param>
+        /// <param name="Param">参数</param>
+        /// <returns>统一出参</returns>
+        Task<OperateResult> RequestAsync(int requestApi, object? Param = null);
+    }
+
+    /// <summary>
+    /// OPCUA 客户端 接口
+    /// </summary>
+    public interface IOpcUaClient : IInit, IOn, IOff, IRead, IWrite, IDisposable
+    {
+        /// <summary>
+        /// 批量添加订阅
+        /// </summary>
+        /// <param name="Param">参数</param>
+        /// <param name="Priority">分配给订阅的优先级</param>
+        /// <returns>只要有一个不成功则直接返回</returns>
+        Task<OperateResult> AddSubscribeAsync(ConcurrentDictionary<string, List<AddressDetails>> param, byte Priority = 100);
+        /// <summary>
+        /// 批量添加订阅
+        /// </summary>
+        /// <param name="Param">参数</param>
+        /// <param name="Priority">分配给订阅的优先级</param>
+        /// <returns>只要有一个不成功则直接返回</returns>
+        OperateResult AddSubscribe(ConcurrentDictionary<string, List<AddressDetails>> param, byte Priority = 100);
+        /// <summary>
+        ///  移除订阅 通过节点来移除订阅，移除指定节点的订阅，没有也返回成功
+        /// </summary>
+        /// <returns>统一出参</returns>
+        Task<OperateResult> RemoveSubscribeAsync(List<AddressDetails> Nodes);
+        /// <summary>
+        ///  移除订阅 通过节点来移除订阅，移除指定节点的订阅，没有也返回成功
+        /// </summary>
+        /// <returns>统一出参</returns>
+        OperateResult RemoveSubscribe(List<AddressDetails> Nodes);
+
+        /// <summary>
+        /// 删除一个节点的操作，除非服务器配置允许，否则引发异常
+        /// </summary>
+        /// <param name="Key">关键字</param>
+        /// <returns>统一出参</returns>
+        Task<OperateResult> DeleteNodeAsync(string Key);
+
+        /// <summary>
+        /// 删除一个节点的操作，除非服务器配置允许，否则引发异常
+        /// </summary>
+        /// <param name="Key">关键字</param>
+        /// <returns>统一出参</returns>
+        OperateResult DeleteNode(string Key);
+    }
+
+    /// <summary>
+    /// OPCUA 服务端接口
+    /// </summary>
+    public interface IOpcUaService : IOn, IOff, IDisposable { }
 }

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

@@ -1,32 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.data;
-using YSAI.Core.@interface.unify;
-
-namespace YSAI.Core.@interface.only
-{
-    /// <summary>
-    /// OPCDAHTTP 接口
-    /// </summary>
-    public interface IOpcDaHttp : IRead, IDisposable
-    {
-        /// <summary>
-        /// 请求
-        /// </summary>
-        /// <param name="request">请求方法</param>
-        /// <param name="Param">参数</param>
-        /// <returns>统一出参</returns>
-        OperateResult Request(int requestApi, object? Param = null);
-
-        /// <summary>
-        /// 请求
-        /// </summary>
-        /// <param name="request">请求方法</param>
-        /// <param name="Param">参数</param>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> RequestAsync(int requestApi, object? Param = null);
-    }
-}

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

@@ -1,57 +0,0 @@
-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 static Microsoft.ClearScript.V8.V8CpuProfile;
-
-namespace YSAI.Core.@interface.only
-{
-    /// <summary>
-    /// OPCUA 客户端 接口
-    /// </summary>
-    public interface IOpcUaClient : IInit, IOn, IOff, IRead, IWrite, IDisposable
-    {
-        /// <summary>
-        /// 批量添加订阅
-        /// </summary>
-        /// <param name="Param">参数</param>
-        /// <param name="Priority">分配给订阅的优先级</param>
-        /// <returns>只要有一个不成功则直接返回</returns>
-        Task<OperateResult> AddSubscribeAsync(ConcurrentDictionary<string, List<AddressDetails>> param, byte Priority = 100);
-        /// <summary>
-        /// 批量添加订阅
-        /// </summary>
-        /// <param name="Param">参数</param>
-        /// <param name="Priority">分配给订阅的优先级</param>
-        /// <returns>只要有一个不成功则直接返回</returns>
-        OperateResult AddSubscribe(ConcurrentDictionary<string, List<AddressDetails>> param, byte Priority = 100);
-        /// <summary>
-        ///  移除订阅 通过节点来移除订阅，移除指定节点的订阅，没有也返回成功
-        /// </summary>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> RemoveSubscribeAsync(List<AddressDetails> Nodes);
-        /// <summary>
-        ///  移除订阅 通过节点来移除订阅，移除指定节点的订阅，没有也返回成功
-        /// </summary>
-        /// <returns>统一出参</returns>
-        OperateResult RemoveSubscribe(List<AddressDetails> Nodes);
-
-        /// <summary>
-        /// 删除一个节点的操作，除非服务器配置允许，否则引发异常
-        /// </summary>
-        /// <param name="Key">关键字</param>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> DeleteNodeAsync(string Key);
-
-        /// <summary>
-        /// 删除一个节点的操作，除非服务器配置允许，否则引发异常
-        /// </summary>
-        /// <param name="Key">关键字</param>
-        /// <returns>统一出参</returns>
-        OperateResult DeleteNode(string Key);
-    }
-}

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

@@ -1,15 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.data;
-using YSAI.Core.@interface.unify;
-
-namespace YSAI.Core.@interface.only
-{
-    /// <summary>
-    /// OPCUA 服务端接口
-    /// </summary>
-    public interface IOpcUaService : IOn, IOff, IDisposable { }
-}

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

@@ -0,0 +1,94 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using YSAI.Core.data;
+using YSAI.Core.@interface.unify;
+
+namespace YSAI.Core.@interface.only
+{
+    /// <summary>
+    /// 发布者接口
+    /// </summary>
+    public interface IRabbitMQPublisher : IOn, IOff, IDisposable
+    {
+        /// <summary>
+        /// 发布 一个交换机可以发布多个消息
+        /// </summary>
+        /// <param name="MessageHead">消息头</param>
+        /// <param name="MessageContent">消息内容</param>
+        /// <param name="Type">
+        /// 通道类型
+        /// direct  //用于AMQP直接交换的交换类型。
+        /// fanout  //用于AMQP扇出交换的交换类型。
+        /// headers //用于AMQP报头交换的交换类型。
+        /// topic   //用于AMQP主题交换的交换类型。
+        /// </param>
+        /// <param name="Durable">持久化</param>
+        /// <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);
+
+        /// <summary>
+        /// 发布 一个交换机可以发布多个消息
+        /// </summary>
+        /// <param name="MessageHead">消息头</param>
+        /// <param name="MessageContent">消息内容</param>
+        /// <param name="Type">
+        /// 通道类型
+        /// direct  //用于AMQP直接交换的交换类型。
+        /// fanout  //用于AMQP扇出交换的交换类型。
+        /// headers //用于AMQP报头交换的交换类型。
+        /// topic   //用于AMQP主题交换的交换类型。
+        /// </param>
+        /// <param name="Durable">持久化</param>
+        /// <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);
+    }
+
+    /// <summary>
+    /// 消费者接口
+    /// </summary>
+    public interface IRabbitMQConsumer : IOn, IOff, IDisposable
+    {
+        /// <summary>
+        /// 消费 一个交换机可以消费多个消息
+        /// </summary>
+        /// <param name="MessageHead">消息头</param>
+        /// <param name="Type">
+        /// 通道类型
+        /// direct  //用于AMQP直接交换的交换类型。
+        /// fanout  //用于AMQP扇出交换的交换类型。
+        /// headers //用于AMQP报头交换的交换类型。
+        /// topic   //用于AMQP主题交换的交换类型。
+        /// </param>
+        /// <param name="AutoAck">自动确认</param>
+        /// <param name="Durable">持久化</param>
+        /// <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);
+
+        /// <summary>
+        /// 消费 一个交换机可以消费多个消息
+        /// </summary>
+        /// <param name="MessageHead">消息头</param>
+        /// <param name="Type">
+        /// 通道类型
+        /// direct  //用于AMQP直接交换的交换类型。
+        /// fanout  //用于AMQP扇出交换的交换类型。
+        /// headers //用于AMQP报头交换的交换类型。
+        /// topic   //用于AMQP主题交换的交换类型。
+        /// </param>
+        /// <param name="AutoAck">自动确认</param>
+        /// <param name="Durable">持久化</param>
+        /// <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);
+    }
+}

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

@@ -8,6 +8,9 @@ using YSAI.Core.script;
 
 namespace YSAI.Core.@interface.only
 {
+    /// <summary>
+    /// 脚本接口
+    /// </summary>
     public interface IScript : IDisposable
     {
         /// <summary>

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

@@ -3,8 +3,9 @@ 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.Core.data;
+using System.Net;
 
 namespace YSAI.Core.@interface.only
 {
@@ -43,4 +44,35 @@ namespace YSAI.Core.@interface.only
         /// <returns></returns>
         Task<OperateResult> SendAsync(byte[] Data, string? Address = null);
     }
+    /// <summary>
+    /// udp接口
+    /// </summary>
+    public interface ISocketUdp : IOn, IOff, IDisposable
+    {
+        /// <summary>
+        /// 读取或发送
+        /// </summary>
+        /// <param name="Data">字节数据</param>
+        /// <returns>统一出参</returns>
+        Task<OperateResult> ReadOrWriteAsync(byte[] Data, IPEndPoint iPEndPoint);
+        /// <summary>
+        /// 读取或发送
+        /// </summary>
+        /// <param name="Data">字节数据</param>
+        /// <returns>统一出参</returns>
+        OperateResult ReadOrWrite(byte[] Data, IPEndPoint iPEndPoint);
+
+        /// <summary>
+        /// 读取或发送
+        /// </summary>
+        /// <param name="Data">字节数据</param>
+        /// <returns>统一出参</returns>
+        Task<OperateResult> SendAsync(byte[] Data, IPEndPoint iPEndPoint);
+        /// <summary>
+        /// 读取或发送
+        /// </summary>
+        /// <param name="Data">字节数据</param>
+        /// <returns>统一出参</returns>
+        OperateResult Send(byte[] Data, IPEndPoint iPEndPoint);
+    }
 }

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

@@ -1,40 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Net;
-using System.Text;
-using System.Threading.Tasks;
-using YSAI.Core.data;
-using YSAI.Core.@interface.unify;
-
-namespace YSAI.Core.@interface.only
-{
-    public interface ISocketUdp : IOn, IOff, IDisposable
-    {
-        /// <summary>
-        /// 读取或发送
-        /// </summary>
-        /// <param name="Data">字节数据</param>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> ReadOrWriteAsync(byte[] Data, IPEndPoint iPEndPoint);
-        /// <summary>
-        /// 读取或发送
-        /// </summary>
-        /// <param name="Data">字节数据</param>
-        /// <returns>统一出参</returns>
-        OperateResult ReadOrWrite(byte[] Data, IPEndPoint iPEndPoint);
-
-        /// <summary>
-        /// 读取或发送
-        /// </summary>
-        /// <param name="Data">字节数据</param>
-        /// <returns>统一出参</returns>
-        Task<OperateResult> SendAsync(byte[] Data, IPEndPoint iPEndPoint);
-        /// <summary>
-        /// 读取或发送
-        /// </summary>
-        /// <param name="Data">字节数据</param>
-        /// <returns>统一出参</returns>
-        OperateResult Send(byte[] Data, IPEndPoint iPEndPoint);
-    }
-}

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

@@ -8,6 +8,9 @@ using YSAI.Core.@interface.unify;
 
 namespace YSAI.Core.@interface.only
 {
+    /// <summary>
+    /// 自定义订阅接口
+    /// </summary>
     public interface ISubscribe : IOn, IOff, IDisposable
     {
         /// <summary>

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

@@ -30,7 +30,7 @@ namespace YSAI.Core.@interface.unify
         /// <summary>
         /// 信息传递
         /// </summary>
-        public static EventHandler<E1>? OnEventHandler;
+        public EventHandler<E1>? OnEventHandler;
     }
 
     /// <summary>
@@ -49,7 +49,7 @@ namespace YSAI.Core.@interface.unify
         /// <summary>
         /// 信息传递
         /// </summary>
-        public static EventHandler<E1>? OnEventHandler;
+        public EventHandler<E1>? OnEventHandler;
 
         /// <summary>
         /// 信息传递

src/YSAI.DAQ/YSAI.Mqtt/service/websocket/MqttWebSocketServiceOperate.cs

@@ -18,7 +18,7 @@ namespace YSAI.Mqtt.service.websocket
     /// <summary>
     /// MQTT WEBSOCKET 服务
     /// </summary>
-    public class MqttWebSocketServiceOperate: IBase<Event>, IMqttWebSocketService
+    public class MqttWebSocketServiceOperate: IBase, IMqttWebSocketService
     {
         private static readonly object Lock = new object();  //锁
         private static List<MqttWebSocketServiceOperate> ThisObjList = new List<MqttWebSocketServiceOperate>(); //自身对象集合
@@ -96,6 +96,20 @@ namespace YSAI.Mqtt.service.websocket
 
         public override string ClassName => "MqttWebSocketServiceOperate";
 
+        /// <summary>
+        /// 信息传递
+        /// </summary>
+        public event EventHandler<Event> OnEvent
+        {
+            add { OnEventHandler += value; }
+            remove { OnEventHandler -= value; }
+        }
+        /// <summary>
+        /// 信息传递
+        /// </summary>
+        public static EventHandler<Event>? OnEventHandler;
+
+
         public Task<OperateResult> OnAsync()
         {
             return Task.Run(()=> On());

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

@@ -0,0 +1,235 @@
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+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 class RabbitMQConsumerOperate : IBase<RabbitMQData.Event>, IRabbitMQConsumer
+    {
+        public override string LogHead => "[ RabbitMQConsumerOperate 操作 ]";
+        public override string ClassName => "RabbitMQConsumerOperate";
+
+        private static readonly object Lock = new object();  //锁
+        private static List<RabbitMQConsumerOperate> ThisObjList = new List<RabbitMQConsumerOperate>(); //自身对象集合
+
+        /// <summary>
+        /// 单例模式
+        /// </summary>
+        /// <returns></returns>
+        public static RabbitMQConsumerOperate Instance(RabbitMQData.Basics basics)
+        {
+            RabbitMQConsumerOperate? 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
+                    {
+                        RabbitMQConsumerOperate exp2 = new RabbitMQConsumerOperate(basics);
+                        ThisObjList.Add(exp2);
+                        return exp2;
+                    }
+                }
+            }
+            return exp;
+        }
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        private RabbitMQData.Basics basics { get; set; }
+
+        /// <summary>
+        /// 构造函数
+        /// </summary>
+        public RabbitMQConsumerOperate(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, 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);
+            }
+        }
+
+        public Task<OperateResult> OnAsync()
+        {
+            return Task.Run(() => On());
+        }
+
+        public OperateResult On()
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.On").StartRecord();
+            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()
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.Off").StartRecord();
+            try
+            {
+                if (connection == null || !connection.IsOpen)
+                {
+                    return Break("Off", false, "RabbitMQ连接未打开");
+                }
+                else
+                {
+                    //关闭
+                    connection.Close();
+                    //释放
+                    connection.Dispose();
+                    //清空通道集合
+                    Channels.Clear();
+                    //队列清空
+                    Queues.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 Consume(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.Consume").StartRecord();
+            try
+            {
+                if (connection == null || !connection.IsOpen)
+                {
+                    return Break("Consume", 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 (!Queues.ContainsKey(MessageHead))
+                    {
+                        //创建队列
+                        Channels[basics.ExChangeName].QueueDeclare(MessageHead, Durable, Exclusive, AutoDelete);
+                        //绑定队列
+                        Channels[basics.ExChangeName].QueueBind(MessageHead, basics.ExChangeName, MessageHead);
+
+                        EventingBasicConsumer consumer = new(Channels[basics.ExChangeName]);
+                        consumer.Received += delegate (object? sender, BasicDeliverEventArgs e)
+                        {
+                            Consumer_Received(sender, e, AutoAck);
+                        };
+                        // 开启消费者与通道、队列关联
+                        Channels[basics.ExChangeName].BasicConsume(MessageHead, AutoAck, consumer);
+                        //返回
+                        return Break("Consume", true);
+                    }
+                    if (OnEventHandler == null)
+                    {
+                        return Break("Consume", false, "已存在此消费，请注册事件返回参数");
+                    }
+                    else
+                    {
+                        return Break("Consume", false, "已存在此消费");
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                return Break("Consume", false, "RabbitMQ消费异常：" + ex.Message);
+            }
+        }
+
+        public Task<OperateResult> ConsumeAsync(string MessageHead, string Type = "topic", bool AutoAck = false, bool Durable = true, bool Exclusive = false, bool AutoDelete = false)
+        {
+            return Task.Run(() => Consume(MessageHead, Type, AutoAck, Durable, Exclusive, AutoDelete));
+        }
+    }
+}

src/YSAI.DAQ/YSAI.RabbitMQ/RabbitMQData.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace YSAI.RabbitMQ
+{
+    /// <summary>
+    /// 数据
+    /// </summary>
+    public class RabbitMQData
+    {
+        /// <summary>
+        /// 基础数据
+        /// </summary>
+        public class Basics
+        {
+            /// <summary>
+            /// 交换机名称
+            /// </summary>
+            public string? ExChangeName { get; set; }
+            /// <summary>
+            /// 要连接的地址
+            /// </summary>
+            public string? HostName { get; set; }
+            /// <summary>
+            /// 端口
+            /// </summary>
+            public int Port { get; set; }
+            /// <summary>
+            /// 账号
+            /// </summary>
+            public string? UserName { get; set; } = null;
+            /// <summary>
+            /// 密码
+            /// </summary>
+            public string? Password { get; set; } = null;
+
+            /// <summary>
+            /// 重写Equals
+            /// </summary>
+            /// <param name="obj"></param>
+            /// <returns></returns>
+            public override bool Equals(object Obj)
+            {
+                Basics? obj = Obj as Basics;
+                if (obj == null) return false;
+                if (obj.ExChangeName == this.ExChangeName &&
+                    obj.HostName == this.HostName&&
+                    obj.Port == this.Port &&
+                    obj.UserName == this.UserName &&
+                    obj.Password == this.Password)
+                {
+                    return true;
+                }
+                else
+                {
+                    return false;
+                }
+            }
+        }
+
+
+        public class Event
+        {
+            /// <summary>
+            /// 消息头
+            /// </summary>
+            public string MessageHead { get; set; }
+            /// <summary>
+            /// 消息的内容
+            /// </summary>
+            public string MessageContent { get; set; }
+
+            /// <summary>
+            /// 时间
+            /// </summary>
+            public DateTime Time { get; set; } = DateTime.Now;
+        }
+    }
+}

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

@@ -0,0 +1,215 @@
+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 class RabbitMQPublisherOperate : IBase, IRabbitMQPublisher
+    {
+        public override string LogHead => "[ RabbitMQPublisherOperate 操作 ]";
+        public 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()
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.On").StartRecord();
+            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()
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.Off").StartRecord();
+            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)
+        {
+            //开始记录运行时间
+            RunTimeTool.Instance($"{ClassName}.Publish").StartRecord();
+            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.RabbitMQ/YSAI.RabbitMQ.csproj

@@ -6,4 +6,12 @@
     <Nullable>enable</Nullable>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="RabbitMQ.Client" Version="6.5.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\YSAI.Core\YSAI.Core.csproj" />
+  </ItemGroup>
+
 </Project>

src/YSAI.DAQ/YSAI.S7/S7Data.cs

@@ -15,14 +15,10 @@ namespace YSAI.S7
         /// </summary>
         public class Basics
         {
-            /// <summary>
-            /// 唯一标识符
-            /// </summary>
-            public string SN { get; set; }
             /// <summary>
             /// ip地址
             /// </summary>
-            public string Ip { get; set; }
+            public string? Ip { get; set; }
             /// <summary>
             /// 端口
             /// </summary>
@@ -36,13 +32,13 @@ namespace YSAI.S7
             /// 通常为0，
             /// 但检查Step7或TIA门户的硬件配置
             /// </summary>
-            public int Rack { get; set; }
+            public int Rack { get; set; } = 0;
             /// <summary>
             /// PLC的CPU插槽，
             /// S7-300/S7-400为2，
             /// S7-1200/S7-1500为0
             /// </summary>
-            public int Slot { get; set; }
+            public int Slot { get; set; } = 0;
             /// <summary>
             /// 重写Equals
             /// </summary>
@@ -52,7 +48,11 @@ namespace YSAI.S7
             {
                 Basics? obj = Obj as Basics;
                 if (obj == null) return false;
-                if (obj.SN == this.SN)
+                if (obj.Ip == this.Ip&&
+                    obj.Port == this.Port &&
+                    obj.S7CpuType == this.S7CpuType &&
+                    obj.Rack == this.Rack &&
+                    obj.Slot == this.Slot)
                 {
                     return true;
                 }