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

完善时间控制

Browse Source
This commit is contained in:
ChenYi 2025-04-22 17:58:14 +08:00
parent 5ca4cbad13
commit 5647385582
7 changed files with 25 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -47,5 +47,10 @@
/// 时区，默认为："UTC+08:00"
/// </summary>
public string ZoneId { get; set; } = "UTC+08:00";
/// <summary>
/// 请求超时时间，单位毫秒，默认为：50000
/// </summary>
public long Timeout { get; set; } = 50000;
}
}

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

@ -214,6 +214,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
}
catch (Exception ex)
{
CurrentSession.Dispose();
_logger.LogError(ex, $"{nameof(QueryAsync)} IoTDB查询数据时发生异常");
throw;
}
@ -414,7 +415,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
private async Task<string> BuildQuerySQL<T>(IoTDBQueryOptions options) where T : IoTEntity
{
var metadata = await GetMetadata<T>();
var sb = new StringBuilder("SELECT ");
var sb = new StringBuilder("SELECT TIME,");
sb.AppendJoin(", ", metadata.ColumnNames);
sb.Append($" FROM {options.TableNameOrTreePath}");

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

@ -70,7 +70,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <returns></returns>
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
return await _sessionPool.ExecuteQueryStatementAsync(sql);
return await _sessionPool.ExecuteQueryStatementAsync(sql, _options.Timeout);
}
public void Dispose()

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

@ -68,7 +68,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <returns></returns>
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
return await _sessionPool.ExecuteQueryStatementAsync(sql);
return await _sessionPool.ExecuteQueryStatementAsync(sql,_options.Timeout);
}
public void Dispose()

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

@ -65,6 +65,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_producerService = producerService;
_redisDataCacheService = redisDataCacheService;
_kafkaOptions = kafkaOptions.Value;
_runtimeContext.UseTableSessionPool = true;
}
/// <summary>
@ -139,14 +141,16 @@ namespace JiShe.CollectBus.ScheduledMeterReading
var meterTypes = EnumExtensions.ToEnumDictionary<MeterTypeEnum>();
var currentTaskTime = tasksToBeIssueModel.NextTaskTime;//程序启动缓存电表的时候，NextTaskTime已经格式化到下一个采集点时间。
if (meteryType == MeterTypeEnum.Ammeter.ToString())
{
//List<MeterReadingTelemetryPacketInfo> pushTaskInfos = new();
_runtimeContext.UseTableSessionPool = true;
//_runtimeContext.UseTableSessionPool = true;
var metadata = await _dbProvider.GetMetadata<MeterReadingTelemetryPacketInfo>();
_ = CreateMeterPublishTask<AmmeterInfo>(
timeDensity: timeDensity,
nextTaskTime: tasksToBeIssueModel.NextTaskTime.CalculateNextCollectionTime(timeDensity),
nextTaskTime: currentTaskTime,
meterType: MeterTypeEnum.Ammeter,
taskCreateAction: (timeDensity, data, groupIndex, timestamps) =>
{
@ -166,7 +170,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
_ = CreateMeterPublishTask<WatermeterInfo>(
timeDensity: timeDensity,
nextTaskTime: tasksToBeIssueModel.NextTaskTime,
nextTaskTime: currentTaskTime,
meterType: MeterTypeEnum.Ammeter,
taskCreateAction: (timeDensity, data, groupIndex, taskBatch) =>
{
@ -183,8 +187,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//根据当前的采集频率和类型，重新更新下一个任务点，把任务的创建源固定在当前逻辑，避免任务处理的逻辑异常导致任务创建失败。
tasksToBeIssueModel.LastTaskTime = tasksToBeIssueModel.NextTaskTime;
tasksToBeIssueModel.NextTaskTime = tasksToBeIssueModel.NextTaskTime.CalculateNextCollectionTime(timeDensity);
tasksToBeIssueModel.LastTaskTime = currentTaskTime;
tasksToBeIssueModel.NextTaskTime = currentTaskTime.CalculateNextCollectionTime(timeDensity);
await FreeRedisProvider.Instance.SetAsync(item, tasksToBeIssueModel);
}
}
@ -248,8 +252,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
timer1.Stop();
_logger.LogError($"读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒");
//DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
//return;
DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions);
return;
#else
var meterInfos = await GetAmmeterInfoList(gatherCode);
#endif
@ -460,7 +464,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//获取缓存中的电表信息
int timeDensity = 15;
//var currentTime = DateTime.Now.CalculateNextCollectionTime(timeDensity);
var currentTime = Convert.ToDateTime("2025-04-21 17:42:00").CalculateNextCollectionTime(timeDensity);
var redisCacheKey = string.Format(RedisConst.CacheTasksToBeIssuedKey, SystemType, ServerTagName,MeterTypeEnum.Ammeter,timeDensity);
var taskInfo = await FreeRedisProvider.Instance.GetAsync<TasksToBeIssueModel>(redisCacheKey);
@ -1031,6 +1034,9 @@ namespace JiShe.CollectBus.ScheduledMeterReading
int pageNumber = 0;
bool hasNext;
var stopwatch = Stopwatch.StartNew();
var ddd = _runtimeContext.UseTableSessionPool;
do
{
options.PageIndex = pageNumber++;

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

@ -93,7 +93,7 @@
"ClusterList": [ "192.168.1.9:6667" ],
"PoolSize": 2,
"DataBaseName": "energy",
"OpenDebugMode": false,
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
},
"Cassandra": {