GithubStar/ThingsGateway
1
0
Fork 0
mirror of https://github.com/ThingsGateway/ThingsGateway.git synced 2026-10-26 10:35:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(RpcService): rpc多写性能优化

Browse Source
This commit is contained in:
Diego
2024-07-12 09:33:17 +08:00
parent b9a36a9223
commit b749cf52c3
4 changed files with 148 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ThingsGateway.Gateway.Application/Plugin/Collect/CollectBase.cs
+97 -20
View File
@@ -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
/// <param name="isRead">指示是否为读取操作</param>
/// <param name="cancellationToken">取消操作的通知</param>
/// <returns>操作结果，包含执行方法的结果</returns>
internal async ValueTask<OperResult<object>> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default)
protected virtual async ValueTask<OperResult<object>> InvokeMethodAsync(VariableMethod variableMethod, string? value = null, bool isRead = true, CancellationToken cancellationToken = default)
{
try
{
// 如果配置为单线程模式，则获取写入锁定
//if (IsSingleThread)
// await WriteLock.WaitAsync(cancellationToken).ConfigureAwait(false);
// 初始化操作结果
OperResult<object> result = new OperResult<object>();
@@ -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);
}
/// <summary>
/// 异步写入方法
/// </summary>
/// <param name="writeInfoLists">要写入的变量及其对应的数据</param>
/// <param name="cancellationToken">取消操作的通知</param>
/// <returns>写入操作的结果字典</returns>
internal async ValueTask<Dictionary<string, OperResult<object>>> InvokeMethodAsync(Dictionary<VariableRunTime, JToken> writeInfoLists, CancellationToken cancellationToken)
{
// 初始化结果字典
Dictionary<string, OperResult<object>> results = new Dictionary<string, OperResult<object>>();
// 遍历写入信息列表
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<object>(Localizer["WriteExpressionsError", deviceVariable.Name, deviceVariable.WriteExpressions, ex.Message], ex));
}
}
}
ConcurrentDictionary<string, OperResult<object>> 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 写入方法
}
src/ThingsGateway.Gateway.Application/Services/Rpc/RpcService.cs
+41 -32
View File
@@ -45,7 +45,7 @@ public class RpcService : IRpcService
{
// 初始化用于存储将要写入的变量和方法的字典
Dictionary<CollectBase, Dictionary<VariableRunTime, JToken>> WriteVariables = new();
Dictionary<CollectBase, Dictionary<VariableRunTime, string>> WriteMethods = new();
Dictionary<CollectBase, Dictionary<VariableRunTime, JToken>> WriteMethods = new();
// 用于存储结果的并发字典
ConcurrentDictionary<string, OperResult> 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<object> result;
try
{
result = await item.Key.InvokeMethodAsync(method, writeMethod.Value, false, cancellationToken);
}
catch (Exception ex)
{
result = new OperResult<object>(ex);
}
// 调用设备的写入方法
var result = await item.Key.InvokeMethodAsync(item.Value, cancellationToken);
Dictionary<string, string> 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<object> 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<VariableRunTime, JToken> a) =>
{
return new KeyValuePair<string, OperResult>(a.Key.Name, new OperResult(ex));
}));
}
}, Environment.ProcessorCount / 2, cancellationToken);
// 返回结果字典
return new(results);
}
src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.cs
+4
View File
@@ -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<VariableDa
});
if (_mqttServer != null)
{
_mqttServer.ClientDisconnectedAsync -= MqttServer_ClientDisconnectedAsync;
_mqttServer.ClientDisconnectedAsync += MqttServer_ClientDisconnectedAsync;
_mqttServer.ValidatingConnectionAsync -= MqttServer_ValidatingConnectionAsync;
_mqttServer.InterceptingPublishAsync -= MqttServer_InterceptingPublishAsync;
_mqttServer.LoadingRetainedMessageAsync -= MqttServer_LoadingRetainedMessageAsync;
@@ -104,6 +107,7 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript<VariableDa
{
if (_mqttServer != null)
{
_mqttServer.ClientDisconnectedAsync -= MqttServer_ClientDisconnectedAsync;
_mqttServer.ValidatingConnectionAsync -= MqttServer_ValidatingConnectionAsync;
_mqttServer.InterceptingPublishAsync -= MqttServer_InterceptingPublishAsync;
_mqttServer.LoadingRetainedMessageAsync -= MqttServer_LoadingRetainedMessageAsync;
src/adapter/ThingsGateway.Plugin.Mqtt/MqttServer/MqttServer.other.cs
+6
View File
@@ -183,6 +183,12 @@ public partial class MqttServer : BusinessBaseWithCacheIntervalScript<VariableDa
await CompletedTask.Instance;
}
private Task MqttServer_ClientDisconnectedAsync(MQTTnet.Server.ClientDisconnectedEventArgs arg)
{
LogMessage?.LogInformation($"{ToString()}-{arg.ClientId}-Client DisConnected-{arg.DisconnectType}");
return Task.CompletedTask;
}
private async Task MqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
{
if (!arg.ClientId.StartsWith(_driverPropertys.StartWithId))