g82tt/ThingsGateway
1
0
Fork 0
mirror of https://gitee.com/ThingsGateway/ThingsGateway.git synced 2025-10-25 20:53:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化sql实时表执行性能

Browse Source
This commit is contained in:
Diego
2025-06-11 21:39:14 +08:00
parent ca2b17d433
commit 19188a62c4
5 changed files with 31 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/Directory.Build.props
View File

@@ -1,8 +1,8 @@
<Project>
<PropertyGroup>
<PluginVersion>10.7.47</PluginVersion>
<ProPluginVersion>10.7.47</ProPluginVersion>
<PluginVersion>10.7.48</PluginVersion>
<ProPluginVersion>10.7.48</ProPluginVersion>
<AuthenticationVersion>2.5.0</AuthenticationVersion>
<NET8Version>8.0.17</NET8Version>
<NET9Version>9.0.6</NET9Version>

54
src/Plugin/ThingsGateway.Plugin.DB/SqlDB/SqlDBProducer.cs
View File

@@ -164,12 +164,15 @@ public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<
.Map(dest => dest.Value, src => src.Value == null ? string.Empty : src.Value.ToString() ?? string.Empty)
.Map(dest => dest.CreateTime, (src) => DateTime.Now);
_exRealTimerTick = new(_driverPropertys.RealTableBusinessInterval);
await base.InitChannelAsync(channel, cancellationToken).ConfigureAwait(false);
}
public override Task AfterVariablesChangedAsync(CancellationToken cancellationToken)
{
RealTimeVariables.Clear();
_initRealData = false;
return base.AfterVariablesChangedAsync(cancellationToken);
}
protected override async Task ProtectedStartAsync(CancellationToken cancellationToken)
{
@@ -215,42 +218,25 @@ public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<
{
if (_driverPropertys.IsReadDB)
{
if (_exRealTimerTick.IsTickHappen())
try
{
try
var varList = RealTimeVariables.ToListWithDequeue();
if (varList.Count > 0)
{
var varList = IdVariableRuntimes.Select(a => a.Value);
if (_driverPropertys.GroupUpdate)
var result = await UpdateAsync(varList.Adapt<List<SQLRealValue>>(), cancellationToken).ConfigureAwait(false);
if (success != result.IsSuccess)
{
var groups = varList.GroupBy(a => a.BusinessGroup);
foreach (var item in groups)
{
var result = await UpdateAsync(item.Adapt<List<SQLRealValue>>(), cancellationToken).ConfigureAwait(false);
if (success != result.IsSuccess)
{
if (!result.IsSuccess)
LogMessage?.LogWarning(result.ToString());
success = result.IsSuccess;
}
}
}
else
{
var result = await UpdateAsync(varList.Adapt<List<SQLRealValue>>(), cancellationToken).ConfigureAwait(false);
if (success != result.IsSuccess)
{
if (!result.IsSuccess)
LogMessage?.LogWarning(result.ToString());
success = result.IsSuccess;
}
if (!result.IsSuccess)
LogMessage?.LogWarning(result.ToString());
success = result.IsSuccess;
}
}
catch (Exception ex)
{
if (success)
LogMessage?.LogWarning(ex);
success = false;
}
}
catch (Exception ex)
{
if (success)
LogMessage?.LogWarning(ex);
success = false;
}
}

7
src/Plugin/ThingsGateway.Plugin.DB/SqlDB/SqlDBProducerProperty.cs
View File

@@ -58,13 +58,6 @@ public class SqlDBProducerProperty : BusinessPropertyWithCacheInterval
[AutoGenerateColumn(ComponentType = typeof(Textarea), Rows = 1)]
public string BigTextConnectStr { get; set; } = "server=.;uid=sa;pwd=111111;database=test;";
/// <summary>
/// 实时表间隔上传时间
/// </summary>
[DynamicProperty]
public virtual string RealTableBusinessInterval { get; set; } = "3000";
/// <summary>
/// 实时表脚本
/// </summary>

18
src/Plugin/ThingsGateway.Plugin.DB/SqlDB/SqlDbProducer.other.cs
View File

@@ -10,11 +10,11 @@
using Mapster;
using System.Collections.Concurrent;
using System.Diagnostics;
using ThingsGateway.Extension.Generic;
using ThingsGateway.Foundation;
using ThingsGateway.NewLife;
using ThingsGateway.Plugin.DB;
using TouchSocket.Core;
@@ -27,7 +27,6 @@ namespace ThingsGateway.Plugin.SqlDB;
public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<SQLHistoryValue>
{
private TypeAdapterConfig _config;
private TimeTick _exRealTimerTick;
private volatile bool _initRealData;
protected override ValueTask<OperResult> UpdateVarModel(IEnumerable<CacheDBItem<SQLHistoryValue>> item, CancellationToken cancellationToken)
@@ -78,6 +77,7 @@ public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<
}
}
private ConcurrentDictionary<long,VariableBasicData> RealTimeVariables { get; } = new ConcurrentDictionary<long, VariableBasicData>();
private void UpdateVariable(VariableRuntime variableRuntime, VariableBasicData variable)
{
if (_driverPropertys.IsHistoryDB)
@@ -93,6 +93,11 @@ public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<
AddQueueVarModel(new CacheDBItem<SQLHistoryValue>(variableRuntime.Adapt<SQLHistoryValue>(_config)));
}
}
if (_driverPropertys.IsReadDB)
{
RealTimeVariables.AddOrUpdate(variable.Id, variable, (key, oldValue) => oldValue);
}
}
@@ -169,25 +174,18 @@ public partial class SqlDBProducer : BusinessBaseWithCacheIntervalVariableModel<
if (!_initRealData)
{
if (datas?.Count > 0)
{
Stopwatch stopwatch = new();
stopwatch.Start();
var ids = (await db.Queryable<SQLRealValue>().AS(_driverPropertys.ReadDBTableName).Select(a => a.Id).ToListAsync(cancellationToken).ConfigureAwait(false)).ToHashSet();
var InsertData = datas.Where(a => !ids.Contains(a.Id)).ToList();
var InsertData = IdVariableRuntimes.Where(a => !ids.Contains(a.Key)).Select(a=>a.Value).Adapt<List<SQLRealValue>>();
var result = await db.Fastest<SQLRealValue>().AS(_driverPropertys.ReadDBTableName).PageSize(100000).BulkCopyAsync(InsertData).ConfigureAwait(false);
//var result = await db.Storageable(datas).As(_driverPropertys.ReadDBTableName).PageSize(5000).ExecuteSqlBulkCopyAsync(cancellationToken).ConfigureAwait(false);
_initRealData = true;
stopwatch.Stop();
if (result > 0)
{
LogMessage?.Trace($"RealTable Insert Data Count：{result}，watchTime: {stopwatch.ElapsedMilliseconds} ms");
}
return OperResult.Success;
}
return OperResult.Success;
}
else
{
if (datas?.Count > 0)
{

2
src/Version.props
View File

@@ -1,6 +1,6 @@
<Project>
<PropertyGroup>
<Version>10.7.47</Version>
<Version>10.7.48</Version>
</PropertyGroup>
<ItemGroup>