From b749cf52c359fce99cbc09bc34d3aeb28f9876e4 Mon Sep 17 00:00:00 2001 From: Diego <2248356998@qq.com> Date: Fri, 12 Jul 2024 09:33:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(RpcService):=20rpc=E5=A4=9A=E5=86=99?= =?UTF-8?q?=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Plugin/Collect/CollectBase.cs | 117 +++++++++++++++--- .../Services/Rpc/RpcService.cs | 73 ++++++----- .../MqttServer/MqttServer.cs | 4 + .../MqttServer/MqttServer.other.cs | 6 + 4 files changed, 148 insertions(+), 52 deletions(-) diff --git a/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs b/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs index 218521737..f6edfc418 100644 --- a/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs +++ b/src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs @@ -50,6 +50,8 @@ public abstract class CollectBase : DriverBase private IStringLocalizer Localizer { get; set; } + public virtual bool IsSingleThread => true; + internal override void Init(DeviceRunTime device) { Localizer = App.CreateLocalizerByType(typeof(CollectBase))!; @@ -89,8 +91,13 @@ public abstract class CollectBase : DriverBase { try { - //if (IsSingleThread) - // await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false); + if (IsSingleThread) + { + while (WriteLock.IsWaitting) + { + await Task.Delay(100).ConfigureAwait(false); + } + } if (cancellationToken.IsCancellationRequested) return new(new OperationCanceledException()); @@ -112,8 +119,6 @@ public abstract class CollectBase : DriverBase } finally { - //if (IsSingleThread) - // WriteLock.Release(); } } @@ -126,8 +131,8 @@ public abstract class CollectBase : DriverBase try { // 如果是单线程模式，则等待写入锁 - //if (IsSingleThread) - // await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false); + if (IsSingleThread) + await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false); // 检查协议是否为空，如果为空则抛出异常 if (Protocol == null) @@ -139,11 +144,18 @@ public abstract class CollectBase : DriverBase // 使用并发方式遍历写入信息列表，并进行异步写入操作 await writeInfoLists.ParallelForEachAsync(async (writeInfo, cancellationToken) => { - // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果 - var result = await Protocol.WriteAsync(writeInfo.Key.RegisterAddress, writeInfo.Value, writeInfo.Key.DataType, cancellationToken).ConfigureAwait(false); + try + { + // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果 + var result = await Protocol.WriteAsync(writeInfo.Key.RegisterAddress, writeInfo.Value, writeInfo.Key.DataType, cancellationToken).ConfigureAwait(false); - // 将操作结果添加到结果字典中，使用变量名称作为键 - operResults.TryAdd(writeInfo.Key.Name, result); + // 将操作结果添加到结果字典中，使用变量名称作为键 + operResults.TryAdd(writeInfo.Key.Name, result); + } + catch (Exception ex) + { + operResults.TryAdd(writeInfo.Key.Name, new(ex)); + } }, CollectProperties.ConcurrentCount, cancellationToken).ConfigureAwait(false); // 返回包含操作结果的字典 @@ -152,8 +164,8 @@ public abstract class CollectBase : DriverBase finally { // 如果是单线程模式，则释放写入锁 - //if (IsSingleThread) - // WriteLock.Release(); + if (IsSingleThread) + WriteLock.Release(); } } @@ -535,14 +547,10 @@ public abstract class CollectBase : DriverBase /// 指示是否为读取操作 /// 取消操作的通知 /// 操作结果，包含执行方法的结果 - internal async ValueTask> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default) + protected virtual async ValueTask> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default) { try { - // 如果配置为单线程模式，则获取写入锁定 - //if (IsSingleThread) - // await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false); - // 初始化操作结果 OperResult result = new OperResult(); @@ -590,9 +598,6 @@ public abstract class CollectBase : DriverBase } finally { - // 如果配置为单线程模式，则释放写入锁定 - //if (IsSingleThread) - // WriteLock.Release(); } } @@ -640,5 +645,77 @@ public abstract class CollectBase : DriverBase return results.Concat(results1).ToDictionary(a => a.Key, a => a.Value); } + /// + /// 异步写入方法 + /// + /// 要写入的变量及其对应的数据 + /// 取消操作的通知 + /// 写入操作的结果字典 + internal async ValueTask>> InvokeMethodAsync(Dictionary writeInfoLists, CancellationToken cancellationToken) + { + // 初始化结果字典 + Dictionary> results = new Dictionary>(); + + // 遍历写入信息列表 + foreach (var (deviceVariable, jToken) in writeInfoLists) + { + // 检查是否有写入表达式 + if (!string.IsNullOrEmpty(deviceVariable.WriteExpressions)) + { + // 提取原始数据 + object rawdata = jToken is JValue jValue ? jValue.Value : jToken is JArray jArray ? jArray : jToken.ToString(); + try + { + // 根据写入表达式转换数据 + object data = deviceVariable.WriteExpressions.GetExpressionsResult(rawdata); + // 将转换后的数据重新赋值给写入信息列表 + writeInfoLists[deviceVariable] = JToken.FromObject(data); + } + catch (Exception ex) + { + // 如果转换失败，则记录错误信息 + results.Add(deviceVariable.Name, new OperResult(Localizer["WriteExpressionsError", deviceVariable.Name, deviceVariable.WriteExpressions, ex.Message], ex)); + } + } + } + + ConcurrentDictionary> operResults = new(); + + try + { + // 如果是单线程模式，则等待写入锁 + if (IsSingleThread) + await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false); + + // 使用并发方式遍历写入信息列表，并进行异步写入操作 + await writeInfoLists + .Where(a => !results.Any(b => b.Key == a.Key.Name)) + .ToDictionary(item => item.Key, item => item.Value).ParallelForEachAsync(async (writeInfo, cancellationToken) => + { + try + { + // 调用协议的写入方法，将写入信息中的数据写入到对应的寄存器地址，并获取操作结果 + var result = await InvokeMethodAsync(writeInfo.Key.VariableMethod, writeInfo.Value?.ToString(), false, cancellationToken).ConfigureAwait(false); + + // 将操作结果添加到结果字典中，使用变量名称作为键 + operResults.TryAdd(writeInfo.Key.Name, result); + } + catch (Exception ex) + { + operResults.TryAdd(writeInfo.Key.Name, new(ex)); + } + }, CollectProperties.ConcurrentCount, cancellationToken).ConfigureAwait(false); + } + finally + { + // 如果是单线程模式，则释放写入锁 + if (IsSingleThread) + WriteLock.Release(); + } + + // 将转换失败的变量和写入成功的变量的操作结果合并到结果字典中 + return results.Concat(operResults).ToDictionary(a => a.Key, a => a.Value); + } + #endregion 写入方法 } diff --git a/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs b/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs index 07cabd0a7..cad16fe0a 100644 --- a/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs +++ b/src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs @@ -45,7 +45,7 @@ public class RpcService : IRpcService { // 初始化用于存储将要写入的变量和方法的字典 Dictionary> WriteVariables = new(); - Dictionary> WriteMethods = new(); + Dictionary> WriteMethods = new(); // 用于存储结果的并发字典 ConcurrentDictionary results = new(); @@ -107,15 +107,16 @@ public class RpcService : IRpcService } else { + JToken tagValue = JTokenUtil.GetJTokenFromString(item.Value); // 执行方法 if (WriteMethods.ContainsKey(dev)) { - WriteMethods[dev].Add(tag, item.Value); + WriteMethods[dev].Add(tag, tagValue); } else { WriteMethods.Add(dev, new()); - WriteMethods[dev].Add(tag, item.Value); + WriteMethods[dev].Add(tag, tagValue); } } } @@ -183,41 +184,49 @@ public class RpcService : IRpcService // 使用并行方式执行方法 await WriteMethods.ParallelForEachAsync(async (item, cancellationToken) => { - await item.Value.ParallelForEachAsync(async (writeMethod, cancellationToken) => + try { - // 执行变量附带的方法 - var method = item.Key.CurrentDevice.VariableMethods.FirstOrDefault(it => it.Variable == writeMethod.Key); - OperResult result; - try - { - result = await item.Key.InvokeMethodAsync(method, writeMethod.Value, false, cancellationToken); - } - catch (Exception ex) - { - result = new OperResult(ex); - } + // 调用设备的写入方法 + var result = await item.Key.InvokeMethodAsync(item.Value, cancellationToken); + + Dictionary operateMethods = item.Value.Select(a => a.Key).ToDictionary(a => a.Name, a => a.OtherMethod!); // 写入日志 - _logQueues.Enqueue( - new RpcLog() + foreach (var resultItem in result) + { + // 写入日志 + _logQueues.Enqueue( + new RpcLog() + { + LogTime = DateTime.Now, + OperateMessage = resultItem.Value.IsSuccess ? null : resultItem.Value.ToString(), + IsSuccess = resultItem.Value.IsSuccess, + OperateMethod = operateMethods[resultItem.Key], + OperateObject = resultItem.Key, + OperateSource = sourceDes, + ParamJson = items[resultItem.Key]?.ToString(), + ResultJson = resultItem.Value.Content?.ToString() + } + ); + + // 不返回详细错误 + if (!resultItem.Value.IsSuccess) { - LogTime = DateTime.Now, - OperateMessage = result.IsSuccess ? null : result.ToString(), - IsSuccess = result.IsSuccess, - OperateMethod = writeMethod.Key.OtherMethod!, - OperateObject = writeMethod.Key.Name, - OperateSource = sourceDes, - ParamJson = writeMethod.Value?.ToString(), - ResultJson = result.Content?.ToString() + OperResult result1 = resultItem.Value; + result1.Exception = null; + result[resultItem.Key] = result1; } - ); - - // 不返回详细错误 - result.Exception = null; - results.TryAdd(writeMethod.Key.Name, result); - }, item.Key.CollectProperties.ConcurrentCount, cancellationToken); + } + } + catch (Exception ex) + { + // 将异常信息添加到结果字典中 + results.AddRange(item.Value.Select((KeyValuePair a) => + { + return new KeyValuePair(a.Key.Name, new OperResult(ex)); + })); + } }, Environment.ProcessorCount / 2, cancellationToken); - // 返回结果字典 return new(results); } diff --git a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs index c355f7d0b..1e5627893 100644 --- a/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs +++ b/src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs @@ -14,6 +14,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using MQTTnet.AspNetCore; +using MQTTnet.Server; using ThingsGateway.Foundation; @@ -81,6 +82,8 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript