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-05-21 17:11:27 +08:00
parent 27b7452cd1
commit 91602a95c6
14 changed files with 381 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
modules/JiShe.CollectBus.Analyzers/ComplexTypeSourceAnalyzers.cs
View File

@ -348,44 +348,58 @@ namespace JiShe.CollectBus.IncrementalGenerator
private static string BuildFactoryCode()
{
return """
using System;
using System.Collections.Concurrent;
using System.Reflection;
using System;
using System.Collections.Concurrent;
using System.Reflection;
namespace JiShe.CollectBus.Analyzers.Shared;
namespace JiShe.CollectBus.Analyzers.Shared;
public static class SourceEntityAccessorFactory
{
private static readonly ConcurrentDictionary<Type, object> _accessors = new();
public static ISourceEntityAccessor<T> GetAccessor<T>()
public static class SourceEntityAccessorFactory
{
return (ISourceEntityAccessor<T>)_accessors.GetOrAdd(typeof(T), t =>
private static readonly ConcurrentDictionary<Type, object> _accessors = new();
public static ISourceEntityAccessor<T> GetAccessor<T>()
{
// 获取泛型类型定义信息（如果是泛型类型）
var isGeneric = t.IsGenericType;
var genericTypeDef = isGeneric ? t.GetGenericTypeDefinition() : null;
var arity = isGeneric ? genericTypeDef!.GetGenericArguments().Length : 0;
// 构建访问器类名
var typeName = isGeneric
? $"{t.Namespace}.{genericTypeDef!.Name.Split('`')[0]}Accessor`{arity}"
: $"{t.Namespace}.{t.Name}Accessor";
// 尝试从当前程序集加载
var accessorType = Assembly.GetAssembly(t)!.GetType(typeName)
?? throw new InvalidOperationException($"Accessor type {typeName} not found");
// 处理泛型参数
if (isGeneric && accessorType.IsGenericTypeDefinition)
return (ISourceEntityAccessor<T>)_accessors.GetOrAdd(typeof(T), t =>
{
accessorType = accessorType.MakeGenericType(t.GetGenericArguments());
}
// 获取泛型类型定义信息（如果是泛型类型）
var isGeneric = t.IsGenericType;
var genericTypeDef = isGeneric ? t.GetGenericTypeDefinition() : null;
var arity = isGeneric ? genericTypeDef!.GetGenericArguments().Length : 0;
return Activator.CreateInstance(accessorType)!;
});
// 构建访问器类名
var typeName = isGeneric
? $"{t.Namespace}.{genericTypeDef!.Name.Split('`')[0]}Accessor`{arity}"
: $"{t.Namespace}.{t.Name}Accessor";
// 尝试从当前程序集加载
var accessorType = Assembly.GetAssembly(t)!.GetType(typeName)
?? throw new InvalidOperationException($"Accessor type {typeName} not found");
// 处理泛型参数
if (isGeneric && accessorType.IsGenericTypeDefinition)
{
accessorType = accessorType.MakeGenericType(t.GetGenericArguments());
}
return Activator.CreateInstance(accessorType)!;
});
}
public static object GetAccessor(Type type)
{
MethodInfo getAccessorMethod = typeof(SourceEntityAccessorFactory)
.GetMethod(
name: nameof(GetAccessor),
bindingAttr: BindingFlags.Public | BindingFlags.Static,
types: Type.EmptyTypes
);
MethodInfo genericMethod = getAccessorMethod.MakeGenericMethod(type);
return genericMethod.Invoke(null, null);
}
}
}
""";
}
@ -531,8 +545,8 @@ namespace JiShe.CollectBus.IncrementalGenerator
var entityType = prop.ContainingType.ToDisplayString();//entity 实体类型名称
var propType = prop.Type;//实体属性的类型
var propTypeName = propType.ToDisplayString(SymbolDisplayFormat.FullyQualifiedFormat);
// var declaredTypeName = propType.Name; // 直接获取类型名称（如 "Int32"）
// 处理可空类型，获取底层具体类型名称
// var declaredTypeName = propType.Name; // 直接获取类型名称（如 "Int32"）
// 处理可空类型，获取底层具体类型名称
var declaredTypeName = propType switch
{
INamedTypeSymbol { OriginalDefinition.SpecialType: SpecialType.System_Nullable_T } nullableType =>
@ -583,19 +597,19 @@ namespace JiShe.CollectBus.IncrementalGenerator
$"GetValueTupleElementDeclaredTypeName(typeof({elementType})), " +//$"\"{elementDeclaredName}\", " +
$"(e) => Get{prop.Name}_{elementName}(({entityType})e), " +
$"(e, v) => Set{prop.Name}_{elementName}(({entityType})e, ({elementType})v))");
}
}
}
}
}
}
code.AppendLine(string.Join(",\n", initializerLines));
code.AppendLine(" };");
code.AppendLine(string.Join(",\n", initializerLines));
code.AppendLine(" };");
code.AppendLine(GetValueTupleElementName());
}
code.AppendLine(GetValueTupleElementName());
}
private static string GetValueTupleElementName()
{
return """
private static string GetValueTupleElementName()
{
return """
public static string GetValueTupleElementDeclaredTypeName(Type declaredType)
{
string typeName;
@ -613,7 +627,7 @@ namespace JiShe.CollectBus.IncrementalGenerator
return typeName;
}
""";
}
}
private static string GenerateAttributeInitializer(AttributeData attribute)

14
modules/JiShe.CollectBus.IoTDB/Attributes/IgnoreInitTableAttribute.cs Normal file
View File

@ -0,0 +1,14 @@
namespace JiShe.CollectBus.IoTDB.Attributes
{
/// <summary>
/// 需要忽略表模型初始化，有此特性无需初始化
/// </summary>
[AttributeUsage(AttributeTargets.Class)]
public class IgnoreInitTableAttribute : System.Attribute
{
public IgnoreInitTableAttribute()
{
}
}
}

17
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBProvider.cs
View File

@ -10,12 +10,11 @@ namespace JiShe.CollectBus.IoTDB.Interface
/// </summary>
public interface IIoTDbProvider
{
///// <summary>
///// 切换 SessionPool
///// </summary>
///// <param name="useTableSession">是否使用表模型</param>
//void SwitchSessionPool(bool useTableSession);
/// <summary>
/// 切换 SessionPool
/// </summary>
/// <param name="sessionpolType">是否使用表模型</param>
/// <returns></returns>
IIoTDbProvider GetSessionPool(bool sessionpolType);
/// <summary>
@ -66,5 +65,11 @@ namespace JiShe.CollectBus.IoTDB.Interface
/// <param name="options"></param>
/// <returns></returns>
Task<BusPagedResult<T>> QueryAsync<T>(IoTDBQueryOptions options) where T : IoTEntity, new();
/// <summary>
/// 初始化表模型
/// </summary>
/// <returns></returns>
Task InitTableSessionModelAsync();
}
}

7
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBSessionPool.cs
View File

@ -32,5 +32,12 @@ namespace JiShe.CollectBus.IoTDB.Interface
/// <param name="sql"></param>
/// <returns></returns>
Task<SessionDataSet> ExecuteQueryStatementAsync(string sql);
/// <summary>
/// 执行无返回结果SQL
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
Task<int> ExecuteNonQueryStatementAsync(string sql);
}
}

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

@ -50,7 +50,7 @@ namespace JiShe.CollectBus.IoTDB.Model
/// </summary>
private string _devicePath;
/// <summary>
/// 设备路径
/// 设备路径,树模型使用，表模型会在数据插入的时候直接获取继承类的名称作为表明
/// </summary>
public virtual string DevicePath
{

1
modules/JiShe.CollectBus.IoTDB/Model/TableModelSingleMeasuringEntity.cs
View File

@ -7,6 +7,7 @@ namespace JiShe.CollectBus.IoTDB.Model
/// Table模型单项数据实体
/// </summary>
[SourceAnalyzers(EntityTypeEnum.TableModel)]
[IgnoreInitTable]
public class TableModelSingleMeasuringEntity<T> : IoTEntity
{
/// <summary>

160
modules/JiShe.CollectBus.IoTDB/Provider/IoTDBProvider.cs
View File

@ -26,6 +26,8 @@ using System.Diagnostics.Metrics;
using Newtonsoft.Json.Linq;
using static System.Runtime.InteropServices.JavaScript.JSType;
using System.Text.RegularExpressions;
using System.Xml.Linq;
using System.Linq;
namespace JiShe.CollectBus.IoTDB.Provider
{
@ -366,6 +368,10 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
tableNameOrTreePath = metadata.TableNameOrTreePath;
}
else if(metadata.IsSingleMeasuring && string.IsNullOrWhiteSpace(metadata.TableNameOrTreePath) == true)//单侧点时，且路径没有指定，取默认实体名称和第一个列名组合。
{
tableNameOrTreePath = $"{metadata.EntityName}_{metadata.ColumnNames.First()}";
}
else
{
tableNameOrTreePath = DevicePathBuilder.GetTableName<T>();
@ -631,9 +637,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
// 过滤元组子项
if (member.NameOrPath.Contains(".Item")) continue;
// 类型名称处理
string declaredTypeName = member.DeclaredTypeName;
// 特性查询优化
var attributes = member.CustomAttributes ?? Enumerable.Empty<Attribute>();
var tagAttr = attributes.OfType<TAGColumnAttribute>().FirstOrDefault();
@ -842,6 +845,25 @@ namespace JiShe.CollectBus.IoTDB.Provider
}
}
/// <summary>
/// 处理不同列类型的逻辑
/// </summary>
/// <param name="groupedColumns"></param>
/// <param name="category"></param>
private string ProcessCategory(IReadOnlyDictionary<ColumnCategory, List<ColumnInfo>> groupedColumns, ColumnCategory category)
{
if (groupedColumns.TryGetValue(category, out var cols))
{
List<string> tempColumnInfos = new List<string>();
foreach (var item in cols)
{
tempColumnInfos.Add($" {item.Name} {item.DataType} {item.Category}");
}
return string.Join(",", tempColumnInfos);
}
return string.Empty;
}
/// <summary>
/// 数据列结构
/// </summary>
@ -976,6 +998,136 @@ namespace JiShe.CollectBus.IoTDB.Provider
return cache;
}
private static readonly Regex _asciiAlphanumericRegex = new Regex(@"^[a-zA-Z0-9]*$", RegexOptions.Compiled);
/// <summary>
/// 初始化表模型
/// </summary>
/// <returns></returns>
public async Task InitTableSessionModelAsync()
{
//获取JiShe.CollectBus.IoTDB程序集和JiShe.CollectBus.Domain程序集中的所有 [SourceAnalyzers(EntityTypeEnum.TableModel)] 的实体
var assemblyNames = new[] { "JiShe.CollectBus.IoTDB", "JiShe.CollectBus.Domain" };
var assemblies = CommonHelper.LoadAssemblies(assemblyNames);
var targetTypes = CollectTargetTypes(assemblies);
if (targetTypes == null || targetTypes.Count <= 0)
{
_logger.LogError($"{nameof(InitTableSessionModelAsync)} 初始化表模型时没有找到对应的实体类信息。");
return;
}
// @"CREATE TABLE table1(
// time TIMESTAMP TIME,
// region STRING TAG,
// plant_id STRING TAG,
// device_id STRING TAG,
// model_id STRING ATTRIBUTE,
// maintenance STRING ATTRIBUTE,
// temperature FLOAT FIELD,
// humidity FLOAT FIELD,
// status Boolean FIELD,
// arrival_time TIMESTAMP FIELD
//) COMMENT 'table1' WITH(TTL = 31536000000);";
foreach (var item in targetTypes)
{
var accessor = SourceEntityAccessorFactory.GetAccessor(item);
//通过 dynamic 简化操作
dynamic dynamicAccessor = accessor;
StringBuilder stringBuilder = new StringBuilder();
stringBuilder.Append("CREATE TABLE IF NOT EXISTS ");
stringBuilder.Append(dynamicAccessor.EntityName);
stringBuilder.Append("( time TIMESTAMP TIME, ");
var columns = new List<ColumnInfo>();
foreach (var member in dynamicAccessor.MemberList)
{
// 过滤元组子项
if (member.NameOrPath.Contains(".Item")) continue;
// 特性查询优化
var attributes = (IEnumerable<Attribute>)(member.CustomAttributes ?? Enumerable.Empty<Attribute>());
var tagAttr = attributes.OfType<TAGColumnAttribute>().FirstOrDefault();
var attrColumn = attributes.OfType<ATTRIBUTEColumnAttribute>().FirstOrDefault();
var fieldColumn = attributes.OfType<FIELDColumnAttribute>().FirstOrDefault();
// 构建ColumnInfo
ColumnInfo? column = null;
if (tagAttr != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.TAG, GetDataTypeFromTypeName(member.DeclaredTypeName), false, member.DeclaredTypeName);
}
else if (attrColumn != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.ATTRIBUTE, GetDataTypeFromTypeName(member.DeclaredTypeName), false, member.DeclaredTypeName);
}
else if (fieldColumn != null)
{
column = new ColumnInfo(member.NameOrPath, ColumnCategory.FIELD, GetDataTypeFromTypeName(member.DeclaredTypeName), false, member.DeclaredTypeName);
}
if (!column.HasValue)
{
_logger.LogError($"{nameof(InitTableSessionModelAsync)} 初始化表模型时实体类{dynamicAccessor.EntityName}的{member.NameOrPath}列的ColumnInfo构建失败。");
continue;
}
columns.Add(column.Value);
}
//按业务逻辑顺序处理（TAG -> ATTRIBUTE -> FIELD）
var groupedColumns = columns
.GroupBy(c => c.Category)
.ToDictionary(g => g.Key, g => g.ToList());
List<string> tempColumInfos = new List<string>();
tempColumInfos.Add( ProcessCategory(groupedColumns, ColumnCategory.TAG));
tempColumInfos.Add(ProcessCategory(groupedColumns, ColumnCategory.ATTRIBUTE));
tempColumInfos.Add(ProcessCategory(groupedColumns, ColumnCategory.FIELD));
stringBuilder.Append(string.Join(",", tempColumInfos.Where(d => !string.IsNullOrWhiteSpace(d))));
stringBuilder.Append($" ) COMMENT '{item.Name}' ");
_logger.LogError($"{dynamicAccessor.EntityName} 初始化语句：{stringBuilder.ToString()}");
await CurrentSession.ExecuteNonQueryStatementAsync($"{stringBuilder.ToString()}");
}
}
/// <summary>
/// 获取程序集中的所有 [SourceAnalyzers(EntityTypeEnum.TableModel)] 的实体
/// </summary>
/// <param name="assemblies"></param>
/// <returns></returns>
private List<Type> CollectTargetTypes(List<Assembly> assemblies)
{
var targetTypes = new List<Type>();
foreach (var assembly in assemblies)
{
try
{
foreach (var type in assembly.GetExportedTypes())
{
//获取表模型特性的类
var sourceAnalyzersAttribute = type.GetCustomAttribute<SourceAnalyzersAttribute>();
//需要忽略表模型初始化，有此特性无需初始化
var ignoreInitTableAttribute = type.GetCustomAttribute<IgnoreInitTableAttribute>();
if (sourceAnalyzersAttribute?.EntityType == EntityTypeEnum.TableModel && ignoreInitTableAttribute == null)
{
if (type.GetConstructor(Type.EmptyTypes) != null)
targetTypes.Add(type);
}
}
}
catch (ReflectionTypeLoadException ex)
{
_logger.LogError($"加载 {assembly} 失败: {string.Join(", ", ex.LoaderExceptions.Select(e => e.Message))}");
}
}
return targetTypes;
}
}
}

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

@ -72,7 +72,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
{
throw new Exception($"{nameof(SessionPoolAdapter)} Tree模型数据入库没有成功，返回结果为：{result}，请检查IoTEntity继承子类属性索引是否有变动。");
}
//await CloseAsync();
return result;
}
@ -84,8 +83,17 @@ namespace JiShe.CollectBus.IoTDB.Provider
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
var result = await _sessionPool.ExecuteQueryStatementAsync(sql, _options.Timeout);
//await result.Close();
//await CloseAsync();
return result;
}
/// <summary>
/// 执行无返回结果SQL
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
public async Task<int> ExecuteNonQueryStatementAsync(string sql)
{
var result = await _sessionPool.ExecuteNonQueryStatementAsync(sql);
return result;
}

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

@ -71,7 +71,6 @@ namespace JiShe.CollectBus.IoTDB.Provider
throw new Exception($"{nameof(TableSessionPoolAdapter)} table模型数据入库没有成功，返回结果为：{result}，请检查IoTEntity继承子类属性索引是否有变动。");
}
//await CloseAsync();
return result;
}
@ -83,8 +82,17 @@ namespace JiShe.CollectBus.IoTDB.Provider
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
var result = await _sessionPool.ExecuteQueryStatementAsync(sql,_options.Timeout);
//await result.Close();
//await CloseAsync();
return result;
}
/// <summary>
/// 执行无返回结果SQL
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
public async Task<int> ExecuteNonQueryStatementAsync(string sql)
{
var result = await _sessionPool.ExecuteNonQueryStatementAsync(sql);
return result;
}

33
services/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -30,6 +30,7 @@ using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Threading.Tasks;
using TouchSocket.Core;
using TouchSocket.Sockets;
@ -207,7 +208,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
time = DateTime.Now;
//System.Reflection.PropertyInfo;
//System.Reflection.FieldInfo
//TreeModelSingleMeasuringEntityAccessor
//TreeModelSingleMeasuringEntityAccessor
var meter = new TreeModelSingleMeasuringEntity<decimal?>()
{
SystemName = "energy",
@ -585,5 +586,35 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
await Task.CompletedTask;
}
/// <summary>
/// IoTDB空表查询情况
/// </summary>
/// <returns></returns>
[HttpGet]
public async Task TestIoTDBEmptyTableQuery()
{
var meter = new MeterReadingTelemetryPacketInfo() { DevicePath = "MeterReadingTelemetryPacketInfo", DeviceId = "1111" };
QueryCondition conditions = new QueryCondition()
{
Field = "DeviceId",
Operator = "=",
Value = meter.DeviceId
};
var query = new IoTDBQueryOptions()
{
TableNameOrTreePath = meter.DevicePath,
PageIndex = 1,
PageSize = 1,
Conditions = new List<QueryCondition>() { conditions },
};
await _iotDBProvider.GetSessionPool(true).InitTableSessionModelAsync();
var pageResult = await _iotDBProvider.GetSessionPool(true).QueryAsync<MeterReadingTelemetryPacketInfo>(query);
await Task.CompletedTask;
}
}

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

@ -329,6 +329,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
// 创建取消令牌源
//var cts = new CancellationTokenSource();
await _dbProvider.GetSessionPool(true).InitTableSessionModelAsync();
_ = _dataChannelManage.ScheduledMeterTaskReadingAsync(DataChannelManage.TaskDataChannel.Reader);
// //此处代码不要删除

3
services/JiShe.CollectBus.Domain/IotSystems/Devices/DeviceTableModelDataInfo.cs
View File

@ -13,7 +13,8 @@ namespace JiShe.CollectBus.IotSystems.Devices
/// 设备表型数据信息
/// </summary>
[SourceAnalyzers(EntityTypeEnum.TableModel)]
public class DeviceTreeModelDataInfo: IoTEntity
[IgnoreInitTable]
public class DeviceTableModelDataInfo : IoTEntity
{
[FIELDColumn]

4
services/JiShe.CollectBus.Domain/IotSystems/Devices/DeviceTreeModelDataInfo.cs
View File

@ -12,8 +12,8 @@ namespace JiShe.CollectBus.IotSystems.Devices
/// <summary>
/// 设备树模型数据信息
/// </summary>
[SourceAnalyzers(EntityTypeEnum.TableModel)]
public class DeviceTableModelDataInfo : IoTEntity
[SourceAnalyzers(EntityTypeEnum.TreeModel)]
public class DeviceTreeModelDataInfo : IoTEntity
{
[FIELDColumn]

67
shared/JiShe.CollectBus.Common/Helpers/CommonHelper.cs
View File

@ -865,5 +865,72 @@ namespace JiShe.CollectBus.Common.Helpers
return Convert.ToInt64(scoresStr);
}
/// <summary>
/// 加载指定名称的程序集
/// </summary>
/// <param name="assemblyNames"></param>
/// <returns></returns>
public static List<Assembly> LoadAssemblies(string[] assemblyNames)
{
var assemblies = new List<Assembly>();
// 获取已加载的程序集
foreach (var asm in AppDomain.CurrentDomain.GetAssemblies())
{
if (assemblyNames.Contains(asm.GetName().Name))
assemblies.Add(asm);
}
// 尝试加载未加载的程序集
foreach (var name in assemblyNames)
{
if (!assemblies.Any(a => a.GetName().Name == name))
{
try
{
var assembly = Assembly.Load(name);
assemblies.Add(assembly);
}
catch (FileNotFoundException)
{
// 若Load失败，尝试从基目录加载
var path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, $"{name}.dll");
if (File.Exists(path))
{
try
{
assemblies.Add(Assembly.LoadFrom(path));
}
catch { /* 记录错误 */ }
}
}
}
}
return assemblies;
}
/// <summary>
/// 创建类型实例
/// </summary>
/// <param name="types"></param>
/// <returns></returns>
public static List<object> CreateInstances(List<Type> types)
{
var instances = new List<object>();
foreach (var type in types)
{
try
{
instances.Add(Activator.CreateInstance(type));
}
catch (Exception)
{
throw;
}
}
return instances;
}
}
}