admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

解决IoTDB写入速度过慢的问题

Browse Source
This commit is contained in:
ChenYi 2025-04-28 16:37:31 +08:00
parent 021153a319
commit d18f60f9a7
14 changed files with 127 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/JiShe.CollectBus.IoTDB/JiShe.CollectBus.IoTDB.csproj
View File

@ -11,5 +11,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.IncrementalGenerator\JiShe.CollectBus.IncrementalGenerator.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
</ItemGroup>
</Project>

5
modules/JiShe.CollectBus.IoTDB/Model/IoTEntity.cs
View File

@ -1,9 +1,10 @@
using JiShe.CollectBus.IoTDB.Attribute;
using JiShe.CollectBus.Common.Attributes;
using JiShe.CollectBus.IoTDB.Attribute;
namespace JiShe.CollectBus.IoTDB.Model
{
/// <summary>
/// IoT实体基类，此类适用于多个数据测点记录场景，单个测点请使用子类 SingleMeasuring
/// IoT实体基类，此类适用于多个数据测点记录场景，单个测点请使用子类 SingleMeasuring，新增字段只能现有字段末尾添加，否则会导致数据写入失败。
/// </summary>
public abstract class IoTEntity
{

2
modules/JiShe.CollectBus.IoTDB/Options/IoTDBOptions.cs
View File

@ -26,7 +26,7 @@
/// <summary>
/// 连接池大小
/// </summary>
public int PoolSize { get; set; } = 2;
public int PoolSize { get; set; } = 8;
/// <summary>
/// 查询时，每次查询的数据量，默认1024

2
modules/JiShe.CollectBus.IoTDB/Provider/SessionPoolAdapter.cs
View File

@ -70,7 +70,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
var result = await _sessionPool.InsertAlignedTabletAsync(tablet);
if (result != 0)
{
throw new Exception($"{nameof(SessionPoolAdapter)} Tree模型数据入库没有成功，返回结果为：{result}");
throw new Exception($"{nameof(SessionPoolAdapter)} Tree模型数据入库没有成功，返回结果为：{result}，请检查IoTEntity继承子类属性索引是否有变动。");
}
//await CloseAsync();
return result;

2
modules/JiShe.CollectBus.IoTDB/Provider/TableSessionPoolAdapter.cs
View File

@ -68,7 +68,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
var result = await _sessionPool.InsertAsync(tablet);
if (result != 0)
{
throw new Exception($"{nameof(TableSessionPoolAdapter)} table模型数据入库没有成功，返回结果为：{result}");
throw new Exception($"{nameof(TableSessionPoolAdapter)} table模型数据入库没有成功，返回结果为：{result}，请检查IoTEntity继承子类属性索引是否有变动。");
}
//await CloseAsync();

15
services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
View File

@ -66,12 +66,17 @@ public class CollectBusApplicationModule : AbpModule
.ToList();
foreach (var type in types) await context.AddBackgroundWorkerAsync(type);
Task.Run(() =>
{
//Task.Run(() =>
//{
// //默认初始化表计信息
// var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
// dbContext.InitAmmeterCacheData();
// //await dbContext.InitWatermeterCacheData();
//}).ConfigureAwait(false);
//默认初始化表计信息
var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
dbContext.InitAmmeterCacheData();
//await dbContext.InitWatermeterCacheData();
}).ConfigureAwait(false);
_= dbContext.InitAmmeterCacheData();
}
}

1
services/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
View File

@ -31,7 +31,6 @@
<ProjectReference Include="..\..\modules\JiShe.CollectBus.Kafka\JiShe.CollectBus.Kafka.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.IncrementalGenerator\JiShe.CollectBus.IncrementalGenerator.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
</ItemGroup>
</Project>

