using JiShe.CollectBus.Ammeters; using JiShe.CollectBus.Application.Contracts; using JiShe.CollectBus.Common.Consts; using JiShe.CollectBus.Common.DeviceBalanceControl; using JiShe.CollectBus.Common.Enums; using JiShe.CollectBus.Common.Extensions; using JiShe.CollectBus.Common.Helpers; using JiShe.CollectBus.FreeSql; using JiShe.CollectBus.IoTDB.Context; using JiShe.CollectBus.IoTDB.Interface; using JiShe.CollectBus.IoTDB.Model; using JiShe.CollectBus.IoTDB.Options; using JiShe.CollectBus.IotSystems.Ammeters; using JiShe.CollectBus.IotSystems.PrepayModel; using JiShe.CollectBus.Kafka.Attributes; using JiShe.CollectBus.Kafka.Internal; using JiShe.CollectBus.Protocol.Interfaces; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Mvc; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using System; using System.Collections.Generic; using System.Diagnostics; using System.Threading.Tasks; namespace JiShe.CollectBus.Samples; public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe { private readonly ILogger _logger; private readonly IIoTDbProvider _iotDBProvider; private readonly IoTDBRuntimeContext _dbContext; private readonly IoTDbOptions _options; private readonly IRedisDataCacheService _redisDataCacheService; public SampleAppService(IIoTDbProvider iotDBProvider, IOptions options, IoTDBRuntimeContext dbContext, ILogger logger, IRedisDataCacheService redisDataCacheService) { _iotDBProvider = iotDBProvider; _options = options.Value; _dbContext = dbContext; _logger = logger; _redisDataCacheService = redisDataCacheService; } /// /// 测试 UseSessionPool /// /// /// [HttpGet] public async Task UseSessionPool(long testTime) { ElectricityMeterTreeModel meter = new ElectricityMeterTreeModel() { SystemName = "energy", DeviceId = "402440506s", DeviceType = "Ammeter", Current = 10, MeterModel = "DDZY-1980", ProjectId = "10059", Voltage = 10, IssuedMessageHexString = "messageHexString", Timestamps = testTime// DateTimeOffset.UtcNow.ToUnixTimeNanoseconds()//testTime.GetDateTimeOffset().ToUnixTimeNanoseconds(), }; //ElectricityMeterExtension.get await _iotDBProvider.InsertAsync(meter); } /// /// 测试Session切换 /// /// [HttpGet] public async Task UseTableSessionPool(DateTime time) { var testTime = time; ElectricityMeterTreeModel meter2 = new ElectricityMeterTreeModel() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", Current = 10, MeterModel = "DDZY-1980", ProjectId = "10059", Voltage = 10, Timestamps = DateTimeOffset.Now.ToUnixTimeNanoseconds(), }; await _iotDBProvider.InsertAsync(meter2); _dbContext.UseTableSessionPool = true; ElectricityMeter meter = new ElectricityMeter() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", Current = 10, MeterModel = "DDZY-1980", ProjectId = "10059", Voltage = 10, Timestamps = DateTimeOffset.Now.ToUnixTimeNanoseconds(), }; await _iotDBProvider.InsertAsync(meter); } /// /// 测试Session切换3 /// /// [HttpGet] public async Task UseTableSessionPool3(DateTime time) { var testTime = time; ElectricityMeterTreeModel meter2 = new ElectricityMeterTreeModel() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", Current = 10, MeterModel = "DDZY-1980", ProjectId = "10059", Voltage = 10, IssuedMessageHexString = "dsdfsfd", Timestamps = DateTimeOffset.UtcNow.ToUnixTimeNanoseconds(), }; await _iotDBProvider.InsertAsync(meter2); _dbContext.UseTableSessionPool = true; ElectricityMeter meter3 = new ElectricityMeter() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", Current = 10, MeterModel = "DDZY-1980", ProjectId = "10059", Voltage = 10, Currentd = 22, IssuedMessageHexString = "dsdfsfd", Timestamps = DateTimeOffset.Now.ToUnixTimeNanoseconds(), }; //var dd = DateTimeOffset.Now.ToUnixTimeMilliseconds(); //var dd3 = DateTimeOffset.Now.ToUnixTimeMicroseconds(); //var dd2 = DateTimeOffset.Now.ToUnixTimeNanoseconds(); await _iotDBProvider.InsertAsync(meter3); } /// /// 测试树模型单个测点数据项 /// /// /// [HttpGet] public async Task TestTreeModelSingleMeasuringEntity(string measuring, string value, DateTime time) { var meter = new TreeModelSingleMeasuringEntity() { SystemName = "energy", DeviceId = "402440506", DeviceType = "1", ProjectId = "10059", Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(), SingleMeasuring = new Tuple(measuring, value) }; await _iotDBProvider.InsertAsync(meter); } /// /// 测试树模型单个测点数据项2 /// /// /// [HttpGet] public async Task TestTreeModelSingleMeasuringEntity2(string measuring, int value, DateTime time) { var meter = new TreeModelSingleMeasuringEntity() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", ProjectId = "10059", Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(), SingleMeasuring = new Tuple(measuring, value) }; await _iotDBProvider.InsertAsync(meter); } /// /// 测试表模型单个测点数据项 /// /// /// [HttpGet] public async Task TestTableModelSingleMeasuringEntity(string measuring, string value, DateTime time) { var meter = new TableModelSingleMeasuringEntity() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", ProjectId = "10059", Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(), SingleColumn = new Tuple(measuring, value) }; _dbContext.UseTableSessionPool = true; await _iotDBProvider.InsertAsync(meter); } /// /// 测试表模型单个测点数据项2 /// /// /// [HttpGet] public async Task TestTableModelSingleMeasuringEntity2(string measuring, int value, DateTime time) { var meter = new TableModelSingleMeasuringEntity() { SystemName = "energy", DeviceId = "402440506", DeviceType = "Ammeter", ProjectId = "10059", Timestamps = time.GetDateTimeOffset().ToUnixTimeMilliseconds(), SingleColumn = new Tuple(measuring, value) }; _dbContext.UseTableSessionPool = true; await _iotDBProvider.InsertAsync(meter); } /// /// 测试设备分组均衡控制算法 /// /// /// [HttpGet] public async Task TestDeviceGroupBalanceControl(int deviceCount = 200000) { //var deviceList = new List(); //for (int i = 0; i < deviceCount; i++) //{ // deviceList.Add($"Device_{Guid.NewGuid()}"); //} //// 初始化缓存 //DeviceGroupBalanceControl.InitializeCache(deviceList); var timeDensity = "15"; //获取缓存中的电表信息 var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*"; var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList); var meterInfos = await GetMeterRedisCacheListData(oneMinutekeyList, "Energy", "JiSheCollectBus", timeDensity, MeterTypeEnum.Ammeter); List focusAddressDataLista = new List(); foreach (var item in meterInfos) { focusAddressDataLista.Add(item.FocusAddress); } DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista); // 打印分布统计 DeviceGroupBalanceControl.PrintDistributionStats(); await Task.CompletedTask; } /// /// 测试设备分组均衡控制算法获取分组Id /// /// /// [HttpGet] public async Task TestGetDeviceGroupBalanceControl(string deviceAddress) { var groupId = DeviceGroupBalanceControl.GetDeviceGroupId(deviceAddress); Console.WriteLine(groupId); await Task.CompletedTask; } /// /// 测试Redis批量读取10万条数据性能 /// /// [HttpGet] public async Task TestRedisCacheGetAllPagedData() { var timeDensity = "15"; string SystemType = "Energy"; string ServerTagName = "JiSheCollectBus2"; var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}"; var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}"; var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}"; var timer1 = Stopwatch.StartNew(); decimal? cursor = null; string member = null; bool hasNext; List meterInfos = new List(); do { var page = await _redisDataCacheService.GetAllPagedData( redisCacheMeterInfoHashKeyTemp, redisCacheMeterInfoZSetScoresIndexKeyTemp, pageSize: 1000, lastScore: cursor, lastMember: member); meterInfos.AddRange(page.Items); cursor = page.HasNext ? page.NextScore : null; member = page.HasNext ? page.NextMember : null; hasNext = page.HasNext; } while (hasNext); timer1.Stop(); _logger.LogError($"读取数据更花费时间{timer1.ElapsedMilliseconds}毫秒"); List focusAddressDataLista = new List(); foreach (var item in meterInfos) { focusAddressDataLista.Add(item.FocusAddress); } DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista); // 打印分布统计 DeviceGroupBalanceControl.PrintDistributionStats(); await Task.CompletedTask; } /// /// 下一个采集时间点验证 /// /// [HttpGet] public async Task TestCalculateNextCollectionTime(string time, int timeDensity) { DateTime nextTaskTime = Convert.ToDateTime(time); return await Task.FromResult(nextTaskTime.CalculateNextCollectionTime(timeDensity)); } public Task GetAsync() { return Task.FromResult( new SampleDto { Value = 42 } ); } [Authorize] public Task GetAuthorizedAsync() { return Task.FromResult( new SampleDto { Value = 42 } ); } [AllowAnonymous] public async Task> Test() { var ammeterList = await SqlProvider.Instance.Change(DbEnum.PrepayDB).Select().Where(d => d.TB_CustomerID == 5).Take(10).ToListAsync(); return ammeterList; } [AllowAnonymous] public bool GetTestProtocol() { var aa = LazyServiceProvider.GetKeyedService("TestProtocolPlugin"); return aa == null; } [KafkaSubscribe(ProtocolConst.TESTTOPIC)] public async Task KafkaSubscribeAsync(object obj) { _logger.LogWarning($"收到订阅消息: {obj}"); return SubscribeAck.Success(); } }