diff --git a/services/JiShe.CollectBus.Application.Contracts/DataChannels/IDataChannelManageService.cs b/services/JiShe.CollectBus.Application.Contracts/DataChannels/IDataChannelManageService.cs index 150b6f9..e247796 100644 --- a/services/JiShe.CollectBus.Application.Contracts/DataChannels/IDataChannelManageService.cs +++ b/services/JiShe.CollectBus.Application.Contracts/DataChannels/IDataChannelManageService.cs @@ -1,7 +1,10 @@ -using System; +using JiShe.CollectBus.IotSystems.MeterReadingRecords; +using Microsoft.AspNetCore.Mvc; +using System; using System.Collections.Generic; using System.Linq; using System.Text; +using System.Threading.Channels; using System.Threading.Tasks; namespace JiShe.CollectBus.DataChannels @@ -11,5 +14,19 @@ namespace JiShe.CollectBus.DataChannels /// public interface IDataChannelManageService { + #region 下发任务通道 + + /// + /// 定时任务数据通道写入 + /// + /// + Task ScheduledMeterTaskWriter(ChannelWriter>> _telemetryPacketInfoWriter, Tuple> dataItems); + + /// + /// 定时任务数据入库和Kafka推送通道 + /// + /// + Task ScheduledMeterTaskReadding(ChannelReader>> _telemetryPacketInfoReader); + #endregion } } diff --git a/services/JiShe.CollectBus.Application.Contracts/DataMigration/Options/DataMigrationOptions.cs b/services/JiShe.CollectBus.Application.Contracts/DataMigration/Options/DataMigrationOptions.cs index 7174cad..db6cfcc 100644 --- a/services/JiShe.CollectBus.Application.Contracts/DataMigration/Options/DataMigrationOptions.cs +++ b/services/JiShe.CollectBus.Application.Contracts/DataMigration/Options/DataMigrationOptions.cs @@ -19,7 +19,7 @@ namespace JiShe.CollectBus.DataMigration.Options /// /// 批量处理通道容量 /// - public int ChannelCapacity { get; set; } = 100; + public int ChannelCapacity { get; set; } = 100_00; /// /// 数据库 每批处理量 diff --git a/services/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs b/services/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs index 4dde0be..66cbf2f 100644 --- a/services/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs +++ b/services/JiShe.CollectBus.Application.Contracts/ScheduledMeterReading/IScheduledMeterReadingService.cs @@ -4,6 +4,7 @@ using System.Threading.Tasks; using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.GatherItem; using JiShe.CollectBus.IotSystems.Ammeters; +using JiShe.CollectBus.IotSystems.MeterReadingRecords; using JiShe.CollectBus.IotSystems.Watermeter; using Volo.Abp.Application.Services; @@ -71,7 +72,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 电表自动阀控 /// /// - Task AmmeterScheduledAutoValveControl(); + Task> AmmeterScheduledAutoValveControl(); /// /// 电表自动校时 @@ -81,7 +82,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - Task AmmeterScheduledAutomaticVerificationTime(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); + Task> AmmeterScheduledAutomaticVerificationTime(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); /// /// 日冻结抄读 @@ -91,7 +92,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - Task AmmeterScheduledGetAutomaticDayFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); + Task> AmmeterScheduledGetAutomaticDayFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); /// /// 月冻结数据抄读 @@ -101,7 +102,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - Task AmmeterScheduledGetAutomaticMonthFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); + Task> AmmeterScheduledGetAutomaticMonthFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); #endregion @@ -138,7 +139,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - Task ConcentratorScheduledAutomaticGetTerminalVersion(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); + Task> ConcentratorScheduledAutomaticGetTerminalVersion(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); /// /// 自动获取远程通信模块(SIM)版本信息 @@ -148,7 +149,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - Task ConcentratorScheduledAutomaticGetTelematicsModule(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); + Task> ConcentratorScheduledAutomaticGetTelematicsModule(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps); #endregion diff --git a/services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs b/services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs index 8b3da38..592d7e5 100644 --- a/services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs +++ b/services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs @@ -1,12 +1,18 @@ -using System.Linq; +using System; +using System.Collections.Generic; +using System.Linq; using System.Reflection; +using System.Threading.Channels; using System.Threading.Tasks; using Cassandra.Mapping; using JiShe.CollectBus.Cassandra; +using JiShe.CollectBus.DataChannels; +using JiShe.CollectBus.DataMigration.Options; using JiShe.CollectBus.FreeRedis; using JiShe.CollectBus.FreeSql; using JiShe.CollectBus.Interceptors; using JiShe.CollectBus.IoTDB; +using JiShe.CollectBus.IotSystems.MeterReadingRecords; using JiShe.CollectBus.Kafka; using JiShe.CollectBus.Mappers; using JiShe.CollectBus.Protocol; @@ -74,6 +80,8 @@ public class CollectBusApplicationModule : AbpModule // //await dbContext.InitWatermeterCacheData(); //}).ConfigureAwait(false); + //下发任务通道构建 + DataChannelManage.TaskDataChannel = Channel.CreateUnbounded>>(); //默认初始化表计信息 var dbContext = context.ServiceProvider.GetRequiredService(); diff --git a/services/JiShe.CollectBus.Application/DataChannels/DataChannelManage.cs b/services/JiShe.CollectBus.Application/DataChannels/DataChannelManage.cs new file mode 100644 index 0000000..e1f3979 --- /dev/null +++ b/services/JiShe.CollectBus.Application/DataChannels/DataChannelManage.cs @@ -0,0 +1,18 @@ +using JiShe.CollectBus.IotSystems.MeterReadingRecords; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Channels; +using System.Threading.Tasks; + +namespace JiShe.CollectBus.DataChannels +{ + public static class DataChannelManage + { + /// + /// 下发任务通道 + /// + public static Channel>> TaskDataChannel; + } +} diff --git a/services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs b/services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs index 03cb24f..c335597 100644 --- a/services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs +++ b/services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs @@ -1,27 +1,37 @@ -using JiShe.CollectBus.Application.Contracts; +using DnsClient.Protocol; +using JiShe.CollectBus.Application.Contracts; using JiShe.CollectBus.Common; +using JiShe.CollectBus.Common.DeviceBalanceControl; using JiShe.CollectBus.IoTDB.Context; using JiShe.CollectBus.IoTDB.Interface; +using JiShe.CollectBus.IotSystems.MeterReadingRecords; using JiShe.CollectBus.Kafka.Internal; using JiShe.CollectBus.Kafka.Producer; using JiShe.CollectBus.Protocol.Interfaces; using JiShe.CollectBus.Protocol.Services; using JiShe.CollectBus.RedisDataCache; using JiShe.CollectBus.ScheduledMeterReading; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.RazorPages; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Text; +using System.Threading; +using System.Threading.Channels; using System.Threading.Tasks; +using Volo.Abp.DependencyInjection; +using Volo.Abp.Timing; namespace JiShe.CollectBus.DataChannels { /// /// 数据通道管理服务 /// - public class DataChannelManageService: CollectBusAppService, IDataChannelManageService + public class DataChannelManageService : IDataChannelManageService, ITransientDependency { private readonly ILogger _logger; private readonly IIoTDbProvider _dbProvider; @@ -46,5 +56,87 @@ namespace JiShe.CollectBus.DataChannels _applicationOptions = applicationOptions.Value; _runtimeContext.UseTableSessionPool = true; } + + /// + /// 定时任务数据通道写入 + /// + /// + public async Task ScheduledMeterTaskWriter(ChannelWriter>> _telemetryPacketInfoWriter, Tuple> dataItems) + { + await _telemetryPacketInfoWriter.WriteAsync(dataItems); + } + + /// + /// 定时任务数据入库和Kafka推送通道 + /// + /// + public async Task ScheduledMeterTaskReadding(ChannelReader>> _telemetryPacketInfoReader) + { + var metadata = await _dbProvider.GetMetadata(); + var batchSize = 10000; + var timeout = TimeSpan.FromSeconds(5); // 默认超时时间为5秒 + + List taskInfoList = new List(); + var startTime = DateTime.Now; + var timer = new Stopwatch(); + while (true) + { + var canRead = await _telemetryPacketInfoReader.WaitToReadAsync(); + if (!canRead) + { + continue; + } + + while (taskInfoList != null && taskInfoList.Count < batchSize && (DateTime.Now - startTime) < timeout) + { + if (_telemetryPacketInfoReader.TryRead(out var dataItem)) + { + taskInfoList.AddRange(dataItem.Item2); + } + else + { + //无消息时短暂等待 + await Task.Delay(5); + } + } + + if (taskInfoList != null && taskInfoList.Count > 0) + { + await _dbProvider.BatchInsertAsync(metadata, taskInfoList); + + await DeviceGroupBalanceControl.ProcessWithThrottleAsync( + items: taskInfoList.ToList(), + deviceIdSelector: data => data.DeviceId, + processor: (data, groupIndex) => + { + // _ = KafkaProducerIssuedMessageAction(dateItem.Item1, data, groupIndex); + } + ); + + taskInfoList.Clear(); + } + + startTime = DateTime.Now; + } + } + + /// + /// Kafka 推送消息 + /// + /// 主题名称 + /// 任务记录 + /// 对应分区，也就是集中器号所在的分组序号 + /// + protected async Task KafkaProducerIssuedMessageAction(string topicName, + T taskRecord, int partition) where T : class + { + if (string.IsNullOrWhiteSpace(topicName) || taskRecord == null) + { + throw new Exception($"{nameof(KafkaProducerIssuedMessageAction)} 推送消息失败，参数异常,-101"); + } + + await _producerService.ProduceAsync(topicName, taskRecord, partition); + } + } } diff --git a/services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs b/services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs index 639a9c3..6d5e01a 100644 --- a/services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs +++ b/services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs @@ -28,6 +28,10 @@ using System.Linq; using System.Threading.Tasks; using JiShe.CollectBus.Protocol.Models; using System.Threading.Channels; +using static IdentityModel.ClaimComparer; +using JiShe.CollectBus.DataChannels; +using JiShe.CollectBus.DataMigration.Options; +using static System.Runtime.InteropServices.JavaScript.JSType; namespace JiShe.CollectBus.ScheduledMeterReading { @@ -38,35 +42,35 @@ namespace JiShe.CollectBus.ScheduledMeterReading { private readonly ILogger _logger; private readonly IIoTDbProvider _dbProvider; - private readonly IProducerService _producerService; + private readonly IDataChannelManageService _dataChannelManage; private readonly IRedisDataCacheService _redisDataCacheService; - private readonly KafkaOptionConfig _kafkaOptions; - private readonly ServerApplicationOptions _applicationOptions; - private readonly IoTDBRuntimeContext _runtimeContext; private readonly IProtocolService _protocolService; + private readonly DataMigrationOptions _dataMigrationOptions; + private readonly KafkaOptionConfig _kafkaOptions; + private readonly ServerApplicationOptions _applicationOptions; int pageSize = 10000; public BasicScheduledMeterReadingService( ILogger logger, - IProducerService producerService, + IDataChannelManageService dataChannelManage, IRedisDataCacheService redisDataCacheService, IIoTDbProvider dbProvider, - IoTDBRuntimeContext runtimeContext, IProtocolService protocolService, + IOptions dataMigrationOptions, IOptions kafkaOptions, IOptions applicationOptions) { _logger = logger; _dbProvider = dbProvider; - _runtimeContext = runtimeContext; - _producerService = producerService; + _dataChannelManage = dataChannelManage; _redisDataCacheService = redisDataCacheService; - _kafkaOptions = kafkaOptions.Value; - _applicationOptions = applicationOptions.Value; _protocolService = protocolService; - _runtimeContext.UseTableSessionPool = true; + _dataMigrationOptions = dataMigrationOptions.Value; + _kafkaOptions = kafkaOptions.Value; + _applicationOptions = applicationOptions.Value; + } /// @@ -111,8 +115,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading _logger.LogWarning($"{nameof(CreateToBeIssueTasks)} 构建待处理的下发指令任务处理时没有缓存数据,-101"); return; } - var currentTime = DateTime.Now; + //定时抄读 foreach (var item in taskInfos) @@ -147,7 +151,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading meterType: MeterTypeEnum.Ammeter, taskCreateAction: async (timeDensity, data, groupIndex, timestamps) => { - await AmmeterScheduledAutomaticVerificationTime(timeDensity, data, groupIndex, timestamps); + var tempTask = await AmmeterScheduledAutomaticVerificationTime(timeDensity, data, groupIndex, timestamps); + + if (tempTask == null || tempTask.Count <= 0) + { + _logger.LogWarning($"电表自动校时 {data.Name} 任务数据构建失败：{data.Serialize()}"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, tempTask)); }); } else if (string.Equals(currentTimeStr, _applicationOptions.AutomaticTerminalVersionTime, StringComparison.CurrentCultureIgnoreCase))//集中器版本号读取 @@ -158,7 +169,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading meterType: MeterTypeEnum.Ammeter, taskCreateAction: async (timeDensity, data, groupIndex, timestamps) => { - await ConcentratorScheduledAutomaticGetTerminalVersion(timeDensity, data, groupIndex, timestamps); + var tempTask = await ConcentratorScheduledAutomaticGetTerminalVersion(timeDensity, data, groupIndex, timestamps); + if (tempTask == null || tempTask.Count <= 0) + { + _logger.LogWarning($"集中器 {data.Name} 任务数据构建失败：{data.Serialize()}"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, tempTask)); }); } else if (string.Equals(currentTimeStr, _applicationOptions.AutomaticTelematicsModuleTime, StringComparison.CurrentCultureIgnoreCase))//SIM卡读取 @@ -169,7 +186,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading meterType: MeterTypeEnum.Ammeter, taskCreateAction: async (timeDensity, data, groupIndex, timestamps) => { - await ConcentratorScheduledAutomaticGetTelematicsModule(timeDensity, data, groupIndex, timestamps); + var tempTask = await ConcentratorScheduledAutomaticGetTelematicsModule(timeDensity, data, groupIndex, timestamps); + if (tempTask == null || tempTask.Count <= 0) + { + _logger.LogWarning($"集中器 {data.Name} 任务数据构建失败：{data.Serialize()}"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, tempTask)); }); } else if (string.Equals(currentTimeStr, _applicationOptions.AutomaticTelematicsModuleTime, StringComparison.CurrentCultureIgnoreCase))//月冻结 @@ -180,7 +203,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading meterType: MeterTypeEnum.Ammeter, taskCreateAction: async (timeDensity, data, groupIndex, timestamps) => { - await AmmeterScheduledGetAutomaticDayFreezeData(timeDensity, data, groupIndex, timestamps); + var tempTask = await AmmeterScheduledGetAutomaticDayFreezeData(timeDensity, data, groupIndex, timestamps); + if (tempTask == null || tempTask.Count <= 0) + { + _logger.LogWarning($"电表 {data.Name} 任务数据构建失败：{data.Serialize()}"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, tempTask)); }); } else if (string.Equals(currentTimeStr, _applicationOptions.AutomaticDayFreezeTime, StringComparison.CurrentCultureIgnoreCase))//日冻结 @@ -191,7 +220,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading meterType: MeterTypeEnum.Ammeter, taskCreateAction: async (timeDensity, data, groupIndex, timestamps) => { - await AmmeterScheduledGetAutomaticMonthFreezeData(timeDensity, data, groupIndex, timestamps); + var tempTask = await AmmeterScheduledGetAutomaticMonthFreezeData(timeDensity, data, groupIndex, timestamps); + if (tempTask == null || tempTask.Count <= 0) + { + _logger.LogWarning($"电表 {data.Name} 任务数据构建失败：{data.Serialize()}"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, tempTask)); }); } else @@ -211,7 +246,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading //tasksToBeIssueModel.NextTaskTime; - var metadata = await _dbProvider.GetMetadata(); if (meteryType == MeterTypeEnum.Ammeter.ToString()) { _ = CreateMeterPublishTask( @@ -226,7 +260,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading _logger.LogWarning($"电表 {data.Name} 任务数据构建失败：{data.Serialize()}"); return; } - _ = _dbProvider.BatchInsertAsync(metadata, tempTask); + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName, tempTask)); }); } else if (meteryType == MeterTypeEnum.WaterMeter.ToString()) @@ -245,7 +279,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading _logger.LogWarning($"水表 {data.Name} 任务数据构建失败：{data.Serialize()}"); return; } - _ = _dbProvider.BatchInsertAsync(metadata, tempTask); + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.WatermeterSubscriberWorkerAutoReadingIssuedEventName, tempTask)); }); } else @@ -264,8 +298,14 @@ namespace JiShe.CollectBus.ScheduledMeterReading } //电表定时阀控任务处理。 - _ = AmmeterScheduledAutoValveControl(); + var autoValveControlTask = await AmmeterScheduledAutoValveControl(); + if (autoValveControlTask == null || autoValveControlTask.Count <= 0) + { + _logger.LogWarning($"{nameof(AmmeterScheduledAutoValveControl)}电表定时阀控没有可操作的任务"); + return; + } + _ = _dataChannelManage.ScheduledMeterTaskWriter(DataChannelManage.TaskDataChannel.Writer, Tuple.Create(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName, autoValveControlTask)); } #region 电表采集处理 @@ -287,6 +327,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// public virtual async Task InitAmmeterCacheData(string gatherCode = "") { + _ = _dataChannelManage.ScheduledMeterTaskReadding(DataChannelManage.TaskDataChannel.Reader); + //此处代码不要删除 #if DEBUG var timeDensity = "15"; @@ -327,7 +369,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading timer1.Stop(); - _logger.LogError($"读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒"); + _logger.LogError($"电表初始化读取数据总共花费时间{timer1.ElapsedMilliseconds}毫秒"); DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista, _kafkaOptions.NumPartitions); return; #else @@ -583,26 +625,26 @@ namespace JiShe.CollectBus.ScheduledMeterReading var protocolPlugin = await _protocolService.GetProtocolServiceAsync(ammeterInfo.BrandType); if (protocolPlugin == null) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 创建电表待发送的任务数据{currentTime}没有找到对应的协议组件,-105"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 创建电表待发送的任务数据{currentTime}没有找到对应的协议组件,-105"); return null; } if (string.IsNullOrWhiteSpace(ammeterInfo.ItemCodes)) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}数据采集指令生成失败，采集项为空,-101"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}数据采集指令生成失败，采集项为空,-101"); return null; } //载波的不处理 if (ammeterInfo.MeteringPort == (int)MeterLinkProtocolEnum.Carrierwave) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}数据采集指令生成失败，载波不处理,-102"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}数据采集指令生成失败，载波不处理,-102"); return null; } if (ammeterInfo.State.Equals(2)) { - _logger.LogWarning($"{nameof(AmmerterCreatePublishTaskAction)} {ammeterInfo.Name} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}状态为禁用，不处理"); + //_logger.LogWarning($"{nameof(AmmerterCreatePublishTaskAction)} {ammeterInfo.Name} 集中器{ammeterInfo.FocusAddress}的电表{ammeterInfo.Name}状态为禁用，不处理"); return null; } @@ -615,22 +657,22 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (string.IsNullOrWhiteSpace(ammeterInfo.AreaCode)) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信区号为空"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信区号为空"); return null; } if (string.IsNullOrWhiteSpace(ammeterInfo.Address)) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信地址为空"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信地址为空"); return null; } if (Convert.ToInt32(ammeterInfo.Address) > 65535) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信地址无效,确保大于65535"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},集中器通信地址无效,确保大于65535"); return null; } if (ammeterInfo.MeteringCode <= 0 || ammeterInfo.MeteringCode > 33) { - _logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},非有效测量点号({ammeterInfo.MeteringCode})"); + //_logger.LogError($"{nameof(AmmerterCreatePublishTaskAction)} 表ID:{ammeterInfo.MeterId},非有效测量点号({ammeterInfo.MeteringCode})"); return null; } @@ -728,7 +770,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 电表自动阀控 /// /// - public virtual Task AmmeterScheduledAutoValveControl() + public virtual Task> AmmeterScheduledAutoValveControl() { throw new NotImplementedException($"{nameof(AmmeterScheduledAutoValveControl)}请根据不同系统类型进行实现"); } @@ -741,7 +783,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - public virtual async Task AmmeterScheduledAutomaticVerificationTime(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) + public virtual async Task> AmmeterScheduledAutomaticVerificationTime(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm:00}"; @@ -752,11 +794,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (!string.Equals(currentTimeStr, _applicationOptions.AutomaticVerificationTime, StringComparison.CurrentCultureIgnoreCase)) { _logger.LogInformation($"{nameof(AmmeterScheduledAutomaticVerificationTime)} 电表自动校时，非自动校时时间"); - return; + return null; } List taskList = new List(); - var metadata = await _dbProvider.GetMetadata(); var itemCode = T37612012PacketItemCodeConst.AFN10HFN01H; var subItemCode = T6452007PacketItemCodeConst.C08; @@ -766,7 +807,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (protocolPlugin == null) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest() @@ -796,21 +837,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有自动阀控任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, data, groupIndex); - } - ); + return null; //todo 阀控记录入库，推送到新的服务 } @@ -829,7 +859,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - public virtual async Task AmmeterScheduledGetAutomaticDayFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) + public virtual async Task> AmmeterScheduledGetAutomaticDayFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm:00}"; @@ -840,19 +870,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (!string.Equals(currentTimeStr, _applicationOptions.AutomaticVerificationTime, StringComparison.CurrentCultureIgnoreCase)) { _logger.LogInformation($"{nameof(AmmeterScheduledAutomaticVerificationTime)} 电表自动校时，非自动校时时间"); - return; + return null; } List taskList = new List(); - var metadata = await _dbProvider.GetMetadata(); - //根据电表型号获取协议插件 var protocolPlugin = await _protocolService.GetProtocolServiceAsync(ammeterInfo.BrandType); if (protocolPlugin == null) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } foreach (var item in DayFreezeCodes) @@ -880,21 +908,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 日冻结抄读时间{currentTime}没有任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, data, groupIndex); - } - ); + return taskList; } catch (Exception) @@ -912,7 +929,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - public virtual async Task AmmeterScheduledGetAutomaticMonthFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) + public virtual async Task> AmmeterScheduledGetAutomaticMonthFreezeData(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm:00}"; @@ -923,19 +940,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (!string.Equals(currentTimeStr, _applicationOptions.AutomaticVerificationTime, StringComparison.CurrentCultureIgnoreCase)) { _logger.LogInformation($"{nameof(AmmeterScheduledAutomaticVerificationTime)} 电表自动校时，非自动校时时间"); - return; + return null; } List taskList = new List(); - var metadata = await _dbProvider.GetMetadata(); - //根据电表型号获取协议插件 var protocolPlugin = await _protocolService.GetProtocolServiceAsync(ammeterInfo.BrandType); if (protocolPlugin == null) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } foreach (var item in DayFreezeCodes) @@ -963,21 +978,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 日冻结抄读时间{currentTime}没有任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, data, groupIndex); - } - ); + return taskList; } catch (Exception) @@ -1034,13 +1038,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading IsNumber = false, Value = false }); - await CreateMeterKafkaTaskMessage(ProtocolConst.AmmeterSubscriberWorkerRetryEventName, new IoTDBQueryOptions() - { - TableNameOrTreePath = DevicePathBuilder.GetTableName(), - PageIndex = 1, - PageSize = pageSize, - Conditions = conditions, - }); + //await CreateMeterKafkaTaskMessage(ProtocolConst.AmmeterSubscriberWorkerRetryEventName, new IoTDBQueryOptions() + //{ + // TableNameOrTreePath = DevicePathBuilder.GetTableName(), + // PageIndex = 1, + // PageSize = pageSize, + // Conditions = conditions, + //}); // 释放锁 tryLock.Unlock(); } @@ -1105,7 +1109,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading TasksToBeIssueModel nextTask = new TasksToBeIssueModel() { LastTaskTime = _applicationOptions.FirstCollectionTime.Value.CalculateNextCollectionTime(item.Key), - TimeDensity = item.Key, + TimeDensity = item.Key, }; nextTask.NextTaskTime = nextTask.LastTaskTime.CalculateNextCollectionTime(item.Key);//使用首次采集时间作为下一次采集时间 @@ -1191,13 +1195,13 @@ namespace JiShe.CollectBus.ScheduledMeterReading Value = pendingCopyReadTime }); - _ = CreateMeterKafkaTaskMessage(ProtocolConst.WatermeterSubscriberWorkerAutoReadingIssuedEventName, new IoTDBQueryOptions() - { - TableNameOrTreePath = DevicePathBuilder.GetTableName(), - PageIndex = 1, - PageSize = pageSize, - Conditions = conditions, - }); + //_ = CreateMeterKafkaTaskMessage(ProtocolConst.WatermeterSubscriberWorkerAutoReadingIssuedEventName, new IoTDBQueryOptions() + //{ + // TableNameOrTreePath = DevicePathBuilder.GetTableName(), + // PageIndex = 1, + // PageSize = pageSize, + // Conditions = conditions, + //}); _logger.LogInformation($"{nameof(WatermeterScheduledMeterAutoReadding)} {timeDensity}分钟采集水表数据处理完成"); } @@ -1321,7 +1325,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - public virtual async Task ConcentratorScheduledAutomaticGetTerminalVersion(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) + public virtual async Task> ConcentratorScheduledAutomaticGetTerminalVersion(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm:00}"; @@ -1332,7 +1336,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (!string.Equals(currentTimeStr, _applicationOptions.AutomaticTerminalVersionTime, StringComparison.CurrentCultureIgnoreCase)) { _logger.LogInformation($"{nameof(ConcentratorScheduledAutomaticGetTerminalVersion)} 集中器自动获取版本号，非自动处理时间"); - return; + return null; } List taskList = new List(); @@ -1346,7 +1350,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (protocolPlugin == null) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 集中器自动获取版本号{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest() @@ -1376,21 +1380,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有自动阀控任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, data, groupIndex); - } - ); + return taskList; } catch (Exception) { @@ -1407,7 +1400,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 集中器所在分组 /// 采集频率对应的时间戳 /// - public virtual async Task ConcentratorScheduledAutomaticGetTelematicsModule(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) + public virtual async Task> ConcentratorScheduledAutomaticGetTelematicsModule(int timeDensity, AmmeterInfo ammeterInfo, int groupIndex, DateTime timestamps) { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm:00}"; @@ -1418,11 +1411,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (!string.Equals(currentTimeStr, _applicationOptions.AutomaticTerminalVersionTime, StringComparison.CurrentCultureIgnoreCase)) { _logger.LogInformation($"{nameof(ConcentratorScheduledAutomaticGetTelematicsModule)} 自动获取远程通信模块(SIM)版本信息，非自动处理时间"); - return; + return null; } List taskList = new List(); - var metadata = await _dbProvider.GetMetadata(); var itemCode = T37612012PacketItemCodeConst.AFN09HFN09H; @@ -1431,7 +1423,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (protocolPlugin == null) { _logger.LogError($"{nameof(ConcentratorScheduledAutomaticGetTelematicsModule)} 自动获取远程通信模块(SIM)版本信息{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest() @@ -1455,21 +1447,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有自动阀控任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerOtherIssuedEventName, data, groupIndex); - } - ); + return taskList; } catch (Exception) { @@ -1549,7 +1530,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading _logger.LogError($"{nameof(CreateMeterPublishTask)} {meterType}的{timeDensity}分钟采集待下发任务创建失败，没有获取到缓存信息,-105"); return; } - _logger.LogError($"{nameof(CreateMeterPublishTask)} 采集待下发任务，缓存获取信息共花费{timer.ElapsedMilliseconds}毫秒"); + _logger.LogError($"{nameof(CreateMeterPublishTask)} 构建采集待下发任务，缓存获取信息共花费{timer.ElapsedMilliseconds}毫秒"); await DeviceGroupBalanceControl.ProcessWithThrottleAsync( items: meterInfos, @@ -1624,7 +1605,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading throw new Exception($"{nameof(KafkaProducerIssuedMessageAction)} 推送消息失败，参数异常,-101"); } - await _producerService.ProduceAsync(topicName, taskRecord, partition); + // await _dataChannelManage.ProduceAsync(topicName, taskRecord, partition); } /// diff --git a/services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs b/services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs index 455f4c2..2a96f64 100644 --- a/services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs +++ b/services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs @@ -5,6 +5,8 @@ using JiShe.CollectBus.Common.DeviceBalanceControl; using JiShe.CollectBus.Common.Enums; using JiShe.CollectBus.Common.Extensions; using JiShe.CollectBus.Common.Helpers; +using JiShe.CollectBus.DataChannels; +using JiShe.CollectBus.DataMigration.Options; using JiShe.CollectBus.FreeSql; using JiShe.CollectBus.GatherItem; using JiShe.CollectBus.IoTDB.Context; @@ -42,19 +44,19 @@ namespace JiShe.CollectBus.ScheduledMeterReading public EnergySystemScheduledMeterReadingService( ILogger logger, + IDataChannelManageService dataChannelManage, + IRedisDataCacheService redisDataCacheService, IIoTDbProvider dbProvider, - IOptions kafkaOptions, - IOptions applicationOptions, - IoTDBRuntimeContext runtimeContext, - IProducerService producerService, IProtocolService protocolService, - IRedisDataCacheService redisDataCacheService) + IOptions dataMigrationOptions, + IOptions kafkaOptions, + IOptions applicationOptions) : base(logger, - producerService, + dataChannelManage, redisDataCacheService, dbProvider, - runtimeContext, protocolService, + dataMigrationOptions, kafkaOptions, applicationOptions) { @@ -181,7 +183,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading /// 电表自动阀控 /// /// - public override async Task AmmeterScheduledAutoValveControl() + public override async Task> AmmeterScheduledAutoValveControl() { var currentTime = DateTime.Now; string currentTimeStr = $"{currentTime:HH:mm}"; @@ -194,13 +196,12 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (settingInfos == null || settingInfos.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 电表自动阀控时，阀控数据为空, -101"); - return; + return null; } //批量获取对应的缓存电表信息 var ammeterInfos = new List(); List taskList = new List(); - var metadata = await _dbProvider.GetMetadata(); foreach (var settingInfo in settingInfos) { @@ -259,7 +260,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (protocolPlugin == null) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有找到对应的协议组件,-105"); - return; + return null; } ProtocolBuildResponse builderResponse = await protocolPlugin.BuildAsync(new ProtocolBuildRequest() @@ -290,22 +291,10 @@ namespace JiShe.CollectBus.ScheduledMeterReading if (taskList == null || taskList.Count <= 0) { _logger.LogError($"{nameof(AmmeterScheduledAutoValveControl)} 定时阀控运行时间{currentTime}没有自动阀控任务生成,-106"); - return; + return null; } - //任务记录入库 - await _dbProvider.BatchInsertAsync(metadata, taskList); - - //任务信息推送Kafka - _ = DeviceGroupBalanceControl.ProcessWithThrottleAsync( - items: taskList, - deviceIdSelector: data => data.DeviceId, - processor: (data, groupIndex) => - { - _ = KafkaProducerIssuedMessageAction(ProtocolConst.AmmeterSubscriberWorkerAutoValveControlIssuedEventName, data, groupIndex); - } - ); - + return taskList; //todo 阀控记录入库，推送到新的服务 } catch (Exception) diff --git a/shared/JiShe.CollectBus.Common/DeviceBalanceControl/DeviceGroupBalanceControl.cs b/shared/JiShe.CollectBus.Common/DeviceBalanceControl/DeviceGroupBalanceControl.cs index 1734f69..4149afd 100644 --- a/shared/JiShe.CollectBus.Common/DeviceBalanceControl/DeviceGroupBalanceControl.cs +++ b/shared/JiShe.CollectBus.Common/DeviceBalanceControl/DeviceGroupBalanceControl.cs @@ -193,7 +193,7 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl int? maxConcurrency = null) { var cache = _currentCache ?? throw new InvalidOperationException("缓存未初始化"); - var timer = Stopwatch.StartNew(); + //var timer = Stopwatch.StartNew(); // 自动计算最佳并发度 int recommendedThreads = CalculateOptimalThreadCount(); @@ -227,8 +227,8 @@ namespace JiShe.CollectBus.Common.DeviceBalanceControl }); await Task.WhenAll(tasks); - timer.Stop(); - Console.WriteLine($"任务处理完成，耗时：{timer.ElapsedMilliseconds}ms"); + //timer.Stop(); + //Console.WriteLine($"任务处理完成，耗时：{timer.ElapsedMilliseconds}ms"); } /// diff --git a/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml b/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml index 8328847..30e91e8 100644 --- a/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml +++ b/web/JiShe.CollectBus.Host/Pages/Monitor.cshtml @@ -5,7 +5,7 @@ @{ Layout = null; -} +}