77
services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -286,52 +286,51 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public virtual async Task InitAmmeterCacheData(string gatherCode = "")
{
//此处代码不要删除
//#if DEBUG
// var timeDensity = "15";
// var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
// var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
// var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
#if DEBUG
var timeDensity = "15";
var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, "JiSheCollectBus2", MeterTypeEnum.Ammeter, timeDensity)}";
// List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
// List<string> focusAddressDataLista = new List<string>();
// var timer1 = Stopwatch.StartNew();
List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
List<string> focusAddressDataLista = new List<string>();
var timer1 = Stopwatch.StartNew();
// var allIds = new HashSet<string>();
// decimal? score = null;
// string member = null;
var allIds = new HashSet<string>();
decimal? score = null;
string member = null;
// while (true)
// {
// var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
// redisCacheMeterInfoHashKeyTemp,
// redisCacheMeterInfoZSetScoresIndexKeyTemp,
// pageSize: 1000,
// lastScore: score,
// lastMember: member);
while (true)
{
var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
redisCacheMeterInfoHashKeyTemp,
redisCacheMeterInfoZSetScoresIndexKeyTemp,
pageSize: 1000,
lastScore: score,
lastMember: member);
// meterInfos.AddRange(page.Items);
// focusAddressDataLista.AddRange(page.Items.Select(d => d.FocusAddress));
// foreach (var item in page.Items)
// {
// if (!allIds.Add(item.MemberId))
// {
// _logger.LogError($"{item.MemberId}Duplicate data found!");
// }
// }
// if (!page.HasNext) break;
// score = page.NextScore;
// member = page.NextMember;
// }
meterInfos.AddRange(page.Items);
focusAddressDataLista.AddRange(page.Items.Select(d => d.FocusAddress));
foreach (var item in page.Items)
{
if (!allIds.Add(item.MemberId))
{
_logger.LogError($"{item.MemberId}Duplicate data found!");
}
}
if (!page.HasNext) break;
score = page.NextScore;
member = page.NextMember;
}
// timer1.Stop();
// _logger.LogError($"读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒");
// DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
// return;
//#else
// var meterInfos = await GetAmmeterInfoList(gatherCode);
//#endif
timer1.Stop();
_logger.LogError($"读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒");
DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
return;
#else
var meterInfos = await GetAmmeterInfoList(gatherCode);
#endif
if (meterInfos == null || meterInfos.Count <= 0)
{
throw new NullReferenceException($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，电表数据为空");

4
services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -35,6 +35,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//[Route($"/energy/app/scheduled")]
public class EnergySystemScheduledMeterReadingService : BasicScheduledMeterReadingService
{
string systemType = string.Empty;
string serverTagName = string.Empty;
private readonly ILogger<BasicScheduledMeterReadingService> _logger;
private readonly IIoTDbProvider _dbProvider;
@ -59,12 +60,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading
applicationOptions)
{
serverTagName = applicationOptions.Value.ServerTagName;
systemType = applicationOptions.Value.SystemType;
_dbProvider = dbProvider;
_logger = logger;
_protocolService = protocolService;
}
public sealed override string SystemType => SystemTypeConst.Energy;
public sealed override string SystemType => systemType;
public sealed override string ServerTagName => serverTagName;

5
services/JiShe.CollectBus.Domain/IotSystems/MeterReadingRecords/MeterReadingTelemetryPacketInfo.cs
View File

@ -1,4 +1,5 @@
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common.Attributes;
using JiShe.CollectBus.IoTDB.Attribute;
using JiShe.CollectBus.IoTDB.Enums;
using JiShe.CollectBus.IoTDB.Model;
@ -89,6 +90,7 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
/// <summary>
/// 抄读计量点
/// </summary>
[FIELDColumn]
public int Pn { get; set; }
/// <summary>
@ -107,6 +109,7 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
/// <summary>
/// 帧序列域 SEQ
/// </summary>
[FIELDColumn]
public int Seq { get; set; }
/// <summary>
@ -125,7 +128,7 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
/// 发送次数
/// </summary>
[FIELDColumn]
public int SendNum { get; set; }
public int? SendNum { get; set; }
/// <summary>
/// 下次发送时间

4
services/JiShe.CollectBus.Domain/JiShe.CollectBus.Domain.csproj
View File

@ -23,8 +23,10 @@
<ItemGroup>
<ProjectReference Include="..\..\modules\JiShe.CollectBus.FreeSql\JiShe.CollectBus.FreeSql.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.IoTDB\JiShe.CollectBus.IoTDB.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.Domain.Shared\JiShe.CollectBus.Domain.Shared.csproj" />
<ProjectReference Include="..\..\shared\JiShe.CollectBus.IncrementalGenerator\JiShe.CollectBus.IncrementalGenerator.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
</ItemGroup>
<ItemGroup>

58
shared/JiShe.CollectBus.IncrementalGenerator/ComplexTypeSourceGenerator.cs
View File

@ -20,13 +20,14 @@ namespace JiShe.CollectBus.IncrementalGenerator
public void Initialize(IncrementalGeneratorInitializationContext context)
{
Debugger.Launch();
//Debugger.Launch();
context.RegisterPostInitializationOutput(ctx =>
{
ctx.AddSource("DebugInit.g.cs", "// Generator initialized");
});
// 步骤1：筛选带有 [GenerateAccessors] 的类
var classDeclarations = context.SyntaxProvider
.CreateSyntaxProvider(
@ -43,28 +44,6 @@ namespace JiShe.CollectBus.IncrementalGenerator
private static bool IsClassWithAttribute(SyntaxNode node) => node is ClassDeclarationSyntax cds && cds.AttributeLists.Count > 0;
//private static ClassDeclarationSyntax? GetClassDeclaration(GeneratorSyntaxContext context)
//{
// var classDecl = (ClassDeclarationSyntax)context.Node;
// foreach (var attributeList in classDecl.AttributeLists)
// {
// foreach (var attribute in attributeList.Attributes)
// {
// var symbol = context.SemanticModel.GetSymbolInfo(attribute).Symbol;
// if (symbol is IMethodSymbol ctor)
// {
// string sdd = ctor.ContainingType.Name;
// var attributeType = context.SemanticModel.Compilation.GetTypeByMetadataName(AttributeFullName);
// if (ctor.ContainingType?.OriginalDefinition?.Equals(attributeType, SymbolEqualityComparer.Default) == true)
// {
// return classDecl;
// }
// }
// }
// }
// return null;
//}
private static ClassDeclarationSyntax? GetClassDeclaration(GeneratorSyntaxContext context)
{
@ -75,7 +54,7 @@ namespace JiShe.CollectBus.IncrementalGenerator
{
var symbol = context.SemanticModel.GetSymbolInfo(attribute).Symbol;
if (symbol is IMethodSymbol ctor &&
ctor.ContainingType?.OriginalDefinition?.Equals(attributeType, SymbolEqualityComparer.Default) == true)
SymbolEqualityComparer.Default.Equals(ctor.ContainingType, attributeType))
{
return classDecl;
}
@ -90,13 +69,27 @@ namespace JiShe.CollectBus.IncrementalGenerator
{
var processedTypes = new HashSet<ITypeSymbol>(SymbolEqualityComparer.Default);
if (!classes.Any())
{
context.ReportDiagnostic(Diagnostic.Create(
new DiagnosticDescriptor("GEN002", "No Targets",
"No classes with [GenerateAccessors] found", "Debug", DiagnosticSeverity.Warning, true),
Location.None));
}
foreach (var classDecl in classes.Distinct())
{
var model = compilation.GetSemanticModel(classDecl.SyntaxTree);
var classSymbol = model.GetDeclaredSymbol(classDecl) as INamedTypeSymbol;
if (classSymbol == null || !processedTypes.Add(classSymbol)) continue;
if (classSymbol == null || !processedTypes.Add(classSymbol))
{
context.ReportDiagnostic(Diagnostic.Create(
new DiagnosticDescriptor("GEN003", "Invalid Symbol",
$"Class symbol is null for {classDecl.Identifier.Text}", "Error", DiagnosticSeverity.Error, true),
Location.None));
continue;
}
context.ReportDiagnostic(Diagnostic.Create(
new DiagnosticDescriptor(
@ -114,6 +107,9 @@ namespace JiShe.CollectBus.IncrementalGenerator
Location.None));
var code = BuildAccessorsForType(classSymbol, compilation, processedTypes);
System.Diagnostics.Debug.WriteLine($"Generated code for {classSymbol.Name}:\n{code}"); // 调试输出
context.AddSource($"{classSymbol.Name}Extension.g.cs", SourceText.From(code, Encoding.UTF8));
}
}
@ -124,6 +120,8 @@ namespace JiShe.CollectBus.IncrementalGenerator
HashSet<ITypeSymbol> processedTypes)
{
var code = new StringBuilder();
code.AppendLine("#pragma warning disable CS0419 // 禁用警告");
code.AppendLine("// Generated code for " + classSymbol.Name);
code.AppendLine("// <auto-generated/>");
code.AppendLine("#nullable enable");
code.AppendLine($"namespace {classSymbol.ContainingNamespace.ToDisplayString()};");
@ -143,8 +141,14 @@ namespace JiShe.CollectBus.IncrementalGenerator
return code.ToString();
}
private static string GetGenericParams(INamedTypeSymbol symbol)
=> symbol.IsGenericType ? $"<{string.Join(", ", symbol.TypeParameters.Select(t => t.Name))}>" : "";
//private static string GetGenericParams(INamedTypeSymbol symbol)
// => symbol.IsGenericType ? $"<{string.Join(", ", symbol.TypeParameters.Select(t => t.Name))}>" : "";
public static string GetGenericParams(INamedTypeSymbol symbol)
{
if (!symbol.IsGenericType) return "";
var parameters = symbol.TypeParameters.Select(t => t.Name);
return $"<{string.Join(", ", parameters)}>";
}
private static void GeneratePropertyAccessors(
IPropertySymbol prop,

21
shared/JiShe.CollectBus.IncrementalGenerator/JiShe.CollectBus.IncrementalGenerator.csproj
View File

@ -1,14 +1,27 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<Nullable>enable</Nullable>
<TargetFramework>netstandard2.0</TargetFramework>
<!--<TargetFramework>net8.0</TargetFramework>-->
<EnforceExtendedAnalyzerRules>true</EnforceExtendedAnalyzerRules>
<TargetFramework>net8.0</TargetFramework>
<IsRoslynComponent>true</IsRoslynComponent>
<NoPackageAnalysis>true</NoPackageAnalysis>
<ImportDirectoryBuildProps>false</ImportDirectoryBuildProps>
<BaseOutputPath>bin</BaseOutputPath>
<IncludeBuildOutput>false</IncludeBuildOutput>
<LangVersion>latest</LangVersion>
<EnforceExtendedAnalyzerRules>true</EnforceExtendedAnalyzerRules>
<EmitCompilerGeneratedFiles>true</EmitCompilerGeneratedFiles>
<!-- 指定生成文件的输出目录 -->
<CompilerGeneratedFilesOutputPath>Generated</CompilerGeneratedFilesOutputPath>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.CodeAnalysis.Analyzers" Version="3.11.0" PrivateAssets="all" />
<PackageReference Include="Microsoft.CodeAnalysis.CSharp" Version="4.13.0" PrivateAssets="all" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" >
<ReferenceOutputAssembly>true</ReferenceOutputAssembly>
<PrivateAssets>all</PrivateAssets>
</ProjectReference>
</ItemGroup>
</Project>

10
web/JiShe.CollectBus.Host/appsettings.json
View File

@ -5,7 +5,7 @@
"Serilog.Sinks.File"
],
"MinimumLevel": {
"Default": "Warning",
"Default": "Information",
"Override": {
"Microsoft": "Warning",
"Volo.Abp": "Warning",
@ -87,7 +87,7 @@
"UserName": "root",
"Password": "root",
"ClusterList": [ "192.168.1.9:6667" ],
"PoolSize": 2,
"PoolSize": 32,
"DataBaseName": "energy",
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
@ -142,9 +142,9 @@
}
},
"ServerApplicationOptions": {
"ServerTagName": "JiSheCollectBus100",
"SystemType": null,
"FirstCollectionTime": "2025-04-22 16:07:00",
"ServerTagName": "JiSheCollectBus4",
"SystemType": "Energy",
"FirstCollectionTime": "2025-04-28 15:07:00",
"AutomaticVerificationTime": "16:07:00",
"AutomaticTerminalVersionTime": "17:07:00",
"AutomaticTelematicsModuleTime": "17:30:00",