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-14 11:33:31 +08:00
parent 35c483d3e3
commit f73254e04f
8 changed files with 173 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
modules/JiShe.CollectBus.IoTDB/Options/QueryCondition.cs
View File

@ -1,4 +1,7 @@
namespace JiShe.CollectBus.IoTDB.Options
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
namespace JiShe.CollectBus.IoTDB.Options
{
/// <summary>
/// 查询条件
@ -19,10 +22,44 @@
/// 是否数值，如果是数值，则进行数值比较，否则进行字符串比较
/// </summary>
public bool IsNumber { get; set; } = false;
private object _rawValue;
/// <summary>
/// 值
/// </summary>
public object Value { get; set; }
public object Value
{
get => ApplyValueConversion(_rawValue);
set => _rawValue = value;
}
/// <summary>
/// 值转换
/// </summary>
/// <param name="rawValue"></param>
/// <returns></returns>
private object ApplyValueConversion(object rawValue)
{
string declaredTypeName = rawValue.GetType().Name;
Func<object, object> converter = GetQueryConditionValue(declaredTypeName);
return converter(rawValue);
}
/// <summary>
/// 查询条件值转换委托
/// </summary>
/// <param name="declaredTypeName"></param>
/// <returns></returns>
private Func<object, object> GetQueryConditionValue(string declaredTypeName)
{
return declaredTypeName?.ToUpper() switch
{
"DATETIME" => v => v != null ? ((DateTime)v).ToUniversalTime().Ticks : null,
"BOOLEAN" => v => v != null && (bool)v ? 1 : 0,
"STRING" => v => v != null ? $"'{v}'" : "''",
_ => v => v
};
}
}
}

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

@ -519,16 +519,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <returns></returns>
/// <exception cref="NotSupportedException"></exception>
private string TranslateCondition(QueryCondition condition)
{
{
return condition.Operator switch
{
">" => condition.IsNumber ? $"{condition.Field} > {condition.Value}" : $"{condition.Field} > '{condition.Value}'",
"<" => condition.IsNumber ? $"{condition.Field} < {condition.Value}" : $"{condition.Field} < '{condition.Value}'",
"=" => condition.IsNumber ? $"{condition.Field} = {condition.Value}" : $"{condition.Field} = '{condition.Value}'",
">" => $"{condition.Field} > {condition.Value}",
"<" => $"{condition.Field} < {condition.Value}",
"=" => $"{condition.Field} = {condition.Value}",
_ => throw new NotSupportedException($"{nameof(TranslateCondition)} 将查询条件转换为SQL语句时操作符 {condition.Operator} 属于异常情况")
};
}
/// <summary>
/// 获取查询条件的总数量
/// </summary>

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

@ -1,4 +1,5 @@
using FreeSql.Internal.CommonProvider;
using DeviceDetectorNET.Parser.Device;
using FreeSql.Internal.CommonProvider;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Consts;
@ -262,10 +263,7 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
SingleMeasuring = ("measuring", true)
};
QueryCondition conditions = new QueryCondition()
{
Field = "DeviceId",
@ -331,6 +329,35 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
};
_dbContext.UseTableSessionPool = true;
await _iotDBProvider.InsertAsync(meter);
var meter3 = new TableModelSingleMeasuringEntity<bool>()
{
SystemName = "energy",
DeviceId = "402440506",
DeviceType = "Ammeter",
ProjectId = "10059",
Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(),
SingleColumn = ("DeviceResult", true)
};
_dbContext.UseTableSessionPool = true;
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 },
};
var pageResult = await _iotDBProvider.QueryAsync<DeviceTreeModelDataInfo>(query);
}
/// <summary>

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

@ -323,7 +323,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
/// <returns></returns>
public virtual async Task InitAmmeterCacheData(string gatherCode = "")
{
//return;
return;
try
{
@ -332,34 +332,35 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_ = _dataChannelManage.ScheduledMeterTaskReadingAsync(DataChannelManage.TaskDataChannel.Reader);
//此处代码不要删除
#if DEBUG
var redisCacheDeviceInfoHashKeyTemp = $"CollectBus:Energy:JiSheCollectBus2:DeviceInfo";
// //此处代码不要删除
//#if DEBUG
// var redisCacheDeviceInfoHashKeyTemp = $"CollectBus:Energy:JiSheCollectBus2:DeviceInfo";
var timer1 = Stopwatch.StartNew();
Dictionary<string, List<DeviceInfo>> keyValuePairsTemps = FreeRedisProvider.Instance.HGetAll<List<DeviceInfo>>(redisCacheDeviceInfoHashKeyTemp);
List<DeviceInfo> meterInfos = new List<DeviceInfo>();
List<string> focusAddressDataLista = new List<string>();
foreach (var item in keyValuePairsTemps)
{
foreach (var subItem in item.Value)
{
if (subItem.MeterType == MeterTypeEnum.Ammeter && subItem.TimeDensity == 15)
{
meterInfos.Add(subItem);
focusAddressDataLista.Add(subItem.MeterId.ToString());
}
}
}
// var timer1 = Stopwatch.StartNew();
// Dictionary<string, List<DeviceInfo>> keyValuePairsTemps = FreeRedisProvider.Instance.HGetAll<List<DeviceInfo>>(redisCacheDeviceInfoHashKeyTemp);
// List<DeviceInfo> meterInfos = new List<DeviceInfo>();
// List<string> focusAddressDataLista = new List<string>();
// foreach (var item in keyValuePairsTemps)
// {
// foreach (var subItem in item.Value)
// {
// if (subItem.MeterType == MeterTypeEnum.Ammeter && subItem.TimeDensity == 15)
// {
// meterInfos.Add(subItem);
// focusAddressDataLista.Add(subItem.MeterId.ToString());
// }
// }
// }
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
var meterInfos = await GetAmmeterInfoList(gatherCode);
if (meterInfos == null || meterInfos.Count <= 0)
{
_logger.LogError($"{nameof(InitAmmeterCacheData)} 初始化电表缓存数据时，电表数据为空");

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

@ -98,8 +98,43 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//[Route($"ammeter/list")]
public override async Task<List<DeviceInfo>> GetAmmeterInfoList(string gatherCode = "V4-Gather-8890")
{
//442400040
//442400039
List<DeviceInfo> ammeterInfos = new List<DeviceInfo>();
ammeterInfos.Add(new DeviceInfo()
{
Baudrate = 2400,
FocusAddress = "442400040",
Name = "保利单箱电表1",
FocusId = 95778,
DatabaseBusiID = 1,
MeteringCode = 1,
MeterAddress = "442405000040",
MeterId = 127136,
TypeName = 3,
DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
TimeDensity = 15,
BrandType = "DDS1980",
});
ammeterInfos.Add(new DeviceInfo()
{
Baudrate = 2400,
FocusAddress = "442400039",
Name = "保利单箱电表2",
FocusId = 95778,
DatabaseBusiID = 1,
MeteringCode = 1,
MeterAddress = "442405000039",
MeterId = 127236,
TypeName = 3,
DataTypes = "449,503,581,582,583,584,585,586,587,588,589,590,591,592,593,594,597,598,599,600,601,602,603,604,605,606,607,608,661,663,677,679",
TimeDensity = 15,
BrandType = "DDS1980",
});
//List<DeviceInfo> ammeterInfos = new List<DeviceInfo>();
//ammeterInfos.Add(new DeviceInfo()
//{
// Baudrate = 2400,
@ -132,7 +167,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
// BrandType = "DDS1980",
//});
//return ammeterInfos;
return ammeterInfos;
try
{

2
services/JiShe.CollectBus.Domain/IotSystems/Devices/DeviceDataInfo.cs → services/JiShe.CollectBus.Domain/IotSystems/Devices/DeviceTableModelDataInfo.cs
View File

@ -10,7 +10,7 @@ using System.Threading.Tasks;
namespace JiShe.CollectBus.IotSystems.Devices
{
/// <summary>
/// 设备树模型数据信息
/// 设备表型数据信息
/// </summary>
[SourceAnalyzers(EntityTypeEnum.TableModel)]
public class DeviceTreeModelDataInfo: IoTEntity

22
services/JiShe.CollectBus.Domain/IotSystems/Devices/DeviceTreeModelDataInfo.cs Normal file
View File

@ -0,0 +1,22 @@
using JiShe.CollectBus.Analyzers.Shared;
using JiShe.CollectBus.IoTDB.Attributes;
using JiShe.CollectBus.IoTDB.Model;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IotSystems.Devices
{
/// <summary>
/// 设备树模型数据信息
/// </summary>
[SourceAnalyzers(EntityTypeEnum.TableModel)]
public class DeviceTableModelDataInfo : IoTEntity
{
[FIELDColumn]
public bool xfdsa { get; set; }
}
}

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

@ -35,13 +35,13 @@
"CorsOrigins": "http://localhost:4200,http://localhost:3100"
},
"ConnectionStrings": {
"Default": "mongodb://mongo_PmEeF3:lixiao1980@192.168.1.9:27017/JiSheCollectBus?authSource=admin&maxPoolSize=400&minPoolSize=10&waitQueueTimeoutMS=5000",
"Kafka": "192.168.1.9:29092,192.168.1.9:39092,192.168.1.9:49092",
"Default": "mongodb://mongo_PmEeF3:lixiao1980@192.168.5.9:27017/JiSheCollectBus?authSource=admin&maxPoolSize=400&minPoolSize=10&waitQueueTimeoutMS=5000",
"Kafka": "192.168.5.9:29092,192.168.5.9:39092,192.168.5.9:49092",
"PrepayDB": "server=118.190.144.92;database=jishe.sysdb;uid=sa;pwd=admin@2023;Encrypt=False;Trust Server Certificate=False",
"EnergyDB": "server=rm-wz9hw529i3j1e3b5fbo.sqlserver.rds.aliyuncs.com,3433;database=db_energy;uid=yjdb;pwd=Kdjdhf+9*7ad222LL;Encrypt=False;Trust Server Certificate=False"
"EnergyDB": "server=118.190.144.92;database=db_energy;uid=sa;pwd=admin@2023;Encrypt=False;Trust Server Certificate=False"
},
"Redis": {
"Configuration": "192.168.1.9:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"Configuration": "192.168.5.9:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"MaxPoolSize": "50",
"DefaultDB": "14",
"HangfireDB": "13"
@ -71,7 +71,7 @@
}
],
"Kafka": {
"BootstrapServers": "192.168.1.9:29092,192.168.1.9:39092,192.168.1.9:49092",
"BootstrapServers": "192.168.5.9:29092,192.168.5.9:39092,192.168.5.9:49092",
"EnableFilter": true,
"EnableAuthorization": false,
"SecurityProtocol": "SaslPlaintext",
@ -85,7 +85,7 @@
"IoTDBOptions": {
"UserName": "root",
"Password": "root",
"ClusterList": [ "192.168.1.9:6667" ],
"ClusterList": [ "192.168.5.9:6667" ],
"PoolSize": 32,
"DataBaseName": "energy",
"OpenDebugMode": false,
@ -103,19 +103,19 @@
},
"Nodes": [
{
"Host": "192.168.1.9",
"Host": "192.168.5.9",
"Port": 9042,
"DataCenter": "dc1",
"Rack": "RAC1"
},
{
"Host": "192.168.1.9",
"Host": "192.168.5.9",
"Port": 9043,
"DataCenter": "dc1",
"Rack": "RAC2"
},
{
"Host": "192.168.1.9",
"Host": "192.168.5.9",
"Port": 9044,
"DataCenter": "dc1",
"Rack": "RAC2"