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:
cli 2025-04-10 17:06:20 +08:00
commit 1c9094fd80
60 changed files with 2959 additions and 420 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
JiShe.CollectBus.sln
View File

@ -33,6 +33,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.FreeRedisP
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.Kafka", "src\JiShe.CollectBus.KafkaProducer\JiShe.CollectBus.Kafka.csproj", "{F0288175-F0EC-48BD-945F-CF1512850943}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.IoTDBProvider", "src\JiShe.CollectBus.IoTDBProvider\JiShe.CollectBus.IoTDBProvider.csproj", "{A3F3C092-0A25-450B-BF6A-5983163CBEF5}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -91,6 +93,14 @@ Global
{C06C4082-638F-2996-5FED-7784475766C1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C06C4082-638F-2996-5FED-7784475766C1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C06C4082-638F-2996-5FED-7784475766C1}.Release|Any CPU.Build.0 = Release|Any CPU
{919F4CDB-5C82-4371-B209-403B408DA248}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{919F4CDB-5C82-4371-B209-403B408DA248}.Debug|Any CPU.Build.0 = Debug|Any CPU
{919F4CDB-5C82-4371-B209-403B408DA248}.Release|Any CPU.ActiveCfg = Release|Any CPU
{919F4CDB-5C82-4371-B209-403B408DA248}.Release|Any CPU.Build.0 = Release|Any CPU
{A3F3C092-0A25-450B-BF6A-5983163CBEF5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{A3F3C092-0A25-450B-BF6A-5983163CBEF5}.Debug|Any CPU.Build.0 = Debug|Any CPU
{A3F3C092-0A25-450B-BF6A-5983163CBEF5}.Release|Any CPU.ActiveCfg = Release|Any CPU
{A3F3C092-0A25-450B-BF6A-5983163CBEF5}.Release|Any CPU.Build.0 = Release|Any CPU
{F0288175-F0EC-48BD-945F-CF1512850943}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{F0288175-F0EC-48BD-945F-CF1512850943}.Debug|Any CPU.Build.0 = Debug|Any CPU
{F0288175-F0EC-48BD-945F-CF1512850943}.Release|Any CPU.ActiveCfg = Release|Any CPU
@ -113,6 +123,8 @@ Global
{8BA01C3D-297D-42DF-BD63-EF07202A0A67} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
{FE0457D9-4038-4A17-8808-DCAD06CFC0A0} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
{C06C4082-638F-2996-5FED-7784475766C1} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
{919F4CDB-5C82-4371-B209-403B408DA248} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
{A3F3C092-0A25-450B-BF6A-5983163CBEF5} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
{F0288175-F0EC-48BD-945F-CF1512850943} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution

3
src/JiShe.CollectBus.Application.Contracts/Subscribers/ISubscriberAppService.cs
View File

@ -7,7 +7,8 @@ namespace JiShe.CollectBus.Subscribers
{
public interface ISubscriberAppService : IApplicationService
{
Task IssuedEvent(IssuedEventMessage issuedEventMessage);
Task LoginIssuedEvent(IssuedEventMessage issuedEventMessage);
Task HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage);
Task ReceivedEvent(MessageReceived receivedMessage);
Task ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage);
Task ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage);

2
src/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
View File

@ -15,8 +15,6 @@ namespace JiShe.CollectBus.Subscribers
#region 电表消息采集
Task<List<MeterReadingRecords>> AmmeterScheduledMeterOneMinuteReadingIssuedEventQuery();
/// <summary>
/// 1分钟采集电表数据下行消息消费订阅
/// </summary>

13
src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
View File

@ -16,9 +16,17 @@ using JiShe.CollectBus.Workers;
using Volo.Abp.BackgroundWorkers.Hangfire;
using JiShe.CollectBus.MongoDB;
using JiShe.CollectBus.ScheduledMeterReading;
using AutoMapper.Configuration.Annotations;
using JiShe.CollectBus.Common.Attributes;
using JiShe.CollectBus.IoTDBProvider;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.Protocol.Contracts;
using System.Collections.Generic;
using Thrift;
using Microsoft.Extensions.Configuration;
using Volo.Abp.EventBus.Kafka;
using Volo.Abp.Kafka;
using Microsoft.Extensions.Configuration;
using Volo.Abp.EventBus;
using Confluent.Kafka;
@ -33,7 +41,8 @@ namespace JiShe.CollectBus;
typeof(CollectBusFreeRedisModule),
typeof(CollectBusFreeSqlModule),
typeof(AbpEventBusModule),
typeof(AbpKafkaModule)
typeof(AbpKafkaModule),
typeof(CollectBusIoTDBModule)
)]
public class CollectBusApplicationModule : AbpModule
{

2
src/JiShe.CollectBus.Application/Consumers/ReceivedConsumer.cs
View File

@ -49,7 +49,7 @@ namespace JiShe.CollectBus.Consumers
var list = new List<MessageReceived>();
foreach (var contextItem in context.Message)
{
await protocolPlugin.AnalyzeAsync<TB3761FN>(contextItem.Message);
await protocolPlugin.AnalyzeAsync<TB3761>(contextItem.Message);
list.Add(contextItem.Message);
}
await _messageReceivedEventRepository.InsertManyAsync(list);

12
src/JiShe.CollectBus.Application/Consumers/WorkerConsumer.cs → src/JiShe.CollectBus.Application/Consumers/ScheduledMeterReadingConsumer.cs
View File

@ -2,6 +2,7 @@
using System.Threading.Tasks;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using MassTransit;
using Microsoft.Extensions.Logging;
using TouchSocket.Sockets;
@ -12,9 +13,9 @@ namespace JiShe.CollectBus.Consumers
/// <summary>
/// 定时抄读任务消费者
/// </summary>
public class WorkerConsumer : IConsumer<IssuedEventMessage>
public class ScheduledMeterReadingConsumer : IConsumer<ScheduledMeterReadingIssuedEventMessage>
{
private readonly ILogger<WorkerConsumer> _logger;
private readonly ILogger<ScheduledMeterReadingConsumer> _logger;
private readonly ITcpService _tcpService;
/// <summary>
@ -22,7 +23,7 @@ namespace JiShe.CollectBus.Consumers
/// </summary>
/// <param name="logger"></param>
/// <param name="tcpService"></param>
public WorkerConsumer(ILogger<WorkerConsumer> logger,
public ScheduledMeterReadingConsumer(ILogger<ScheduledMeterReadingConsumer> logger,
ITcpService tcpService)
{
_logger = logger;
@ -30,9 +31,10 @@ namespace JiShe.CollectBus.Consumers
}
public async Task Consume(ConsumeContext<IssuedEventMessage> context)
public async Task Consume(ConsumeContext<ScheduledMeterReadingIssuedEventMessage> context)
{
await _tcpService.SendAsync(context.Message.ClientId, context.Message.Message);
_logger.LogError($"{nameof(ScheduledMeterReadingConsumer)} 集中器的消息消费{context.Message.FocusAddress}");
await _tcpService.SendAsync(context.Message.FocusAddress, context.Message.MessageHexString);
}
}
}

40
src/JiShe.CollectBus.Application/DataMigration/DataMigrationService.cs
View File

@ -66,21 +66,21 @@ namespace JiShe.CollectBus.DataMigration
/// <returns></returns>
private async Task ProduceDataAsync(ChannelWriter<MeterReadingRecords[]> writer)
{
while (true)
{
var queryable = await _meterReadingRecordsRepository.GetQueryableAsync();
var batchRecords = queryable.Where(d => d.MigrationStatus == Common.Enums.RecordsDataMigrationStatusEnum.NotStarted)
.Take(_options.MongoDbDataBatchSize)
.ToArray();
//while (true)
//{
// var queryable = await _meterReadingRecordsRepository.GetQueryableAsync();
// var batchRecords = queryable.Where(d => d.MigrationStatus == Common.Enums.RecordsDataMigrationStatusEnum.NotStarted)
// .Take(_options.MongoDbDataBatchSize)
// .ToArray();
if (batchRecords == null || batchRecords.Length == 0)
{
writer.Complete();
break;
}
// if (batchRecords == null || batchRecords.Length == 0)
// {
// writer.Complete();
// break;
// }
await writer.WriteAsync(batchRecords);
}
// await writer.WriteAsync(batchRecords);
//}
}
/// <summary>
@ -111,14 +111,14 @@ namespace JiShe.CollectBus.DataMigration
//await writer.WriteAsync(dataTable);
// 批量更新标记
var ids = batch.Select(d => d.Id).ToArray();
foreach (var item in batch)
{
item.MigrationStatus = Common.Enums.RecordsDataMigrationStatusEnum.InProgress;
item.MigrationTime = DateTime.Now;
}
//var ids = batch.Select(d => d.Id).ToArray();
//foreach (var item in batch)
//{
// item.MigrationStatus = Common.Enums.RecordsDataMigrationStatusEnum.InProgress;
// item.MigrationTime = DateTime.Now;
//}
await _meterReadingRecordsRepository.UpdateManyAsync(batch);
//await _meterReadingRecordsRepository.UpdateManyAsync(batch);
}
writer.Complete();
}

34
src/JiShe.CollectBus.Application/EnergySystem/EnergySystemAppService.cs
View File

@ -71,7 +71,7 @@ namespace JiShe.CollectBus.EnergySystem
return result;
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -108,7 +108,7 @@ namespace JiShe.CollectBus.EnergySystem
foreach (var bytes in bytesList)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -149,7 +149,7 @@ namespace JiShe.CollectBus.EnergySystem
}).ToList();
var bytes = Build3761SendData.BuildAmmeterParameterSetSendCmd(address, meterParameters);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -178,7 +178,7 @@ namespace JiShe.CollectBus.EnergySystem
{
var dataUnit = Build645SendData.BuildReadMeterAddressSendDataUnit(detail.MeterAddress);
var bytes =Build3761SendData.BuildTransparentForwardingSendCmd(address, detail.Port, detail.BaudRate.ToString(), dataUnit, StopBit.Stop1, Parity.None);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -261,7 +261,7 @@ namespace JiShe.CollectBus.EnergySystem
if (bytes != null)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -320,7 +320,7 @@ namespace JiShe.CollectBus.EnergySystem
var bytes = Build3761SendData.BuildCommunicationParametersSetSendCmd(address, masterIP, materPort,
backupIP, backupPort, input.Data.APN);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -347,7 +347,7 @@ namespace JiShe.CollectBus.EnergySystem
var address = $"{input.AreaCode}{input.Address}";
var bytes = Build3761SendData.BuildTerminalCalendarClockSendCmd(address);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -375,7 +375,7 @@ namespace JiShe.CollectBus.EnergySystem
bool isManual = !input.AreaCode.Equals("5110");//低功耗集中器不是长连接，在连接的那一刻再发送
var bytes = Build3761SendData.BuildConrCheckTimeSendCmd(address,DateTime.Now, isManual);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -402,7 +402,7 @@ namespace JiShe.CollectBus.EnergySystem
var address = $"{input.AreaCode}{input.Address}";
var bytes = Build3761SendData.BuildConrRebootSendCmd(address);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -430,7 +430,7 @@ namespace JiShe.CollectBus.EnergySystem
var address = $"{input.AreaCode}{input.Address}";
var pnList = input.Data.Split(',').Select(it => int.Parse(it)).ToList();
var bytes = Build3761SendData.BuildAmmeterParameterReadingSendCmd(address, pnList);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -479,7 +479,7 @@ namespace JiShe.CollectBus.EnergySystem
foreach (var bytes in bytesList)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -548,7 +548,7 @@ namespace JiShe.CollectBus.EnergySystem
foreach (var bytes in bytesList)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -577,7 +577,7 @@ namespace JiShe.CollectBus.EnergySystem
var address = $"{code.AreaCode}{code.Address}";
var bytes = Build3761SendData.BuildAmmeterReportCollectionItemsSetSendCmd(address,input.Detail.Pn, input.Detail.Unit,input.Detail.Cycle,input.Detail.BaseTime,
input.Detail.CurveRatio,input.Detail.Details.Select(it => new PnFn(it.Pn,it.Fn)).ToList());
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -605,7 +605,7 @@ namespace JiShe.CollectBus.EnergySystem
{
var address = $"{code.AreaCode}{code.Address}";
var bytes = Build3761SendData.BuildAmmeterAutoUpSwitchSetSendCmd(address, input.Detail.Pn,input.Detail.IsOpen);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -631,7 +631,7 @@ namespace JiShe.CollectBus.EnergySystem
var result = new BaseResultDto();
var address = $"{input.AreaCode}{input.Address}";
var bytes = Build3761SendData.BuildAmmeterReadAutoUpSwitchSendCmd(address, input.Detail.Pn);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -658,7 +658,7 @@ namespace JiShe.CollectBus.EnergySystem
{
var address = $"{data.AreaCode}{data.Address}";
var bytes = Build3761SendData.BuildTerminalVersionInfoReadingSendCmd(address);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,
@ -713,7 +713,7 @@ namespace JiShe.CollectBus.EnergySystem
foreach (var bytes in bytesList)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerManualValveReadingIssuedEventName, new IssuedEventMessage
{
//ClientId = messageReceived.ClientId,
DeviceNo = address,

5
src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
View File

@ -15,13 +15,14 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="MassTransit.Kafka" Version="8.4.0" />
<PackageReference Include="Volo.Abp.AspNetCore.Mvc" Version="8.3.3" />
<PackageReference Include="Volo.Abp.AutoMapper" Version="8.3.3" />
<PackageReference Include="Volo.Abp.BackgroundWorkers.Hangfire" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Ddd.Application" Version="8.3.3" />
<PackageReference Include="TouchSocket" Version="2.1.9" />
<PackageReference Include="TouchSocket.Hosting" Version="2.1.9" />
<PackageReference Include="TouchSocket" Version="3.0.19" />
<PackageReference Include="TouchSocket.Hosting" Version="3.0.19" />
<PackageReference Include="DotNetCore.CAP" Version="8.3.1" />
<PackageReference Include="Volo.Abp.EventBus.Kafka" Version="8.3.3" />

14
src/JiShe.CollectBus.Application/Plugins/CloseMonitor.cs
View File

@ -7,10 +7,10 @@ using TouchSocket.Sockets;
namespace JiShe.CollectBus.Plugins
{
public partial class TcpCloseMonitor(ILogger<TcpCloseMonitor> logger) : PluginBase
public partial class TcpCloseMonitor(ILogger<TcpCloseMonitor> logger) : PluginBase, ITcpReceivedPlugin
{
[GeneratorPlugin(typeof(ITcpReceivedPlugin))]
public async Task OnTcpReceived(ITcpSessionClient client, ReceivedDataEventArgs e)
public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e)
{
try
{
@ -19,21 +19,21 @@ namespace JiShe.CollectBus.Plugins
catch (CloseException ex)
{
logger.LogInformation("拦截到CloseException");
client.Close(ex.Message);
await client.CloseAsync(ex.Message);
}
catch (Exception exx)
{
// ignored
}
finally
{
}
}
}
public partial class UdpCloseMonitor(ILogger<TcpCloseMonitor> logger) : PluginBase
public partial class UdpCloseMonitor(ILogger<TcpCloseMonitor> logger) : PluginBase, IUdpReceivedPlugin
{
[GeneratorPlugin(typeof(IUdpReceivedPlugin))]
public Task OnUdpReceived(IUdpSessionBase client, UdpReceivedDataEventArgs e)
{
throw new NotImplementedException();

4
src/JiShe.CollectBus.Application/Plugins/ServerMonitor.cs
View File

@ -5,9 +5,8 @@ using TouchSocket.Sockets;
namespace JiShe.CollectBus.Plugins
{
public partial class ServerMonitor(ILogger<ServerMonitor> logger) : PluginBase
public partial class ServerMonitor(ILogger<ServerMonitor> logger) : PluginBase, IServerStartedPlugin, IServerStopedPlugin
{
[GeneratorPlugin(typeof(IServerStartedPlugin))]
public Task OnServerStarted(IServiceBase sender, ServiceStateEventArgs e)
{
switch (sender)
@ -32,7 +31,6 @@ namespace JiShe.CollectBus.Plugins
return e.InvokeNext();
}
[GeneratorPlugin(typeof(IServerStopedPlugin))]
public Task OnServerStoped(IServiceBase sender,ServiceStateEventArgs e)
{
logger.LogInformation("服务已停止");

148
src/JiShe.CollectBus.Application/Plugins/TcpMonitor.cs
View File

@ -1,5 +1,7 @@
using System;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using DeviceDetectorNET.Parser.Device;
using DotNetCore.CAP;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Enums;
@ -17,12 +19,13 @@ using Volo.Abp.Caching;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Domain.Entities;
using Volo.Abp.Domain.Repositories;
using static FreeSql.Internal.GlobalFilter;
namespace JiShe.CollectBus.Plugins
{
public partial class TcpMonitor : PluginBase, ITransientDependency
public partial class TcpMonitor : PluginBase, ITransientDependency, ITcpReceivedPlugin, ITcpConnectingPlugin, ITcpConnectedPlugin, ITcpClosedPlugin
{
private readonly ICapPublisher _capBus;
private readonly ICapPublisher _producerBus;
private readonly ILogger<TcpMonitor> _logger;
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly IDistributedCache<AmmeterInfo> _ammeterInfoCache;
@ -30,23 +33,22 @@ namespace JiShe.CollectBus.Plugins
/// <summary>
///
/// </summary>
/// <param name="capBus"></param>
/// <param name="producerBus"></param>
/// <param name="logger"></param>
/// <param name="deviceRepository"></param>
/// <param name="ammeterInfoCache"></param>
public TcpMonitor(ICapPublisher capBus,
public TcpMonitor(ICapPublisher producerBus,
ILogger<TcpMonitor> logger,
IRepository<Device, Guid> deviceRepository,
IDistributedCache<AmmeterInfo> ammeterInfoCache)
{
_capBus = capBus;
_producerBus = producerBus;
_logger = logger;
_deviceRepository = deviceRepository;
_ammeterInfoCache = ammeterInfoCache;
}
[GeneratorPlugin(typeof(ITcpReceivedPlugin))]
public async Task OnTcpReceived(ITcpSessionClient client, ReceivedDataEventArgs e)
public async Task OnTcpReceived(ITcpSession client, ReceivedDataEventArgs e)
{
var messageHexString = Convert.ToHexString(e.ByteBlock.Span);
var hexStringList = messageHexString.StringToPairs();
@ -55,15 +57,20 @@ namespace JiShe.CollectBus.Plugins
var aTuple = (Tuple<string, int>)hexStringList.GetAnalyzeValue(CommandChunkEnum.A);
if (aFn.HasValue && fn.HasValue && aTuple != null && !string.IsNullOrWhiteSpace(aTuple.Item1))
{
var tcpSessionClient = (ITcpSessionClient)client;
if ((AFN)aFn == AFN.链路接口检测)
{
switch (fn)
{
case 1:
await OnTcpLoginReceived(client, messageHexString, aTuple.Item1);
await OnTcpLoginReceived(tcpSessionClient, messageHexString, aTuple.Item1);
break;
case 3:
await OnTcpHeartbeatReceived(client, messageHexString, aTuple.Item1);
//心跳帧有两种情况：
//1. 集中器先有登录帧，再有心跳帧
//2. 集中器没有登录帧，只有心跳帧
await OnTcpHeartbeatReceived(tcpSessionClient, messageHexString, aTuple.Item1);
break;
default:
_logger.LogError($"指令初步解析失败，指令内容：{messageHexString}");
@ -72,7 +79,7 @@ namespace JiShe.CollectBus.Plugins
}
else
{
await OnTcpNormalReceived(client, messageHexString, aTuple.Item1);
await OnTcpNormalReceived(tcpSessionClient, messageHexString, aTuple.Item1,aFn.ToString()!.PadLeft(2,'0'));
}
}
else
@ -83,24 +90,31 @@ namespace JiShe.CollectBus.Plugins
await e.InvokeNext();
}
[GeneratorPlugin(typeof(ITcpConnectingPlugin))]
public async Task OnTcpConnecting(ITcpSessionClient client, ConnectingEventArgs e)
//[GeneratorPlugin(typeof(ITcpConnectingPlugin))]
public async Task OnTcpConnecting(ITcpSession client, ConnectingEventArgs e)
{
_logger.LogInformation($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}正在连接中...");
var tcpSessionClient = (ITcpSessionClient)client;
_logger.LogInformation($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}正在连接中...");
await e.InvokeNext();
}
[GeneratorPlugin(typeof(ITcpConnectedPlugin))]
public async Task OnTcpConnected(ITcpSessionClient client, ConnectedEventArgs e)
//[GeneratorPlugin(typeof(ITcpConnectedPlugin))]
public async Task OnTcpConnected(ITcpSession client, ConnectedEventArgs e)
{
_logger.LogInformation($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}已连接");
var tcpSessionClient = (ITcpSessionClient)client;
_logger.LogInformation($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}已连接");
await e.InvokeNext();
}
[GeneratorPlugin(typeof(ITcpClosedPlugin))]
public async Task OnTcpClosed(ITcpSessionClient client, ClosedEventArgs e)
//[GeneratorPlugin(typeof(ITcpClosedPlugin))]//ITcpSessionClient
public async Task OnTcpClosed(ITcpSession client, ClosedEventArgs e)
{
var entity = await _deviceRepository.FindAsync(a=>a.ClientId == client.Id);
var tcpSessionClient = (ITcpSessionClient)client;
var entity = await _deviceRepository.FindAsync(a => a.ClientId == tcpSessionClient.Id);
if (entity != null)
{
entity.UpdateByOnClosed();
@ -108,7 +122,7 @@ namespace JiShe.CollectBus.Plugins
}
else
{
_logger.LogWarning($"[TCP] ID:{client.Id} IP:{client.GetIPPort()}已关闭连接,但采集程序检索失败");
_logger.LogWarning($"[TCP] ID:{tcpSessionClient.Id} IP:{client.GetIPPort()}已关闭连接,但采集程序检索失败");
}
await e.InvokeNext();
@ -123,55 +137,97 @@ namespace JiShe.CollectBus.Plugins
/// <returns></returns>
private async Task OnTcpLoginReceived(ITcpSessionClient client, string messageHexString, string deviceNo)
{
string oldClientId = $"{client.Id}";
await client.ResetIdAsync(deviceNo);
var entity = await _deviceRepository.FindAsync(a => a.Number == deviceNo);
if (entity == null)
{
await _deviceRepository.InsertAsync(new Device(deviceNo, oldClientId, DateTime.Now, DateTime.Now, DeviceStatus.Online));
}
else
{
entity.UpdateByLoginAndHeartbeat(oldClientId);
await _deviceRepository.UpdateAsync(entity);
}
var messageReceivedLoginEvent = new MessageReceivedLogin
{
ClientId = client.Id,
ClientId = deviceNo,
ClientIp = client.IP,
ClientPort = client.Port,
MessageHexString = messageHexString,
DeviceNo = deviceNo,
MessageId = NewId.NextGuid().ToString()
};
await _capBus.PublishAsync(ProtocolConst.SubscriberReceivedLoginEventName, messageReceivedLoginEvent);
var entity = await _deviceRepository.FindAsync(a => a.Number == deviceNo);
if (entity == null)
{
await _deviceRepository.InsertAsync(new Device(deviceNo, client.Id,DateTime.Now, DateTime.Now, DeviceStatus.Online));
}
else
{
entity.UpdateByLoginAndHeartbeat(client.Id);
await _deviceRepository.UpdateAsync(entity);
}
await _producerBus.PublishAsync(ProtocolConst.SubscriberLoginReceivedEventName, messageReceivedLoginEvent);
//await _producerBus.Publish( messageReceivedLoginEvent);
}
private async Task OnTcpHeartbeatReceived(ITcpSessionClient client, string messageHexString, string deviceNo)
{
string clientId = deviceNo;
string oldClientId = $"{client.Id}";
var entity = await _deviceRepository.FindAsync(a => a.Number == deviceNo);
if (entity == null) //没有登录帧的设备，只有心跳帧
{
await client.ResetIdAsync(clientId);
await _deviceRepository.InsertAsync(new Device(deviceNo, oldClientId, DateTime.Now, DateTime.Now, DeviceStatus.Online));
}
else
{
if (clientId != oldClientId)
{
entity.UpdateByLoginAndHeartbeat(oldClientId);
}
else
{
entity.UpdateByLoginAndHeartbeat();
}
await _deviceRepository.UpdateAsync(entity);
}
var messageReceivedHeartbeatEvent = new MessageReceivedHeartbeat
{
ClientId = client.Id,
ClientId = clientId,
ClientIp = client.IP,
ClientPort = client.Port,
MessageHexString = messageHexString,
DeviceNo = deviceNo,
MessageId = NewId.NextGuid().ToString()
};
await _capBus.PublishAsync(ProtocolConst.SubscriberReceivedHeartbeatEventName, messageReceivedHeartbeatEvent);
var entity = await _deviceRepository.FindAsync(a => a.Number == deviceNo);
if (entity == null)
{
await _deviceRepository.InsertAsync(new Device(deviceNo, client.Id, DateTime.Now, DateTime.Now, DeviceStatus.Online));
}
else
{
entity.UpdateByLoginAndHeartbeat(client.Id);
await _deviceRepository.UpdateAsync(entity);
}
await _producerBus.PublishAsync(ProtocolConst.SubscriberHeartbeatReceivedEventName, messageReceivedHeartbeatEvent);
//await _producerBus.Publish(messageReceivedHeartbeatEvent);
}
private async Task OnTcpNormalReceived(ITcpSessionClient client, string messageHexString, string deviceNo)
/// <summary>
/// 正常帧处理，将不同的AFN进行分发
/// </summary>
/// <param name="client"></param>
/// <param name="messageHexString"></param>
/// <param name="deviceNo"></param>
/// <param name="aFn"></param>
/// <returns></returns>
private async Task OnTcpNormalReceived(ITcpSessionClient client, string messageHexString, string deviceNo,string aFn)
{
await _capBus.PublishAsync(ProtocolConst.SubscriberReceivedEventName, new MessageReceived
//await _producerBus.Publish(new MessageReceived
//{
// ClientId = client.Id,
// ClientIp = client.IP,
// ClientPort = client.Port,
// MessageHexString = messageHexString,
// DeviceNo = deviceNo,
// MessageId = NewId.NextGuid().ToString()
//});
string topicName = string.Format(ProtocolConst.AFNTopicNameFormat, aFn);
await _producerBus.PublishAsync(topicName, new MessageReceived
{
ClientId = client.Id,
ClientIp = client.IP,

3
src/JiShe.CollectBus.Application/Plugins/UdpMonitor.cs
View File

@ -5,9 +5,8 @@ using TouchSocket.Sockets;
namespace JiShe.CollectBus.Plugins
{
public partial class UdpMonitor : PluginBase
public partial class UdpMonitor : PluginBase, IUdpReceivedPlugin
{
[GeneratorPlugin(typeof(IUdpReceivedPlugin))]
public async Task OnUdpReceived(IUdpSessionBase client, UdpReceivedDataEventArgs e)
{
var udpSession = client as UdpSession;

108
src/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -1,20 +1,86 @@
using System.Collections.Generic;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using Confluent.Kafka;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.PrepayModel;
using Microsoft.AspNetCore.Authorization;
using Volo.Abp.EventBus.Distributed;
using Volo.Abp.EventBus.Kafka;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.IoTDBProvider.Context;
using Microsoft.Extensions.Logging;
namespace JiShe.CollectBus.Samples;
public class SampleAppService : CollectBusAppService, ISampleAppService
{
private readonly ILogger<SampleAppService> _logger;
private readonly IIoTDBProvider _iotDBProvider;
private readonly IoTDBRuntimeContext _dbContext;
private readonly IoTDBOptions _options;
private readonly IDistributedEventBus _distributedEventBus;
public SampleAppService(IDistributedEventBus distributedEventBus)
public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options,
IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger)
{
_distributedEventBus = distributedEventBus;
_iotDBProvider = iotDBProvider;
_options = options.Value;
_dbContext = dbContext;
_logger = logger;
}
[HttpGet]
public async Task UseSessionPool(long timestamps)
{
string? messageHexString = null;
if (timestamps == 0)
{
timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
_logger.LogError($"timestamps_{timestamps}");
}
else
{
messageHexString = messageHexString + timestamps;
}
ElectricityMeter meter = new ElectricityMeter()
{
SystemName = "energy",
DeviceId = "402440506",
DeviceType = "Ammeter",
Current = 10,
MeterModel = "DDZY-1980",
ProjectCode = "10059",
Voltage = 10,
IssuedMessageHexString = messageHexString,
Timestamps = timestamps,
};
await _iotDBProvider.InsertAsync(meter);
}
[HttpGet]
public async Task UseTableSessionPool()
{
//_dbContext.UseTableSessionPool = true;
_iotDBProvider.SwitchSessionPool(true);
ElectricityMeter meter = new ElectricityMeter()
{
SystemName = "energy",
DeviceId = "402440506",
DeviceType = "Ammeter",
Current = 10,
MeterModel = "DDZY-1980",
ProjectCode = "10059",
Voltage = 10,
Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
};
await _iotDBProvider.InsertAsync(meter);
}
@ -46,34 +112,4 @@ public class SampleAppService : CollectBusAppService, ISampleAppService
var ammeterList = await SqlProvider.Instance.Change(DbEnum.PrepayDB).Select<Vi_BaseAmmeterInfo>().Where(d => d.TB_CustomerID == 5).Take(10).ToListAsync();
return ammeterList;
}
[AllowAnonymous]
public async Task KafkaSendTest()
{
await _distributedEventBus.PublishAsync(
new SampleDto
{
Value = 123456,
}
);
await _distributedEventBus.PublishAsync(
new SampleDto2
{
Value = 456789,
}
);
await _distributedEventBus.PublishAsync(
new SampleDto2
{
Value = 159753,
}
);
await _distributedEventBus.PublishAsync(
new SampleDto2
{
Value = 159753,
}
);
}
}

144
src/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -7,6 +7,7 @@ using System.Threading.Tasks;
using DotNetCore.CAP;
using DotNetCore.CAP.Messages;
using FreeSql;
using FreeSql.Internal.CommonProvider;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common.BuildSendDatas;
@ -17,6 +18,7 @@ using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.Enums;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
@ -28,6 +30,7 @@ using JiShe.CollectBus.Repository.MeterReadingRecord;
using JiShe.CollectBus.Workers;
using MassTransit;
using MassTransit.Internals.GraphValidation;
using MassTransit.Transports;
using Microsoft.Extensions.Logging;
using Volo.Abp.Domain.Repositories;
using static FreeSql.Internal.GlobalFilter;
@ -40,20 +43,20 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
{
private readonly ILogger<BasicScheduledMeterReadingService> _logger;
private readonly ICapPublisher _capBus;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly ICapPublisher _producerBus;
private readonly IIoTDBProvider _dbProvider;
private readonly IMeterReadingRecordRepository _meterReadingRecordRepository;
public BasicScheduledMeterReadingService(
ILogger<BasicScheduledMeterReadingService> logger,
ICapPublisher capBus,
IRepository<Device, Guid> deviceRepository,
IMeterReadingRecordRepository meterReadingRecordsRepository)
ICapPublisher producerBus,
IMeterReadingRecordRepository meterReadingRecordRepository,
IIoTDBProvider dbProvider)
{
_capBus = capBus;
_producerBus = producerBus;
_logger = logger;
_meterReadingRecordsRepository = meterReadingRecordsRepository;
_deviceRepository = deviceRepository;
_dbProvider = dbProvider;
_meterReadingRecordRepository = meterReadingRecordRepository;
}
/// <summary>
@ -204,7 +207,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
foreach (var ammeter in item)
{
//处理ItemCode
if (string.IsNullOrWhiteSpace(ammeter.ItemCodes))
if (string.IsNullOrWhiteSpace(ammeter.ItemCodes) && !string.IsNullOrWhiteSpace(ammeter.DataTypes))
{
var itemArr = ammeter.DataTypes.Split(',').ToList();
@ -272,6 +275,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
//获取缓存中的电表信息
int timeDensity = 5;
var currentTime = DateTime.Now;
var redisKeyList = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, MeterTypeEnum.Ammeter, timeDensity)}*";
var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
if (oneMinutekeyList == null || oneMinutekeyList.Length <= 0)
@ -302,14 +307,19 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
_ = _capBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500), ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
_ = _producerBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500), ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
//_= _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
//_dbProvider.SwitchSessionPool(true);
//await _dbProvider.InsertAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList,currentTime);
}
//删除任务数据
@ -337,6 +347,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
//获取缓存中的电表信息
int timeDensity = 5;
var currentTime = DateTime.Now;
var redisKeyList = $"{string.Format(RedisConst.CacheTelemetryPacketInfoKey, SystemType, MeterTypeEnum.Ammeter, timeDensity)}*";
var fiveMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
if (fiveMinutekeyList == null || fiveMinutekeyList.Length <= 0)
@ -367,14 +379,16 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
_= _capBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500), ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName, tempMsg);
_= _producerBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500), ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName, tempMsg);
//_ = _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList,currentTime);
}
//删除任务数据
@ -436,14 +450,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
_ = _capBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500) ,ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName, tempMsg);
_ = _producerBus.PublishDelayAsync(TimeSpan.FromMicroseconds(500) ,ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName, tempMsg);
//_ = _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList,currentDateTime);
}
//删除任务数据
@ -703,6 +720,8 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusID = ammeter.FocusID,
AFN = aFN,
Fn = fn,
ItemCode = tempItem,
ManualOrNot = false,
Pn = ammeter.MeteringCode,
IssuedMessageId = GuidGenerator.Create().ToString(),
IssuedMessageHexString = Convert.ToHexString(dataInfos),
@ -824,14 +843,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
await _producerBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
//_ = _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
}
//删除任务数据
@ -890,14 +912,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
await _producerBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
//_ = _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
}
//删除任务数据
@ -955,14 +980,17 @@ namespace JiShe.CollectBus.ScheduledMeterReading
FocusAddress = ammerterItem.Value.FocusAddress,
TimeDensity = timeDensity.ToString(),
};
await _capBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
//await _producerBus.PublishAsync(ProtocolConst.AmmeterSubscriberWorkerOneMinuteIssuedEventName, tempMsg);
//_ = _producerBus.Publish(tempMsg);
meterTaskInfosList.Add(ammerterItem.Value);
}
}
if (meterTaskInfosList != null && meterTaskInfosList.Count > 0)
{
await _meterReadingRecordsRepository.InsertManyAsync(meterTaskInfosList);
await _meterReadingRecordRepository.InsertManyAsync(meterTaskInfosList);
}
//删除任务数据
@ -1063,77 +1091,5 @@ namespace JiShe.CollectBus.ScheduledMeterReading
#endregion
/// <summary>
/// 测试MongoDB插入数据
/// </summary>
/// <returns></returns>
public async Task TestBatchMongoDBInsert(int totalRecords = 100_00)
{
int fetchSize = 500;
int processedSize = 4;
var tasks = new List<Task<Device>>();
var timer = Stopwatch.StartNew();
List<Device> devices = new List<Device>();
for (long timestamp = 0; timestamp < totalRecords; timestamp++)
{
var device = new Device(timestamp.ToString(),$"client{timestamp}",DateTime.Now, DateTime.Now, DeviceStatus.Online);
devices.Add(device);
if (timestamp % fetchSize == 0)
{
await _deviceRepository.InsertManyAsync(devices);
devices = new List<Device>();
}
}
timer.Stop();
var message = $"批量插入{totalRecords} 条记录，总共耗时：{timer.ElapsedMilliseconds}毫秒";
_logger.LogError(message);
}
/// <summary>
/// 测试MongoDB插入数据
/// </summary>
/// <returns></returns>
public async Task TestBatchMongoDBInsert2(int totalRecords = 100_000)
{
int fetchSize = 500;
int processedSize = 4;
var tasks = new List<Task<Device>>();
var timer = Stopwatch.StartNew();
List<Device> devices = new List<Device>();
for (long timestamp = 0; timestamp < totalRecords; timestamp++)
{
var device = new Device(timestamp.ToString(), $"client{timestamp}", DateTime.Now, DateTime.Now, DeviceStatus.Online);
devices.Add(device);
}
await _deviceRepository.InsertManyAsync(devices);
timer.Stop();
var message = $"批量插入{totalRecords} 条记录，总共耗时：{timer.ElapsedMilliseconds}毫秒";
_logger.LogError(message);
}
/// <summary>
/// 测试MongoDB插入数据
/// </summary>
/// <returns></returns>
public async Task TestSingleMongoDBInsert(int totalRecords = 100_00)
{
int fetchSize = 500;
int processedSize = 4;
var tasks = new List<Task<Device>>();
var timer = Stopwatch.StartNew();
List<Device> devices = new List<Device>();
for (long timestamp = 0; timestamp < totalRecords; timestamp++)
{
var device = new Device(timestamp.ToString(), $"client{timestamp}", DateTime.Now, DateTime.Now, DeviceStatus.Online);
devices.Add(device);
await _deviceRepository.InsertAsync(device);
}
timer.Stop();
var message = $"单次插入{totalRecords} 条记录，总共耗时：{timer.ElapsedMilliseconds}毫秒";
_logger.LogError(message);
}
}
}

37
src/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -6,12 +6,14 @@ using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Repository;
using JiShe.CollectBus.Repository.MeterReadingRecord;
using MassTransit;
using Microsoft.AspNetCore.Authorization;
using Microsoft.Extensions.Logging;
using Volo.Abp.Domain.Repositories;
@ -28,7 +30,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
{
public EnergySystemScheduledMeterReadingService(ILogger<EnergySystemScheduledMeterReadingService> logger,
ICapPublisher capBus, IMeterReadingRecordRepository _meterReadingRecordsRepository, IRepository<Device, Guid> deviceRepository) :base(logger, capBus, deviceRepository, _meterReadingRecordsRepository)
ICapPublisher producerBus, IIoTDBProvider dbProvider, IMeterReadingRecordRepository meterReadingRecordRepository) : base(logger, producerBus, meterReadingRecordRepository, dbProvider)
{
}
@ -63,6 +65,39 @@ namespace JiShe.CollectBus.ScheduledMeterReading
//[Route($"ammeter/list")]
public override async Task<List<AmmeterInfo>> GetAmmeterInfoList(string gatherCode = "V4-Gather-8890")
{
List<AmmeterInfo> ammeterInfos = new List<AmmeterInfo>();
ammeterInfos.Add(new AmmeterInfo()
{
Baudrate = 2400,
FocusAddress = "402440506",
Name = "张家祠工务(三相电表)",
FocusID = 95780,
DatabaseBusiID = 1,
MeteringCode = 1,
AmmerterAddress = "402410040506",
ID = 127035,
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,
});
ammeterInfos.Add(new AmmeterInfo()
{
Baudrate = 2400,
FocusAddress = "542400504",
Name = "五号配（长芦二所四排）(单相电表)",
FocusID = 69280,
DatabaseBusiID = 1,
MeteringCode = 2,
AmmerterAddress = "542410000504",
ID = 95594,
TypeName = 1,
DataTypes = "581,589,592,597,601",
TimeDensity = 15,
});
return ammeterInfos;
string sql = $@"SELECT C.ID,C.Name,C.FocusID,C.SingleRate,C.MeteringCode,C.Code AS BrandType,C.Baudrate,C.Password,C.MeteringPort,C.[Address] AS AmmerterAddress,C.TypeName,C.Protocol,C.TripState,C.[State],B.[Address],B.AreaCode,B.AutomaticReport,D.DataTypes,B.TimeDensity,A.GatherCode,C.Special,C.[ProjectID],B.AbnormalState,B.LastTime,CONCAT(B.AreaCode, B.[Address]) AS FocusAddress,(select top 1 DatabaseBusiID from TB_Project where ID = B.ProjectID) AS DatabaseBusiID
FROM TB_GatherInfo(NOLOCK) AS A
INNER JOIN TB_FocusInfo(NOLOCK) AS B ON A.ID = B.GatherInfoID AND B.RemoveState >= 0 AND B.State>=0

92
src/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -1,7 +1,11 @@
using System;
using System.Linq;
using System.Threading.Tasks;
using DeviceDetectorNET.Parser.Device;
using DotNetCore.CAP;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
@ -9,6 +13,7 @@ using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Protocol.Contracts.Models;
using JiShe.CollectBus.Repository.MeterReadingRecord;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using TouchSocket.Sockets;
@ -16,7 +21,7 @@ using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.Subscribers
{
public class SubscriberAppService : CollectBusAppService, ISubscriberAppService,ICapSubscribe
public class SubscriberAppService : CollectBusAppService, ISubscriberAppService, ICapSubscribe
{
private readonly ILogger<SubscriberAppService> _logger;
private readonly ITcpService _tcpService;
@ -25,7 +30,7 @@ namespace JiShe.CollectBus.Subscribers
private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
private readonly IRepository<MessageReceived, Guid> _messageReceivedEventRepository;
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly IRepository<MeterReadingRecords, Guid> _meterReadingRecordsRepository;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
@ -43,7 +48,7 @@ namespace JiShe.CollectBus.Subscribers
IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
IRepository<MessageReceived, Guid> messageReceivedEventRepository,
IRepository<Device, Guid> deviceRepository, IRepository<MeterReadingRecords, Guid> meterReadingRecordsRepository)
IRepository<Device, Guid> deviceRepository, IMeterReadingRecordRepository meterReadingRecordsRepository)
{
_logger = logger;
_tcpService = tcpService;
@ -55,19 +60,15 @@ namespace JiShe.CollectBus.Subscribers
_meterReadingRecordsRepository = meterReadingRecordsRepository;
}
[CapSubscribe(ProtocolConst.SubscriberIssuedEventName)]
public async Task IssuedEvent(IssuedEventMessage issuedEventMessage)
[CapSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
public async Task LoginIssuedEvent(IssuedEventMessage issuedEventMessage)
{
switch (issuedEventMessage.Type)
{
case IssuedEventType.Heartbeat:
_logger.LogInformation($"IssuedEvent:{issuedEventMessage.MessageId}");
var heartbeatEntity = await _messageReceivedHeartbeatEventRepository.GetAsync(a => a.MessageId == issuedEventMessage.MessageId);
heartbeatEntity.AckTime = Clock.Now;
heartbeatEntity.IsAck = true;
await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
break;
case IssuedEventType.Login:
_logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 登录回复下发内容:{issuedEventMessage.Serialize()}");
var loginEntity = await _messageReceivedLoginEventRepository.GetAsync(a => a.MessageId == issuedEventMessage.MessageId);
loginEntity.AckTime = Clock.Now;
loginEntity.IsAck = true;
@ -78,11 +79,41 @@ namespace JiShe.CollectBus.Subscribers
default:
throw new ArgumentOutOfRangeException();
}
var device = await _deviceRepository.FindAsync(a => a.Number == issuedEventMessage.DeviceNo);
if (device!=null)
{
await _tcpService.SendAsync(device.ClientId, issuedEventMessage.Message);
//var device = await _deviceRepository.FindAsync(a => a.Number == issuedEventMessage.DeviceNo);
//if (device != null)
//{
// await _tcpService.SendAsync(device.ClientId, issuedEventMessage.Message);
//}
await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
}
[CapSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
public async Task HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage)
{
switch (issuedEventMessage.Type)
{
case IssuedEventType.Heartbeat:
_logger.LogWarning($"集中器地址{issuedEventMessage.ClientId} 心跳回复下发内容:{issuedEventMessage.Serialize()}");
var heartbeatEntity = await _messageReceivedHeartbeatEventRepository.GetAsync(a => a.MessageId == issuedEventMessage.MessageId);
heartbeatEntity.AckTime = Clock.Now;
heartbeatEntity.IsAck = true;
await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
break;
case IssuedEventType.Data:
break;
default:
throw new ArgumentOutOfRangeException();
}
//var device = await _deviceRepository.FindAsync(a => a.Number == issuedEventMessage.DeviceNo);
//if (device != null)
//{
// await _tcpService.SendAsync(device.ClientId, issuedEventMessage.Message);
//}
await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
}
[CapSubscribe(ProtocolConst.SubscriberReceivedEventName)]
@ -96,12 +127,39 @@ namespace JiShe.CollectBus.Subscribers
else
{
//todo 会根据不同的协议进行解析，然后做业务处理
TB3761FN fN = await protocolPlugin.AnalyzeAsync<TB3761FN>(receivedMessage);
TB3761 fN = await protocolPlugin.AnalyzeAsync<TB3761>(receivedMessage);
if(fN == null)
{
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
return;
}
var tb3761FN = fN.FnList.FirstOrDefault();
if (tb3761FN == null)
{
Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
return;
}
//todo 查找是否有下发任务
await _meterReadingRecordsRepository.InsertAsync(new MeterReadingRecords()
{
ReceivedMessageHexString = receivedMessage.MessageHexString,
AFN = fN.Afn,
Fn = tb3761FN.Fn,
Pn = 0,
FocusAddress = "",
MeterAddress = "",
//DataResult = tb3761FN.Text,
});
//await _messageReceivedEventRepository.InsertAsync(receivedMessage);
}
}
[CapSubscribe(ProtocolConst.SubscriberReceivedHeartbeatEventName)]
[CapSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
public async Task ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage)
{
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
@ -116,7 +174,7 @@ namespace JiShe.CollectBus.Subscribers
}
}
[CapSubscribe(ProtocolConst.SubscriberReceivedLoginEventName)]
[CapSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
public async Task ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage)
{
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");

41
src/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
View File

@ -56,23 +56,6 @@ namespace JiShe.CollectBus.Subscribers
#region 电表消息采集
/// <summary>
/// 一分钟定时抄读任务消息消费订阅
/// </summary>
/// <returns></returns>
[HttpGet]
[Route("ammeter/oneminute/issued-eventQuery")]
public async Task<List<MeterReadingRecords>> AmmeterScheduledMeterOneMinuteReadingIssuedEventQuery()
{
var currentDateTime = DateTime.Now;
var list = await _meterReadingRecordsRepository.ParallelQueryAsync(currentDateTime.AddMinutes(-20), currentDateTime.AddMinutes(10));
return list;
//return null;
}
/// <summary>
/// 一分钟定时抄读任务消息消费订阅
/// </summary>
@ -147,8 +130,6 @@ namespace JiShe.CollectBus.Subscribers
}
else
{
// var dd = await _meterReadingRecordsRepository.FirstOrDefaultAsync(d=>d.ManualOrNot== true);
var device = await _deviceRepository.FirstOrDefaultAsync(a => a.Number == receivedMessage.FocusAddress);
if (device != null)
{
@ -167,7 +148,7 @@ namespace JiShe.CollectBus.Subscribers
#region 水表消息采集
/// <summary>
/// 一分钟定时抄读任务消息消费订阅
/// 1分钟采集水表数据下行消息消费订阅
/// </summary>
/// <param name="receivedMessage"></param>
/// <returns></returns>
@ -176,11 +157,11 @@ namespace JiShe.CollectBus.Subscribers
[CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerOneMinuteIssuedEventName)]
public async Task WatermeterScheduledMeterOneMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("1分钟采集电表数据下行消息消费队列开始处理");
_logger.LogInformation("1分钟采集水表数据下行消息消费队列开始处理");
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
{
_logger.LogError("【1分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
_logger.LogError("【1分钟采集水表数据下行消息消费队列开始处理】协议不存在！");
}
else
{
@ -194,20 +175,20 @@ namespace JiShe.CollectBus.Subscribers
}
/// <summary>
/// 5分钟采集电表数据下行消息消费订阅
/// 5分钟采集水表数据下行消息消费订阅
/// </summary>
/// <param name="receivedMessage"></param>
/// <returns></returns>
[HttpPost]
[Route("watermeter/fiveminute/issued-event")]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFiveMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerFiveMinuteIssuedEventName)]
public async Task WatermeterScheduledMeterFiveMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("5分钟采集电表数据下行消息消费队列开始处理");
_logger.LogInformation("5分钟采集水表数据下行消息消费队列开始处理");
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
{
_logger.LogError("【5分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
_logger.LogError("【5分钟采集水表数据下行消息消费队列开始处理】协议不存在！");
}
else
{
@ -221,20 +202,20 @@ namespace JiShe.CollectBus.Subscribers
}
/// <summary>
/// 15分钟采集电表数据下行消息消费订阅
/// 15分钟采集水表数据下行消息消费订阅
/// </summary>
/// <param name="receivedMessage"></param>
/// <returns></returns>
[HttpPost]
[Route("watermeter/fifteenminute/issued-event")]
[CapSubscribe(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName)]
[CapSubscribe(ProtocolConst.WatermeterSubscriberWorkerFifteenMinuteIssuedEventName)]
public async Task WatermeterScheduledMeterFifteenMinuteReadingIssuedEvent(ScheduledMeterReadingIssuedEventMessage receivedMessage)
{
_logger.LogInformation("15分钟采集电表数据下行消息消费队列开始处理");
_logger.LogInformation("15分钟采集水表数据下行消息消费队列开始处理");
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
{
_logger.LogError("【15分钟采集电表数据下行消息消费队列开始处理】协议不存在！");
_logger.LogError("【15分钟采集水表数据下行消息消费队列开始处理】协议不存在！");
}
else
{

28
src/JiShe.CollectBus.Common/AttributeInfo/NumericalOrderAttribute.cs Normal file
View File

@ -0,0 +1,28 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Common.AttributeInfo
{
/// <summary>
/// 排序序号
/// </summary>
public class NumericalOrderAttribute : Attribute
{
/// <summary>
/// 序号
/// </summary>
public int Index { get; set; }
/// <summary>
/// 排序序号
/// </summary>
/// <param name="index"></param>
public NumericalOrderAttribute(int index)
{
Index = index;
}
}
}

10
src/JiShe.CollectBus.Common/Extensions/DateTimeExtensions.cs
View File

@ -167,5 +167,15 @@ namespace JiShe.CollectBus.Common.Extensions
)
);
}
/// <summary>
/// 获取数据表分片策略
/// </summary>
/// <param name="dateTime"></param>
/// <returns></returns>
public static string GetDataTableShardingStrategy(this DateTime dateTime)
{
return $"{dateTime:yyyyMMddHHmm}";
}
}
}

83
src/JiShe.CollectBus.Common/Extensions/EnumExtensions.cs Normal file
View File

@ -0,0 +1,83 @@
using System;
namespace JiShe.CollectBus.Common.Extensions
{
public static class EnumExtensions
{
/// <summary>
/// 将枚举转换为<string, int>字典
/// </summary>
/// <typeparam name="TEnum"></typeparam>
/// <returns></returns>
public static Dictionary<string, int> ToDictionary<TEnum>() where TEnum : Enum
{
return Enum.GetValues(typeof(TEnum))
.Cast<TEnum>()
.ToDictionary(
e => e.ToString(),
e => Convert.ToInt32(e)
);
}
/// <summary>
/// 将枚举转换为<string, int>字典
/// </summary>
/// <typeparam name="TEnum"></typeparam>
/// <returns></returns>
public static Dictionary<string, TEnum> ToEnumDictionary<TEnum>() where TEnum : Enum
{
return Enum.GetValues(typeof(TEnum))
.Cast<TEnum>()
.ToDictionary(
e => e.ToString(),
e => e
);
}
/// <summary>
/// 将枚举转换为<int, string>字典
/// </summary>
/// <typeparam name="TEnum"></typeparam>
/// <returns></returns>
public static Dictionary<int, string> ToValueNameDictionary<TEnum>() where TEnum : Enum
{
return Enum.GetValues(typeof(TEnum))
.Cast<TEnum>()
.ToDictionary(
e => Convert.ToInt32(e),
e => e.ToString()
);
}
/// <summary>
/// 将枚举转换为<int, string>字典
/// </summary>
/// <typeparam name="TEnum"></typeparam>
/// <returns></returns>
public static Dictionary<string, int> ToNameValueDictionary<TEnum>() where TEnum : Enum
{
return Enum.GetValues(typeof(TEnum))
.Cast<TEnum>()
.ToDictionary(
e => e.ToString(),
e => Convert.ToInt32(e)
);
}
/// <summary>
/// 将枚举转换为<TEnum, string>字典
/// </summary>
/// <typeparam name="TEnum"></typeparam>
/// <returns></returns>
public static Dictionary<TEnum, string> ToEnumNameDictionary<TEnum>() where TEnum : Enum
{
return Enum.GetValues(typeof(TEnum))
.Cast<TEnum>()
.ToDictionary(
e => e,
e => e.ToString()
);
}
}
}

798
src/JiShe.CollectBus.Common/Helpers/CommonHelper.cs Normal file
View File

@ -0,0 +1,798 @@
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.ComponentModel;
using System.Linq;
using System.Reflection;
using System.Runtime.InteropServices;
using System.Security.Cryptography;
using System.Text;
using System.Threading.Tasks;
using JiShe.CollectBus.Common.AttributeInfo;
namespace JiShe.CollectBus.Common.Helpers
{
public static class CommonHelper
{
/// <summary>
/// 获得无符号GUID
/// </summary>
/// <returns></returns>
public static string GetGUID()
{
return Guid.NewGuid().ToString("N");
}
/// <summary>
/// 获取时间戳
/// </summary>
/// <param name="isSeconds">是否返回秒，false返回毫秒</param>
/// <returns></returns>
public static long GetTimeStampTen(bool isSeconds)
{
if (isSeconds)
{
return DateTimeOffset.UtcNow.ToUnixTimeSeconds();
}
else
{
return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
}
/// <summary>
/// 获取指定长度的随机数
/// </summary>
/// <param name="length"></param>
/// <returns></returns>
public static string GetRandomNumber(int length = 8)
{
if (length <= 8)
{
length = 8;
}
if (length > 31)
{
length = 32;
}
var randomArray = RandomNumberGenerator.GetBytes(length);
StringBuilder stringBuilder = new StringBuilder();
foreach (var item in randomArray)
{
stringBuilder.Append(item);
}
return stringBuilder.ToString().Substring(0, length);
}
/// <summary>
/// C#反射遍历对象属性获取键值对
/// </summary>
/// <typeparam name="T">对象类型</typeparam>
/// <param name="model">对象</param>
public static Dictionary<string, object> GetClassProperties<T>(T model)
{
Type t = model.GetType();
List<PropertyInfo> propertyList = new List<PropertyInfo>();
PropertyInfo[] tempPropertyList = t.GetProperties();
if (tempPropertyList != null && tempPropertyList.Length > 0)
{
propertyList.AddRange(tempPropertyList);
}
var parentPropertyInfo = t.BaseType?.GetProperties();
if (parentPropertyInfo != null && parentPropertyInfo.Length > 0)
{
foreach (var item in parentPropertyInfo)
{
if (!propertyList.Any(d => d.Name == item.Name)) //如果子类已经包含父类的属性或者字段，跳过不处理
{
propertyList.Add(item);
}
}
}
Dictionary<string, object> resultData = new Dictionary<string, object>();
foreach (PropertyInfo item in propertyList)
{
resultData.Add(item.Name, item.GetValue(model, null));
}
return resultData;
}
/// <summary>
/// C#反射遍历对象属性，将键值对赋值给属性
/// </summary>
public static object SetClassProperties(string typeModel, Dictionary<string, object> keyValues)
{
if (keyValues.Count <= 0)
{
return null;
}
Type tType = Type.GetType(typeModel);
PropertyInfo[] PropertyList = tType.GetProperties();
object objModel = tType.Assembly.CreateInstance(tType.FullName);
foreach (PropertyInfo item in PropertyList)
{
if (keyValues.ContainsKey(item.Name))
{
object objectValue = keyValues[item.Name];
item.SetValue(objModel, objectValue);
}
}
return objModel;
}
/// <summary>
/// 取得某月的第一天
/// </summary>
/// <param name="datetime">要取得月份第一天的时间</param>
/// <returns></returns>
public static DateTime FirstDayOfMonth(this DateTime datetime)
{
return datetime.AddDays(1 - datetime.Day);
}
///<summary>
/// 取得某月的最后一天
/// </summary>
/// <param name="datetime">要取得月份最后一天的时间</param>
/// <returns></returns>
public static DateTime LastDayOfMonth(this DateTime datetime)
{
return datetime.AddDays(1 - datetime.Day).AddMonths(1).AddDays(-1);
}
/// <summary>
/// 取得某月第一天0点以及最后一天的23:59:59时间范围
/// </summary>
/// <param name="datetime"></param>
/// <returns></returns>
public static Tuple<DateTime, DateTime> GetMonthDateRange(this DateTime datetime)
{
var lastDayOfMonthDate = LastDayOfMonth(datetime);
return new Tuple<DateTime, DateTime>(datetime.FirstDayOfMonth(), new DateTime(lastDayOfMonthDate.Year, lastDayOfMonthDate.Month, lastDayOfMonthDate.Day, 23, 59, 59));
}
/// <summary>
/// 取得某一天0点到当月最后一天的23:59:59时间范围
/// </summary>
/// <param name="datetime"></param>
/// <returns></returns>
public static Tuple<DateTime, DateTime> GetCurrentDateToLastDayRange(this DateTime datetime)
{
var lastDayOfMonthDate = LastDayOfMonth(datetime);
return new Tuple<DateTime, DateTime>(datetime.Date, new DateTime(lastDayOfMonthDate.Year, lastDayOfMonthDate.Month, lastDayOfMonthDate.Day, 23, 59, 59));
}
/// <summary>
/// 取得某一天0点到23:59:59时间范围
/// </summary>
/// <param name="datetime"></param>
/// <returns></returns>
public static Tuple<DateTime, DateTime> GetCurrentDateRange(this DateTime datetime)
{
return new Tuple<DateTime, DateTime>(datetime.Date, new DateTime(datetime.Year, datetime.Month, datetime.Day, 23, 59, 59));
}
/// <summary>
/// 获取指定枚举的所有 Attribute 说明以及value组成的键值对
/// </summary>
/// <param name="type">对象类</param>
/// <param name="getPropertie">false获取字段、true获取属性</param>
/// <returns></returns>
public static List<SelectResult> GetEnumAttributeList(Type type, bool getPropertie = false)
{
if (type == null)
{
return null;
}
List<SelectResult> selectResults = new List<SelectResult>();
List<MemberInfo> memberInfos = new List<MemberInfo>();
if (getPropertie == false)
{
FieldInfo[] fieldArray = type.GetFields();
if (null == fieldArray || fieldArray.Length <= 0)
{
return null;
}
memberInfos.AddRange(fieldArray);
//获取父类的字段
var parentFieldInfo = type.BaseType?.GetFields();
if (parentFieldInfo != null && parentFieldInfo.Length > 0)
{
foreach (var item in parentFieldInfo)
{
if (!memberInfos.Any(d => d.Name == item.Name)) //如果子类已经包含父类的属性或者字段，跳过不处理
{
memberInfos.Add(item);
}
}
}
}
else
{
PropertyInfo[] properties = type.GetProperties();
if (null == properties || properties.Length <= 0)
{
return null;
}
memberInfos.AddRange(properties);
//获取父类的属性
var parentPropertyInfo = type.BaseType?.GetProperties();
if (parentPropertyInfo != null && parentPropertyInfo.Length > 0)
{
foreach (var item in parentPropertyInfo)
{
if (!memberInfos.Any(d => d.Name == item.Name)) //如果子类已经包含父类的属性或者字段，跳过不处理
{
memberInfos.Add(item);
}
}
}
}
foreach (var item in memberInfos)
{
DescriptionAttribute[] EnumAttributes =
(DescriptionAttribute[])item.GetCustomAttributes(typeof(DescriptionAttribute), false);
dynamic infoObject = null;
if (getPropertie == false)
{
infoObject = (FieldInfo)item;
}
else
{
infoObject = (PropertyInfo)item;
}
if (EnumAttributes.Length > 0)
{
SelectResult selectResult = new SelectResult()
{
Key = Convert.ToInt32(infoObject.GetValue(null)).ToString(),
Value = EnumAttributes[0].Description,
};
selectResults.Add(selectResult);
}
DisplayAttribute[] DisplayAttributes =
(DisplayAttribute[])item.GetCustomAttributes(typeof(DisplayAttribute), false);
if (DisplayAttributes.Length > 0)
{
SelectResult selectResult =
selectResults.FirstOrDefault(e => e.Key == Convert.ToInt32(infoObject.GetValue(null)).ToString());
if (selectResult != null)
{
selectResult.SecondValue = DisplayAttributes[0].Name;
}
}
}
return selectResults;
}
/// <summary>
/// 获取指定枚举的所有 Attribute 说明以及value组成的键值对
/// </summary>
/// <param name="type">对象类</param>
/// <param name="thirdAttributeType">第三个标签类型</param>
/// <param name="thirdAttributePropertieName">第三个标签类型取值名称</param>
/// <param name="getPropertie">false获取字段、true获取属性</param>
/// <returns></returns>
public static List<SelectResult> GetEnumAttributeListWithThirdValue(Type type, Type thirdAttributeType, string thirdAttributePropertieName, bool getPropertie = false)
{
if (type == null)
{
return null;
}
List<SelectResult> selectResults = new List<SelectResult>();
List<MemberInfo> memberInfos = new List<MemberInfo>();
if (getPropertie == false)
{
FieldInfo[] EnumInfo = type.GetFields();
if (null == EnumInfo || EnumInfo.Length <= 0)
{
return null;
}
memberInfos.AddRange(EnumInfo);
var parentFieldInfo = type.BaseType?.GetFields();
if (parentFieldInfo != null && parentFieldInfo.Length > 0)
{
memberInfos.AddRange(parentFieldInfo);
}
}
else
{
PropertyInfo[] EnumInfo = type.GetProperties();
if (null == EnumInfo || EnumInfo.Length <= 0)
{
return null;
}
memberInfos.AddRange(EnumInfo);
var parentPropertyInfo = type.BaseType?.GetProperties();
if (parentPropertyInfo != null && parentPropertyInfo.Length > 0)
{
memberInfos.AddRange(parentPropertyInfo);
}
}
foreach (var item in memberInfos)
{
var thirdAttributes = item.
GetCustomAttributes(thirdAttributeType, false);
if (thirdAttributes == null || thirdAttributes.Length <= 0)
{
continue;
}
DescriptionAttribute[] descriptionAttributes = (DescriptionAttribute[])item.
GetCustomAttributes(typeof(DescriptionAttribute), false);
dynamic infoObject = null;
if (getPropertie == false)
{
infoObject = (FieldInfo)item;
}
else
{
infoObject = (PropertyInfo)item;
}
if (descriptionAttributes.Length > 0)
{
SelectResult selectResult = new SelectResult()
{
Key = infoObject.Name,
Value = descriptionAttributes[0].Description,
};
selectResults.Add(selectResult);
}
DisplayAttribute[] displayAttributes = (DisplayAttribute[])item.
GetCustomAttributes(typeof(DisplayAttribute), false);
if (displayAttributes.Length > 0)
{
SelectResult selectResult = selectResults.FirstOrDefault(e => e.Key == infoObject.Name);
if (selectResult != null)
{
selectResult.SecondValue = displayAttributes[0].Name;
}
}
if (thirdAttributes.Length > 0 && !string.IsNullOrWhiteSpace(thirdAttributePropertieName))
{
foreach (var attr in thirdAttributes)
{
// 使用反射获取特性的属性值
var properties = thirdAttributeType.GetProperties();
foreach (var prop in properties)
{
// 假设你要获取特性的某个属性值，例如 TypeName
if (prop.Name == thirdAttributePropertieName)
{
object value = prop.GetValue(attr);
SelectResult selectResult = selectResults.FirstOrDefault(e => e.Key == infoObject.Name);
if (selectResult != null)
{
selectResult.ThirdValue = value?.ToString(); // 将属性值赋给 ThirdValue
}
break; // 如果找到了需要的属性，可以跳出循环
}
}
}
}
}
return selectResults;
}
/// <summary>
/// 获取指定类、指定常量值的Description说明
/// 包含直接继承的父级字段
/// </summary>
/// <param name="t">对象类</param>
/// <param name="getPropertie"></param>
/// <returns></returns>
public static List<Tuple<string, string, int>> GetTypeDescriptionListToTuple(Type t, bool getPropertie = false)
{
if (t == null)
{
return null;
}
List<MemberInfo> memberInfos = new List<MemberInfo>();
if (getPropertie == false)
{
FieldInfo[] fieldInfo = t.GetFields();
if (null == fieldInfo || fieldInfo.Length <= 0)
{
return null;
}
memberInfos.AddRange(fieldInfo);
var parentFieldInfo = t.BaseType?.GetFields();
if (parentFieldInfo != null && parentFieldInfo.Length > 0)
{
foreach (var item in parentFieldInfo)
{
if (!memberInfos.Any(d => d.Name == item.Name)) //如果子类已经包含父类的属性或者字段，跳过不处理
{
memberInfos.Add(item);
}
}
}
}
else
{
PropertyInfo[] fieldInfo = t.GetProperties();
if (null == fieldInfo || fieldInfo.Length <= 0)
{
return null;
}
memberInfos.AddRange(fieldInfo);
var parentPropertyInfo = t.BaseType?.GetProperties();
if (parentPropertyInfo != null && parentPropertyInfo.Length > 0)
{
foreach (var item in parentPropertyInfo)
{
if (!memberInfos.Any(d => d.Name == item.Name)) //如果子类已经包含父类的属性或者字段，跳过不处理
{
memberInfos.Add(item);
}
}
}
}
List<Tuple<string, string, int>> tuples = new List<Tuple<string, string, int>>();
foreach (var item in memberInfos)
{
DescriptionAttribute[] descriptionAttribute =
(DescriptionAttribute[])item.GetCustomAttributes(typeof(DescriptionAttribute), false);
NumericalOrderAttribute[] indexAttributes =
(NumericalOrderAttribute[])item.GetCustomAttributes(typeof(NumericalOrderAttribute), false);
if (descriptionAttribute.Length > 0 && indexAttributes.Length > 0)
{
Tuple<string, string, int> tuple = new Tuple<string, string, int>(item.Name,
descriptionAttribute[0].Description, indexAttributes[0].Index);
tuples.Add(tuple);
}
}
return tuples;
}
/// <summary>
/// 获取指定类、指定常量值的常量说明
/// </summary>
/// <param name="fieldName">常量字段名称</param>
///<param name="getPropertie">属性还是字段</param>
/// <returns></returns>
public static string GetTypeDescriptionName<T>(string fieldName, bool getPropertie = false)
{
if (string.IsNullOrEmpty(fieldName))
{
return "";
}
MemberInfo memberInfo = null;
if (getPropertie == false)
{
memberInfo = typeof(T).GetField(fieldName);
}
else
{
memberInfo = typeof(T).GetProperty(fieldName);
}
if (null == memberInfo)
{
return "";
}
DescriptionAttribute[] EnumAttributes =
(DescriptionAttribute[])memberInfo.GetCustomAttributes(typeof(DescriptionAttribute), false);
if (EnumAttributes.Length <= 0)
{
return "";
}
return EnumAttributes[0].Description;
}
/// <summary>
/// 获取指定命名空间下指定常量值的常量说明
/// </summary>
/// <param name="fieldName">常量字段名称</param>
/// <param name="assemblyName">命名空间，主要用来找到对应程序集</param>
///<param name="getPropertie">属性还是字段</param>
/// <returns></returns>
public static string GetTypeDescriptionName(string fieldName, string assemblyName, bool getPropertie = false)
{
if (string.IsNullOrWhiteSpace(fieldName) || string.IsNullOrWhiteSpace(assemblyName))
{
return null;
}
string desc = "";
foreach (var item in GetEnumList(assemblyName))
{
desc = GetTypeDescriptionName(item, fieldName, getPropertie);
if (!string.IsNullOrEmpty(desc))
{
break;
}
}
return desc;
}
/// <summary>
/// 获取指定类、指定常量值的常量说明
/// </summary>
/// <param name="t">对象类</param>
/// <param name="fieldName">常量字段名称</param>
///<param name="getPropertie">属性还是字段</param>
/// <returns></returns>
public static string GetTypeDescriptionName(this Type t, string fieldName, bool getPropertie = false)
{
if (string.IsNullOrWhiteSpace(fieldName))
{
return "";
}
MemberInfo memberInfo = null;
if (getPropertie == false)
{
memberInfo = t.GetField(fieldName);
}
else
{
memberInfo = t.GetProperty(fieldName);
}
if (null != memberInfo)
{
DescriptionAttribute[] EnumAttributes =
(DescriptionAttribute[])memberInfo.GetCustomAttributes(typeof(DescriptionAttribute), false);
if (EnumAttributes.Length > 0)
{
return EnumAttributes[0].Description;
}
}
return "";
}
/// <summary>
/// 扩展方法，获得枚举值集合
///</summary>
///<returns>枚举的DisplayName</returns>
public static List<T> GetEnumList<T>() where T : Enum
{
List<T> enumList = new List<T>();
foreach (T value in Enum.GetValues(typeof(T)))
{
enumList.Add(value);
}
return enumList;
}
private static List<Type> GetEnumList(string assemblyName)
{
if (!String.IsNullOrEmpty(assemblyName))
{
Assembly assembly = Assembly.Load(assemblyName);
List<Type> ts = assembly.GetTypes().Where(x => x.GetTypeInfo().IsClass).ToList();
return ts;
}
return new List<Type>();
}
/// <summary>
/// 扩展方法，获得枚举的Display值
///</summary>
///<param name="value">枚举值</param>
///<param name="nameInstead">当枚举值没有定义DisplayNameAttribute，是否使用枚举名代替，默认是使用</param>
///<param name="getPropertie">属性还是字段</param>
///<returns>枚举的DisplayName</returns>
public static string GetDisplayName(this Enum value, Boolean nameInstead = true, bool getPropertie = false)
{
Type type = value.GetType();
string name = Enum.GetName(type, value);
if (name == null)
{
return null;
}
DisplayAttribute attribute = null;
if (getPropertie == false)
{
attribute = Attribute.GetCustomAttribute(type.GetField(name), typeof(DisplayAttribute)) as DisplayAttribute;
}
else
{
attribute =
Attribute.GetCustomAttribute(type.GetProperty(name), typeof(DisplayAttribute)) as DisplayAttribute;
}
if (attribute == null && nameInstead == true)
{
return name;
}
return attribute == null ? null : attribute.Name;
}
/// <summary>
/// 获取枚举的描述信息
/// </summary>
/// <param name="value"></param>
/// <returns></returns>
public static string GetEnumDescription(this Enum value)
{
var name = value.ToString();
var field = value.GetType().GetField(name);
if (field == null) return name;
var att = Attribute.GetCustomAttribute(field, typeof(DescriptionAttribute), false);
return att == null ? field.Name : ((DescriptionAttribute)att).Description;
}
/// <summary>
/// 将传入的字符串中间部分字符替换成特殊字符
/// </summary>
/// <param name="value">需要替换的字符串</param>
/// <param name="startLen">前保留长度</param>
/// <param name="endLen">尾保留长度</param>
/// <param name="specialChar">特殊字符</param>
/// <returns>被特殊字符替换的字符串</returns>
public static string ReplaceWithSpecialChar(this string value, int startLen = 1, int endLen = 1,
char specialChar = '*')
{
if (string.IsNullOrEmpty(value))
{
return value;
}
try
{
if (value.Length <= startLen + endLen)
{
var temStartVal = value.Substring(0, startLen);
return $"{temStartVal}{"".PadLeft(endLen, specialChar)}";
}
if (value.Length == 10 && endLen == 1)
{
endLen = 3;
}
var startVal = value.Substring(0, startLen);
var endVal = value.Substring(value.Length - endLen);
if (value.Length == 2)
{
endVal = "";
}
value = $"{startVal}{endVal.PadLeft(value.Length - startLen, specialChar)}";
}
catch (Exception)
{
throw;
}
return value;
}
/// <summary>
/// Linux下字体名称转换
/// </summary>
/// <param name="fontValue"></param>
/// <returns></returns>
public static string GetLinuxFontFamily(this string fontValue)
{
if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
{
if (fontValue == "楷体")
{
fontValue = "KaiTi";
}
else if (fontValue == "隶书")
{
fontValue = "LiSu";
}
else if (fontValue == "宋体")
{
fontValue = "SimSun";
}
else if (fontValue == "微软雅黑")
{
fontValue = "Microsoft YaHei";
}
else if (fontValue == "新宋体")
{
fontValue = "NSimSun";
}
else if (fontValue == "仿宋")
{
fontValue = "FangSong";
}
else if (fontValue == "黑体")
{
fontValue = "SimHei";
}
}
return fontValue;
}
/// <summary>
/// 获取设备Id哈希值
/// </summary>
/// <param name="deviceId"></param>
/// <param name="TotalShards"></param>
/// <returns></returns>
public static int GetDeviceHashCode(string deviceId, int TotalShards = 100)
{
// 计算哈希分组ID
return Math.Abs(deviceId.GetHashCode() % TotalShards);
}
/// <summary>
/// 获取设备Id哈希分组
/// </summary>
/// <param name="deviceList"></param>
/// <returns></returns>
public static Dictionary<string, List<string>> GetDeviceHashGroup(List<string> deviceList)
{
Dictionary<string, List<string>> keyValuePairs = new Dictionary<string, List<string>>();
foreach (var deviceId in deviceList)
{
var hashCode = GetDeviceHashCode(deviceId);
if (!keyValuePairs.ContainsKey(hashCode.ToString()))
{
keyValuePairs.Add(hashCode.ToString(), new List<string>());
}
keyValuePairs[hashCode.ToString()].Add(deviceId);
}
return keyValuePairs;
}
}
}

28
src/JiShe.CollectBus.Common/Helpers/JsonHelper.cs
View File

@ -123,4 +123,32 @@ namespace JiShe.CollectBus.Common.Helpers
writer.WriteStringValue(value.ToString(_dateFormatString));
}
}
/// <summary>
/// Unix格式时间格式化
/// </summary>
public class UnixTimeConverter : JsonConverter<DateTime>
{
public override DateTime Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
{
if (reader.TokenType == JsonTokenType.String)
{
if (long.TryParse(reader.GetString(), out long timestamp))
return DateTimeOffset.FromUnixTimeSeconds(timestamp).DateTime;
}
if (reader.TokenType == JsonTokenType.Number)
{
long timestamp = reader.GetInt64();
return DateTimeOffset.FromUnixTimeSeconds(timestamp).DateTime;
}
return reader.GetDateTime();
}
public override void Write(Utf8JsonWriter writer, DateTime value, JsonSerializerOptions options)
{
long timestamp = new DateTimeOffset(value).ToUnixTimeSeconds();
writer.WriteStringValue(timestamp.ToString());
}
}
}

35
src/JiShe.CollectBus.Common/Helpers/SelectResult.cs Normal file
View File

@ -0,0 +1,35 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Common.Helpers
{
/// <summary>
/// 下拉框选项元素
/// </summary>
public class SelectResult
{
/// <summary>
/// 下拉框 键
/// </summary>
public string Key { get; set; }
/// <summary>
/// 下拉框 值
/// </summary>
public string Value { get; set; }
/// <summary>
/// 下拉框 值2
/// </summary>
public string SecondValue { get; set; }
/// <summary>
/// 下拉框 值3
/// </summary>
public object ThirdValue { get; set; }
}
}

36
src/JiShe.CollectBus.Domain/Ammeters/ElectricityMeter.cs Normal file
View File

@ -0,0 +1,36 @@
using JiShe.CollectBus.IoTDBProvider;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Ammeters
{
public class ElectricityMeter : IoTEntity
{
[ATTRIBUTEColumn]
public string MeterModel { get; set; }
/// <summary>
/// 下发消息内容
/// </summary>
[FIELDColumn]
public string IssuedMessageHexString { get; set; }
///// <summary>
///// 下发消息Id
///// </summary>
//[FIELDColumn]
//public string IssuedMessageId { get; set; }
[FIELDColumn]
public double Voltage { get; set; }
[FIELDColumn]
public double Current { get; set; }
[FIELDColumn]
public double Power => Voltage * Current;
}
}

6
src/JiShe.CollectBus.Domain/IotSystems/Devices/Device.cs
View File

@ -64,6 +64,12 @@ namespace JiShe.CollectBus.IotSystems.Devices
Status = DeviceStatus.Online;
}
public void UpdateByLoginAndHeartbeat()
{
LastOnlineTime = DateTime.Now;
Status = DeviceStatus.Online;
}
public void UpdateByOnClosed()
{
LastOfflineTime = DateTime.Now;

5
src/JiShe.CollectBus.Domain/IotSystems/MessageIssueds/ScheduledMeterReadingIssuedEventMessage.cs
View File

@ -29,5 +29,10 @@ namespace JiShe.CollectBus.IotSystems.MessageIssueds
/// </summary>
public string MessageId { get; set; }
/// <summary>
/// 最后一次消息Id，用于在消费消息时检查上一个任务是否处理完。
/// </summary>
public string LastMessageId { get; set; }
}
}

28
src/JiShe.CollectBus.Domain/IotSystems/MeterReadingRecords/MeterReadingRecords.cs
View File

@ -14,7 +14,6 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
/// </summary>
public class MeterReadingRecords : AggregateRoot<Guid>
{
/// <summary>
/// 是否手动操作
/// </summary>
@ -85,11 +84,15 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
/// </summary>
public int Pn { get; set; }
/// <summary>
/// 采集项编码
/// </summary>
public string ItemCode { get; set;}
/// <summary>
/// 是否下发成功
/// 是否超时
/// </summary>
public bool WasSuccessful { get; set; }
public bool IsTimeout { get; set; } = false;
/// <summary>
/// 创建时间
@ -112,24 +115,9 @@ namespace JiShe.CollectBus.IotSystems.MeterReadingRecords
public string ReceivedMessageId { get; set; }
/// <summary>
/// 数据迁移状态
/// 上报报文解析备注，异常情况下才有
/// </summary>
public RecordsDataMigrationStatusEnum MigrationStatus { get; set; }
/// <summary>
/// 数据结果,最终的解析报文结果值
/// </summary>
public string DataResult { get; set; }
/// <summary>
/// 数据时间,如冻结时间、事件发生事件等
/// </summary>
public DateTime? DataGenerationTimestamp { get; set; }
/// <summary>
/// 数据迁移时间
/// </summary>
public DateTime? MigrationTime { get; set; }
public string ReceivedRemark { get; set; }
public void CreateDataId(Guid Id)
{

1
src/JiShe.CollectBus.Domain/JiShe.CollectBus.Domain.csproj
View File

@ -24,6 +24,7 @@
<ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.Domain.Shared\JiShe.CollectBus.Domain.Shared.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.FreeSql\JiShe.CollectBus.FreeSql.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.IoTDBProvider\JiShe.CollectBus.IoTDBProvider.csproj" />
</ItemGroup>
<ItemGroup>

83
src/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs
View File

@ -20,6 +20,10 @@ using JiShe.CollectBus.Plugins;
using JiShe.CollectBus.Consumers;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using Confluent.Kafka;
using MassTransit.SqlTransport.Topology;
using Confluent.Kafka.Admin;
namespace JiShe.CollectBus.Host
@ -281,11 +285,18 @@ namespace JiShe.CollectBus.Host
/// </summary>
/// <param name="context">The context.</param>
/// <param name="configuration">The configuration.</param>
/// <summary>
/// Configures the mass transit.
/// </summary>
public void ConfigureMassTransit(ServiceConfigurationContext context, IConfiguration configuration)
{
context.Services.AddMassTransit(x =>
var consumerConfig = new ConsumerConfig { GroupId = ProtocolConst.SubscriberGroup };
var producerConfig = new ProducerConfig();
context.Services
.AddMassTransit(x =>
{
x.UsingInMemory();
x.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context));
x.AddConfigureEndpointsCallback((c, name, cfg) =>
{
@ -307,36 +318,88 @@ namespace JiShe.CollectBus.Host
.SetTimeLimitStart(BatchTimeLimitStart.FromLast)
.SetConcurrencyLimit(10));
});
rider.AddConsumer<ScheduledMeterReadingConsumer>();
rider.AddProducer<string, MessageReceivedLogin>(ProtocolConst.SubscriberLoginReceivedEventName);
rider.AddProducer<string, ReceivedHeartbeatConsumer>(ProtocolConst.SubscriberHeartbeatReceivedEventName);
rider.UsingKafka((c, cfg) =>
{
cfg.Host(configuration.GetConnectionString("Kafka"));
cfg.TopicEndpoint<MessageReceivedHeartbeat>(ProtocolConst.SubscriberReceivedHeartbeatEventName, ProtocolConst.SubscriberGroup, configurator =>
cfg.TopicEndpoint<MessageReceivedHeartbeat>(ProtocolConst.SubscriberHeartbeatReceivedEventName, consumerConfig, configurator =>
{
configurator.AutoOffsetReset = AutoOffsetReset.Earliest;
configurator.ConfigureConsumer<ReceivedHeartbeatConsumer>(c);
configurator.ConfigureConsumeTopology = false;
});
cfg.TopicEndpoint<MessageReceivedLogin>(ProtocolConst.SubscriberReceivedLoginEventName, ProtocolConst.SubscriberGroup, configurator =>
cfg.TopicEndpoint<MessageReceivedLogin>(ProtocolConst.SubscriberLoginReceivedEventName, consumerConfig, configurator =>
{
configurator.ConfigureConsumer<ReceivedLoginConsumer>(c);
configurator.ConfigureConsumeTopology = false;
configurator.AutoOffsetReset = AutoOffsetReset.Earliest;
});
cfg.TopicEndpoint<MessageReceived>(ProtocolConst.SubscriberReceivedEventName, ProtocolConst.SubscriberGroup, configurator =>
cfg.TopicEndpoint<MessageReceived>(ProtocolConst.SubscriberReceivedEventName, consumerConfig, configurator =>
{
configurator.ConfigureConsumer<ReceivedConsumer>(c);
configurator.ConfigureConsumeTopology = false;
configurator.AutoOffsetReset = AutoOffsetReset.Earliest;
});
cfg.TopicEndpoint<MessageReceived>(ProtocolConst.SubscriberIssuedEventName, ProtocolConst.SubscriberGroup, configurator =>
cfg.TopicEndpoint<MessageReceived>(ProtocolConst.SubscriberReceivedEventName, consumerConfig, configurator =>
{
configurator.ConfigureConsumer<IssuedConsumer>(c);
configurator.ConfigureConsumeTopology = false;
configurator.AutoOffsetReset = AutoOffsetReset.Earliest;
});
cfg.TopicEndpoint<ScheduledMeterReadingIssuedEventMessage>(ProtocolConst.AmmeterSubscriberWorkerFifteenMinuteIssuedEventName, consumerConfig, configurator =>
{
configurator.ConfigureConsumer<ScheduledMeterReadingConsumer>(c);
configurator.AutoOffsetReset = AutoOffsetReset.Earliest;
});
});
});
});
}
/// <summary>
/// 配置Kafka主题
/// </summary>
/// <param name="context"></param>
/// <param name="configuration"></param>
public void ConfigureKafkaTopic(ServiceConfigurationContext context, IConfiguration configuration)
{
var adminClient = new AdminClientBuilder(new AdminClientConfig
{
BootstrapServers = configuration.GetConnectionString("Kafka")
}).Build();
try
{
string serverTagName = configuration.GetSection("ServerTagName").Value!;
List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued(serverTagName);
topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived(serverTagName));
List<TopicSpecification> topicSpecifications = new List<TopicSpecification>();
foreach (var item in topics)
{
topicSpecifications.Add(new TopicSpecification
{
Name = item,
NumPartitions = 3,
ReplicationFactor = 1
});
}
adminClient.CreateTopicsAsync(topicSpecifications).ConfigureAwait(false).GetAwaiter().GetResult();
}
catch (CreateTopicsException e)
{
if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists)
{
throw;
}
}
}
}
}

1
src/JiShe.CollectBus.Host/CollectBusHostModule.cs
View File

@ -45,6 +45,7 @@ namespace JiShe.CollectBus.Host
ConfigureHangfire(context);
ConfigureCap(context, configuration);
//ConfigureMassTransit(context, configuration);
ConfigureKafkaTopic(context, configuration);
ConfigureAuditLog(context);
ConfigureCustom(context, configuration);
}

10
src/JiShe.CollectBus.Host/JiShe.CollectBus.Host.csproj
View File

@ -25,8 +25,8 @@
<PackageReference Include="DotNetCore.CAP.Kafka" Version="8.3.1" />
<PackageReference Include="DotNetCore.CAP.MongoDB" Version="8.3.1" />
<PackageReference Include="Hangfire.Redis.StackExchange" Version="1.9.4" />
<PackageReference Include="MassTransit" Version="8.3.2" />
<PackageReference Include="MassTransit.Kafka" Version="8.3.2" />
<PackageReference Include="MassTransit" Version="8.4.0" />
<PackageReference Include="MassTransit.Kafka" Version="8.4.0" />
<PackageReference Include="Microsoft.AspNetCore.DataProtection.StackExchangeRedis" Version="9.0.0" />
<PackageReference Include="Serilog" Version="4.1.0" />
<PackageReference Include="Serilog.AspNetCore" Version="8.0.3" />
@ -58,6 +58,12 @@
<ProjectReference Include="..\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
</ItemGroup>
<ItemGroup>
<Content Update="appsettings.json">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
</Content>
</ItemGroup>
<ItemGroup>
<None Update="Plugins\JiShe.CollectBus.Protocol.dll">
<CopyToOutputDirectory>Always</CopyToOutputDirectory>

59
src/JiShe.CollectBus.Host/appsettings.json
View File

@ -5,11 +5,11 @@
"Serilog.Sinks.File"
],
"MinimumLevel": {
"Default": "Warning",
"Default": "Information",
"Override": {
"Microsoft": "Information",
"Microsoft": "Warning",
"Volo.Abp": "Warning",
"Hangfire": "Information",
"Hangfire": "Warning",
"DotNetCore.CAP": "Warning",
"Serilog.AspNetCore": "Information",
"Microsoft.EntityFrameworkCore": "Warning",
@ -36,16 +36,13 @@
"ConnectionStrings": {
"Default": "mongodb://admin:admin02023@118.190.144.92:37117,118.190.144.92:37119,118.190.144.92:37120/JiSheCollectBus?authSource=admin&maxPoolSize=400&minPoolSize=10&waitQueueTimeoutMS=5000",
"Kafka": "121.42.242.91:29092,121.42.242.91:39092,121.42.242.91:49092",
//"Kafka": "8.148.227.21:9092,8.148.224.127:9092,8.138.38.208:9092",
"PrepayDB": "server=118.190.144.92;database=jishe.sysdb;uid=sa;pwd=admin@2023;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": "118.190.144.92:6379,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"Configuration": "127.0.0.1:6379,password=123456@qwer,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"Configuration": "120.24.52.151:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"DefaultDB": "14",
"HangfireDB": "15"
},
"Jwt": {
"Audience": "JiShe.CollectBus",
@ -84,41 +81,21 @@
"Port": 5672
}
},
//"Kafka": {
// "BootstrapServers": "121.42.242.91:29092,121.42.242.91:39092,121.42.242.91:49092",
// "EnableAuthorization": false,
// "SecurityProtocol": "SASL_PLAINTEXT",
// "SaslMechanism": "PLAIN",
// "SaslUserName": "lixiao",
// "SaslPassword": "lixiao1980",
// "Topic": {
// "ReplicationFactor": 3,
// "NumPartitions": 1000
// }
"Kafka": {
"Connections": {
"Default": {
"BootstrapServers": "121.42.242.91:29092,121.42.242.91:39092,121.42.242.91:49092"
// "SecurityProtocol": "SASL_PLAINTEXT",
// "SaslMechanism": "PLAIN",
// "SaslUserName": "lixiao",
// "SaslPassword": "lixiao1980",
}
"EnableAuthorization": false,
"SecurityProtocol": "SASL_PLAINTEXT",
"SaslMechanism": "PLAIN",
"SaslUserName": "lixiao",
"SaslPassword": "lixiao1980"
},
"Consumer": {
"GroupId": "JiShe.CollectBus"
"IoTDBOptions": {
"UserName": "root",
"Password": "root",
"ClusterList": [ "192.168.56.102:6667" ],
"PoolSize": 2,
"DataBaseName": "energy",
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
},
"Producer": {
"MessageTimeoutMs": 6000,
"Acks": -1
},
"Topic": {
"ReplicationFactor": 3,
"NumPartitions": 1000
},
"EventBus": {
"GroupId": "JiShe.CollectBus",
"TopicName": "DefaultTopicName"
}
}
"ServerTagName": "JiSheCollectBus"
}

16
src/JiShe.CollectBus.IoTDBProvider/Attribute/ATTRIBUTEColumnAttribute.cs Normal file
View File

@ -0,0 +1,16 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// Column分类标记特性（ATTRIBUTE字段）,也就是属性字段
/// </summary>
[AttributeUsage(AttributeTargets.Property)]
public class ATTRIBUTEColumnAttribute : Attribute
{
}
}

16
src/JiShe.CollectBus.IoTDBProvider/Attribute/FIELDColumnAttribute.cs Normal file
View File

@ -0,0 +1,16 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// Column分类标记特性（FIELD字段），数据列字段
/// </summary>
[AttributeUsage(AttributeTargets.Property)]
public class FIELDColumnAttribute : Attribute
{
}
}

16
src/JiShe.CollectBus.IoTDBProvider/Attribute/TAGColumnAttribute.cs Normal file
View File

@ -0,0 +1,16 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// Column分类标记特性（TAG字段），标签字段
/// </summary>
[AttributeUsage(AttributeTargets.Property)]
public class TAGColumnAttribute : Attribute
{
}
}

31
src/JiShe.CollectBus.IoTDBProvider/CollectBusIoTDBModule.cs Normal file
View File

@ -0,0 +1,31 @@
using JiShe.CollectBus.IoTDBProvider.Context;
using JiShe.CollectBus.IoTDBProvider.Interface;
using JiShe.CollectBus.IoTDBProvider.Provider;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Volo.Abp.Modularity;
namespace JiShe.CollectBus.IoTDBProvider
{
public class CollectBusIoTDBModule : AbpModule
{
public override void ConfigureServices(ServiceConfigurationContext context)
{
context.Services.Configure<IoTDBOptions>(context.Services.GetConfiguration().GetSection(nameof(IoTDBOptions)));
// 注册上下文为Scoped
context.Services.AddScoped<IoTDBRuntimeContext>();
// 注册Session工厂
context.Services.AddSingleton<IIoTDBSessionFactory, IoTDBSessionFactory>();
// 注册Provider
context.Services.AddScoped<IIoTDBProvider, IoTDBProvider>();
}
}
}

36
src/JiShe.CollectBus.IoTDBProvider/Context/IoTDBRuntimeContext.cs Normal file
View File

@ -0,0 +1,36 @@
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider.Context
{
/// <summary>
/// IoTDB SessionPool 运行时上下文
/// </summary>
public class IoTDBRuntimeContext
{
private readonly bool _defaultValue;
public IoTDBRuntimeContext(IOptions<IoTDBOptions> options)
{
_defaultValue = options.Value.UseTableSessionPoolByDefault;
UseTableSessionPool = _defaultValue;
}
/// <summary>
/// 是否使用表模型存储, 默认false，使用tree模型存储
/// </summary>
public bool UseTableSessionPool { get; set; }
/// <summary>
/// 重置为默认值
/// </summary>
public void ResetToDefault()
{
UseTableSessionPool = _defaultValue;
}
}
}

53
src/JiShe.CollectBus.IoTDBProvider/Interface/IIoTDBProvider.cs Normal file
View File

@ -0,0 +1,53 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// IoTDB数据源,数据库能同时存多个时序模型，但数据是完全隔离的，不能跨时序模型查询，通过连接字符串配置
/// </summary>
public interface IIoTDBProvider
{
/// <summary>
/// 切换 SessionPool
/// </summary>
/// <param name="useTableSession">是否使用表模型</param>
void SwitchSessionPool(bool useTableSession);
/// <summary>
/// 插入数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entity"></param>
/// <returns></returns>
Task InsertAsync<T>(T entity) where T : IoTEntity;
/// <summary>
/// 批量插入数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entities"></param>
/// <returns></returns>
Task BatchInsertAsync<T>(IEnumerable<T> entities) where T : IoTEntity;
/// <summary>
/// 删除数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
Task<object> DeleteAsync<T>(QueryOptions options) where T : IoTEntity;
/// <summary>
/// 查询数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
Task<PagedResult<T>> QueryAsync<T>(QueryOptions options) where T : IoTEntity, new();
}
}

16
src/JiShe.CollectBus.IoTDBProvider/Interface/IIoTDBSessionFactory.cs Normal file
View File

@ -0,0 +1,16 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider.Interface
{
/// <summary>
/// Session 工厂接口
/// </summary>
public interface IIoTDBSessionFactory
{
IIoTDBSessionPool GetSessionPool(bool useTableSession);
}
}

35
src/JiShe.CollectBus.IoTDBProvider/Interface/IIoTDBSessionPool.cs Normal file
View File

@ -0,0 +1,35 @@
using Apache.IoTDB.DataStructure;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider.Interface
{
/// <summary>
/// Session 连接池
/// </summary>
public interface IIoTDBSessionPool : IDisposable
{
/// <summary>
/// 打开连接池
/// </summary>
/// <returns></returns>
Task OpenAsync();
/// <summary>
/// 插入数据
/// </summary>
/// <param name="tablet"></param>
/// <returns></returns>
Task<int> InsertAsync(Tablet tablet);
/// <summary>
/// 查询数据
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
Task<SessionDataSet> ExecuteQueryStatementAsync(string sql);
}
}

16
src/JiShe.CollectBus.IoTDBProvider/JiShe.CollectBus.IoTDBProvider.csproj Normal file
View File

@ -0,0 +1,16 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<!--<PackageReference Include="Apache.IoTDB" Version="1.3.3.1" />-->
<PackageReference Include="Apache.IoTDB" Version="2.0.1" />
<PackageReference Include="Volo.Abp" Version="8.3.3" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
</ItemGroup>
</Project>

52
src/JiShe.CollectBus.IoTDBProvider/Options/IoTDBOptions.cs Normal file
View File

@ -0,0 +1,52 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// IOTDB配置
/// </summary>
public class IoTDBOptions
{
/// <summary>
/// 数据库名称，表模型才有，树模型为空
/// </summary>
public string DataBaseName { get; set; }
/// <summary>
/// 集群列表
/// </summary>
public List<string> ClusterList { get; set; }
/// <summary>
/// 用户名
/// </summary>
public string UserName { get; set; }
/// <summary>
/// 密码
/// </summary>
public string Password { get; set; }
/// <summary>
/// 连接池大小
/// </summary>
public int PoolSize { get; set; } = 2;
/// <summary>
/// 查询时，每次查询的数据量，默认1024
/// </summary>
public int FetchSize { get; set; } = 1024;
/// <summary>
/// 是否开启调试模式，生产环境请关闭，因为底层的实现方式，可能会导致内存持续增长。
/// </summary>
public bool OpenDebugMode { get; set;}
/// <summary>
/// 是否使用表模型存储, 默认false，使用tree模型存储
/// </summary>
public bool UseTableSessionPoolByDefault { get; set; } = false;
}
}

25
src/JiShe.CollectBus.IoTDBProvider/Options/PagedResult.cs Normal file
View File

@ -0,0 +1,25 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// 查询结果
/// </summary>
/// <typeparam name="T"></typeparam>
public class PagedResult<T>
{
/// <summary>
/// 总条数
/// </summary>
public int TotalCount { get; set; }
/// <summary>
/// 数据集合
/// </summary>
public IEnumerable<T> Items { get; set; }
}
}

27
src/JiShe.CollectBus.IoTDBProvider/Options/QueryCondition.cs Normal file
View File

@ -0,0 +1,27 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// 查询条件
/// </summary>
public class QueryCondition
{
/// <summary>
/// 字段
/// </summary>
public string Field { get; set; }
/// <summary>
/// 操作符
/// </summary>
public string Operator { get; set; }
/// <summary>
/// 值
/// </summary>
public object Value { get; set; }
}
}

34
src/JiShe.CollectBus.IoTDBProvider/Options/QueryOptions.cs Normal file
View File

@ -0,0 +1,34 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// 查询条件
/// </summary>
public class QueryOptions
{
/// <summary>
/// 表模型的表名称或者树模型的设备路径
/// </summary>
public required string TableNameOrTreePath { get; set; }
/// <summary>
/// 分页
/// </summary>
public int Page { get; set; }
/// <summary>
/// 分页大小
/// </summary>
public int PageSize { get; set; }
/// <summary>
/// 查询条件
/// </summary>
public List<QueryCondition> Conditions { get; } = new();
}
}

30
src/JiShe.CollectBus.IoTDBProvider/Provider/DeviceMetadata.cs Normal file
View File

@ -0,0 +1,30 @@
using Apache.IoTDB;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// 设备元数据
/// </summary>
public class DeviceMetadata
{
/// <summary>
/// 测量值集合，用于构建Table的测量值，也就是columnNames参数
/// </summary>
public List<string> ColumnNames { get; } = new();
/// <summary>
/// 列类型集合，用于构建Table的列类型，也就是columnCategories参数
/// </summary>
public List<ColumnCategory> ColumnCategories { get; } = new();
/// <summary>
/// 值类型集合，用于构建Table的值类型，也就是dataTypes参数
/// </summary>
public List<TSDataType>DataTypes { get; } = new();
}
}

39
src/JiShe.CollectBus.IoTDBProvider/Provider/DevicePathBuilder.cs Normal file
View File

@ -0,0 +1,39 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// 设备路径构建器
/// </summary>
public static class DevicePathBuilder
{
/// <summary>
/// 构建设备路径，由于路径的层级约束规范不能是纯数字字符，所以需要做特殊处理。
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entity"></param>
/// <returns></returns>
public static string GetDevicePath<T>(T entity) where T : IoTEntity
{
return $"root.{entity.SystemName.ToLower()}.`{entity.ProjectCode}`.`{entity.DeviceId}`";
}
/// <summary>
/// 获取表名称
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entity"></param>
/// <returns></returns>
public static string GetTableName<T>() where T : IoTEntity
{
var type = typeof(T);
return $"{type.Name.ToLower()}";
}
}
}

550
src/JiShe.CollectBus.IoTDBProvider/Provider/IoTDBProvider.cs Normal file
View File

@ -0,0 +1,550 @@
using Apache.IoTDB;
using Apache.IoTDB.DataStructure;
using JiShe.CollectBus.Common.Extensions;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.IoTDBProvider.Context;
using JiShe.CollectBus.IoTDBProvider.Interface;
using JiShe.CollectBus.IoTDBProvider.Provider;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using static Thrift.Protocol.Utilities.TJSONProtocolConstants;
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// IoTDB数据源
/// </summary>
public class IoTDBProvider : IIoTDBProvider
{
private IIoTDBSessionPool _currentSession;
private static readonly ConcurrentDictionary<Type, DeviceMetadata> _metadataCache = new();
private readonly ILogger<IoTDBProvider> _logger;
private readonly IIoTDBSessionFactory _sessionFactory;
private readonly IoTDBRuntimeContext _runtimeContext;
public IoTDBProvider(
ILogger<IoTDBProvider> logger,
IIoTDBSessionFactory sessionFactory,
IoTDBRuntimeContext runtimeContext)
{
_logger = logger;
_sessionFactory = sessionFactory;
_runtimeContext = runtimeContext;
RefreshSessionPool();
}
private void RefreshSessionPool()
{
try
{
_currentSession?.Dispose();
_currentSession = _sessionFactory.GetSessionPool(_runtimeContext.UseTableSessionPool);
_currentSession.OpenAsync().ConfigureAwait(false).GetAwaiter().GetResult();
}
catch (Exception ex)
{
//自动回退到备用SessionPool
_logger.LogError($"{nameof(RefreshSessionPool)} 切换SessionPool失败，回退到默认配置：{ex.Serialize()}");
_runtimeContext.UseTableSessionPool = !_runtimeContext.UseTableSessionPool;
_currentSession = _sessionFactory.GetSessionPool(_runtimeContext.UseTableSessionPool);
}
}
/// <summary>
/// 切换 SessionPool
/// </summary>
/// <param name="useTableSession"></param>
public void SwitchSessionPool(bool useTableSession)
{
if (_runtimeContext.UseTableSessionPool != useTableSession)
{
_runtimeContext.UseTableSessionPool = useTableSession;
RefreshSessionPool();
}
}
/// <summary>
/// 插入数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entity"></param>
/// <returns></returns>
public async Task InsertAsync<T>(T entity) where T : IoTEntity
{
var metadata = GetMetadata<T>();
var tablet = BuildTablet(new[] { entity }, metadata);
int result = await _currentSession.InsertAsync(tablet);
if (result <= 0)
{
_logger.LogError($"{typeof(T).Name}插入数据没有成功");
}
}
/// <summary>
/// 批量插入数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
public async Task BatchInsertAsync<T>(IEnumerable<T> entities) where T : IoTEntity
{
var metadata = GetMetadata<T>();
var batchSize = 1000;
var batches = entities.Chunk(batchSize);
foreach (var batch in batches)
{
var tablet = BuildTablet(batch, metadata);
var result = await _currentSession.InsertAsync(tablet);
if (result <= 0)
{
_logger.LogWarning($"{typeof(T).Name} 批量插入数据第{batch}批次没有成功，共{batches}批次。");
}
}
}
/// <summary>
/// 删除数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
public async Task<object> DeleteAsync<T>(QueryOptions options) where T : IoTEntity
{
var query = BuildDeleteSQL<T>(options);
var sessionDataSet = await _currentSession.ExecuteQueryStatementAsync(query);
if (!sessionDataSet.HasNext())
{
_logger.LogWarning($"{typeof(T).Name} 删除数据时，没有返回受影响记录数量。");
return 0;
}
//获取唯一结果行
var row = sessionDataSet.Next();
return row.Values[0];
}
/// <summary>
/// 查询数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
public async Task<PagedResult<T>> QueryAsync<T>(QueryOptions options) where T : IoTEntity, new()
{
var query = BuildQuerySQL<T>(options);
var sessionDataSet = await _currentSession.ExecuteQueryStatementAsync(query);
var result = new PagedResult<T>
{
TotalCount = await GetTotalCount<T>(options),
Items = ParseResults<T>(sessionDataSet, options.PageSize)
};
return result;
}
/// <summary>
/// 构建Tablet
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="entities">表实体</param>
/// <param name="metadata">设备元数据</param></param>
/// <returns></returns>
private Tablet BuildTablet<T>(IEnumerable<T> entities, DeviceMetadata metadata) where T : IoTEntity
{
var timestamps = new List<long>();
var values = new List<List<object>>();
var devicePaths = new HashSet<string>();
foreach (var entity in entities)
{
timestamps.Add(entity.Timestamps);
var rowValues = new List<object>();
foreach (var measurement in metadata.ColumnNames)
{
PropertyInfo propertyInfo = typeof(T).GetProperty(measurement);
if (propertyInfo==null)
{
throw new Exception($"{nameof(BuildTablet)} 构建表模型{typeof(T).Name}时，没有找到{measurement}属性，属于异常情况。");
}
var value = propertyInfo.GetValue(entity);
if (value != null)
{
rowValues.Add(value);
}
else
{
DataTypeValueMap.TryGetValue(propertyInfo.PropertyType.Name, out object defaultValue);
rowValues.Add(defaultValue);
}
}
values.Add(rowValues);
if (!_runtimeContext.UseTableSessionPool)//树模型
{
devicePaths.Add(DevicePathBuilder.GetDevicePath(entity));
}
else
{
devicePaths.Add(DevicePathBuilder.GetTableName<T>());
}
}
if (devicePaths.Count > 1)
{
throw new Exception($"{nameof(BuildTablet)} 构建Tablet《{typeof(T).Name}》时，批量插入的设备路径不一致。");
}
return _runtimeContext.UseTableSessionPool
? BuildTableSessionTablet(metadata, devicePaths.First(), values, timestamps)
: BuildSessionTablet(metadata, devicePaths.First(), values, timestamps);
}
/// <summary>
/// 构建tree模型的Tablet
/// </summary>
/// <param name="metadata"></param>
/// <param name="devicePath"></param>
/// <param name="values"></param>
/// <param name="timestamps"></param>
/// <returns></returns>
private Tablet BuildSessionTablet(DeviceMetadata metadata, string devicePath,
List<List<object>> values, List<long> timestamps)
{
return new Tablet(
devicePath,
metadata.ColumnNames,
metadata.DataTypes,
values,
timestamps
);
}
/// <summary>
/// 构建表模型的Tablet
/// </summary>
/// <param name="metadata"></param>
/// <param name="devicePath"></param>
/// <param name="values"></param>
/// <param name="timestamps"></param>
/// <returns></returns>
private Tablet BuildTableSessionTablet(DeviceMetadata metadata, string devicePath,
List<List<object>> values, List<long> timestamps)
{
var tablet = new Tablet(
devicePath,
metadata.ColumnNames,
metadata.ColumnCategories,
metadata.DataTypes,
values,
timestamps
);
return tablet;
}
/// <summary>
/// 构建查询语句
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
private string BuildQuerySQL<T>(QueryOptions options) where T : IoTEntity
{
var metadata = GetMetadata<T>();
var sb = new StringBuilder("SELECT ");
sb.AppendJoin(", ", metadata.ColumnNames);
sb.Append($" FROM {options.TableNameOrTreePath}");
if (options.Conditions.Any())
{
sb.Append(" WHERE ");
sb.AppendJoin(" AND ", options.Conditions.Select(TranslateCondition));
}
sb.Append($" LIMIT {options.PageSize} OFFSET {options.Page * options.PageSize}");
return sb.ToString();
}
/// <summary>
/// 构建删除语句
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
private string BuildDeleteSQL<T>(QueryOptions options) where T : IoTEntity
{
var metadata = GetMetadata<T>();
var sb = new StringBuilder();
if (!_runtimeContext.UseTableSessionPool)
{
sb.Append("DELETE ");
}
else
{
sb.Append("DROP ");
}
sb.Append($" FROM {options.TableNameOrTreePath}");
sb.AppendJoin(", ", metadata.ColumnNames);
if (options.Conditions.Any())
{
sb.Append(" WHERE ");
sb.AppendJoin(" AND ", options.Conditions.Select(TranslateCondition));
}
return sb.ToString();
}
/// <summary>
/// 将查询条件转换为SQL语句
/// </summary>
/// <param name="condition"></param>
/// <returns></returns>
/// <exception cref="NotSupportedException"></exception>
private string TranslateCondition(QueryCondition condition)
{
return condition.Operator switch
{
">" => $"{condition.Field} > {condition.Value}",
"<" => $"{condition.Field} < {condition.Value}",
"=" => $"{condition.Field} = '{condition.Value}'",
_ => throw new NotSupportedException($"Operator {condition.Operator} not supported")
};
}
/// <summary>
/// 获取查询条件的总数量
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="options"></param>
/// <returns></returns>
private async Task<int> GetTotalCount<T>(QueryOptions options) where T : IoTEntity
{
var countQuery = $"SELECT COUNT(*) FROM {options.TableNameOrTreePath}";
if (options.Conditions.Any())
{
countQuery += " WHERE " + string.Join(" AND ", options.Conditions.Select(TranslateCondition));
}
var result = await _currentSession.ExecuteQueryStatementAsync(countQuery);
return result.HasNext() ? Convert.ToInt32(result.Next().Values[0]) : 0;
}
/// <summary>
/// 解析查询结果
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="dataSet"></param>
/// <param name="pageSize"></param>
/// <returns></returns>
private IEnumerable<T> ParseResults<T>(SessionDataSet dataSet, int pageSize) where T : IoTEntity, new()
{
var results = new List<T>();
var metadata = GetMetadata<T>();
var properties = typeof(T).GetProperties();
while (dataSet.HasNext() && results.Count < pageSize)
{
var record = dataSet.Next();
var entity = new T
{
Timestamps = record.Timestamps
};
foreach (var measurement in metadata.ColumnNames)
{
var value = record.Values;
var prop = properties.FirstOrDefault(p =>
p.Name.Equals(measurement, StringComparison.OrdinalIgnoreCase));
if (prop != null)
{
typeof(T).GetProperty(measurement)?.SetValue(entity, value);
}
}
results.Add(entity);
}
return results;
}
/// <summary>
/// 获取设备元数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
private DeviceMetadata GetMetadata<T>() where T : IoTEntity
{
return _metadataCache.GetOrAdd(typeof(T), type =>
{
var columns = CollectColumnMetadata(type);
var metadata = BuildDeviceMetadata(columns);
return metadata;
});
}
/// <summary>
/// 获取设备元数据的列
/// </summary>
/// <param name="type"></param>
/// <returns></returns>
private List<ColumnInfo> CollectColumnMetadata(Type type)
{
var columns = new List<ColumnInfo>();
foreach (var prop in type.GetProperties())
{
//按优先级顺序检查属性，避免重复反射
ColumnInfo? column = prop.GetCustomAttribute<TAGColumnAttribute>() is not null ? new ColumnInfo(
name: prop.Name, //使用属性名
category: ColumnCategory.TAG,
dataType: GetDataTypeFromTypeName(prop.PropertyType.Name)
) : prop.GetCustomAttribute<ATTRIBUTEColumnAttribute>() is not null ? new ColumnInfo(
prop.Name,
ColumnCategory.ATTRIBUTE,
GetDataTypeFromTypeName(prop.PropertyType.Name)
) : prop.GetCustomAttribute<FIELDColumnAttribute>() is not null ? new ColumnInfo(
prop.Name,
ColumnCategory.FIELD,
GetDataTypeFromTypeName(prop.PropertyType.Name)
) : null;
if (column.HasValue)
{
columns.Add(column.Value);
}
}
return columns;
}
/// <summary>
/// 构建设备元数据
/// </summary>
/// <param name="columns"></param>
/// <returns></returns>
private DeviceMetadata BuildDeviceMetadata(List<ColumnInfo> columns)
{
var metadata = new DeviceMetadata();
//按业务逻辑顺序处理（TAG -> FIELD -> ATTRIBUTE）
var groupedColumns = columns
.GroupBy(c => c.Category)
.ToDictionary(g => g.Key, g => g.ToList());
ProcessCategory(groupedColumns, ColumnCategory.TAG, metadata);
ProcessCategory(groupedColumns, ColumnCategory.ATTRIBUTE, metadata);
ProcessCategory(groupedColumns, ColumnCategory.FIELD, metadata);
return metadata;
}
/// <summary>
/// 处理不同列类型的逻辑
/// </summary>
/// <param name="groupedColumns"></param>
/// <param name="category"></param>
/// <param name="metadata"></param>
private void ProcessCategory(IReadOnlyDictionary<ColumnCategory, List<ColumnInfo>> groupedColumns, ColumnCategory category, DeviceMetadata metadata)
{
if (groupedColumns.TryGetValue(category, out var cols))
{
metadata.ColumnNames.AddRange(cols.Select(c => c.Name));
metadata.ColumnCategories.AddRange(cols.Select(c => c.Category));
metadata.DataTypes.AddRange(cols.Select(c => c.DataType));
}
}
/// <summary>
/// 数据列结构
/// </summary>
private readonly struct ColumnInfo
{
public string Name { get; }
public ColumnCategory Category { get; }
public TSDataType DataType { get; }
public ColumnInfo(string name, ColumnCategory category, TSDataType dataType)
{
Name = name;
Category = category;
DataType = dataType;
}
}
/// <summary>
/// 根据类型名称获取对应的 IoTDB 数据类型
/// </summary>
/// <param name="typeName">类型名称（不区分大小写）</param>
/// <returns>对应的 TSDataType，默认返回 TSDataType.STRING</returns>
private TSDataType GetDataTypeFromTypeName(string typeName)
{
if (string.IsNullOrWhiteSpace(typeName))
return TSDataType.STRING;
return DataTypeMap.TryGetValue(typeName.Trim(), out var dataType)
? dataType
: TSDataType.STRING;
}
/// <summary>
/// 根据类型名称获取 IoTDB 数据类型
/// </summary>
private readonly IReadOnlyDictionary<string, TSDataType> DataTypeMap =
new Dictionary<string, TSDataType>(StringComparer.OrdinalIgnoreCase)
{
["BOOLEAN"] = TSDataType.BOOLEAN,
["INT32"] = TSDataType.INT32,
["INT64"] = TSDataType.INT64,
["FLOAT"] = TSDataType.FLOAT,
["DOUBLE"] = TSDataType.DOUBLE,
["TEXT"] = TSDataType.TEXT,
["NULLTYPE"] = TSDataType.NONE,
["TIMESTAMP"] = TSDataType.TIMESTAMP,
["DATE"] = TSDataType.DATE,
["BLOB"] = TSDataType.BLOB,
["DECIMAL"] = TSDataType.STRING,
["STRING"] = TSDataType.STRING
};
/// <summary>
/// 根据类型名称获取 IoTDB 数据默认值
/// </summary>
private readonly IReadOnlyDictionary<string, object> DataTypeValueMap =
new Dictionary<string, object>(StringComparer.OrdinalIgnoreCase)
{
["BOOLEAN"] = false,
["INT32"] = 0,
["INT64"] = 0,
["FLOAT"] = 0.0f,
["DOUBLE"] = 0.0d,
["TEXT"] = string.Empty,
["NULLTYPE"] = null,
["TIMESTAMP"] = null,
["DATE"] = null,
["BLOB"] = null,
["DECIMAL"] = "0.0",
["STRING"] = string.Empty
};
}
}

36
src/JiShe.CollectBus.IoTDBProvider/Provider/IoTDBSessionFactory.cs Normal file
View File

@ -0,0 +1,36 @@
using JiShe.CollectBus.IoTDBProvider.Interface;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.IoTDBProvider.Provider
{
/// <summary>
/// 实现带缓存的Session工厂
/// </summary>
public class IoTDBSessionFactory : IIoTDBSessionFactory
{
private readonly IoTDBOptions _options;
private readonly ConcurrentDictionary<bool, IIoTDBSessionPool> _pools = new();
public IoTDBSessionFactory(IOptions<IoTDBOptions> options)
{
_options = options.Value;
}
public IIoTDBSessionPool GetSessionPool(bool useTableSession)
{
return _pools.GetOrAdd(useTableSession, key =>
{
return key
? new TableSessionPoolAdapter(_options)
: new SessionPoolAdapter(_options);
});
}
}
}

37
src/JiShe.CollectBus.IoTDBProvider/Provider/IoTEntity.cs Normal file
View File

@ -0,0 +1,37 @@
namespace JiShe.CollectBus.IoTDBProvider
{
/// <summary>
/// IoT实体基类
/// </summary>
public abstract class IoTEntity
{
/// <summary>
/// 系统名称
/// </summary>
[TAGColumn]
public string SystemName { get; set; }
/// <summary>
/// 项目编码
/// </summary>
[TAGColumn]
public string ProjectCode { get; set; }
/// <summary>
/// 设备类型集中器、电表、水表、流量计、传感器等
/// </summary>
[TAGColumn]
public string DeviceType { get; set; }
/// <summary>
/// 设备ID
/// </summary>
[TAGColumn]
public string DeviceId { get; set; }
/// <summary>
/// 当前时间戳，单位毫秒
/// </summary>
public long Timestamps { get; set; } = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
}

74
src/JiShe.CollectBus.IoTDBProvider/Provider/SessionPoolAdapter.cs Normal file
View File

@ -0,0 +1,74 @@
using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using JiShe.CollectBus.IoTDBProvider.Interface;
using Microsoft.Extensions.Logging;
namespace JiShe.CollectBus.IoTDBProvider.Provider
{
/// <summary>
/// 树模型连接池
/// </summary>
public class SessionPoolAdapter : IIoTDBSessionPool
{
private readonly SessionPool _sessionPool;
private readonly IoTDBOptions _options;
public SessionPoolAdapter(IoTDBOptions options)
{
_options = options;
_sessionPool = new SessionPool.Builder()
.SetNodeUrl(options.ClusterList)
.SetUsername(options.UserName)
.SetPassword(options.Password)
.SetFetchSize(options.FetchSize)
.SetPoolSize(options.PoolSize)
.Build();
}
/// <summary>
/// 打开连接池
/// </summary>
/// <returns></returns>
public async Task OpenAsync()
{
await _sessionPool.Open(false);
if (_options.OpenDebugMode)
{
_sessionPool.OpenDebugMode(builder =>
{
builder.AddConsole();
});
}
}
/// <summary>
/// 批量插入对齐时间序列数据
/// </summary>
/// <param name="tablet"></param>
/// <returns></returns>
public async Task<int> InsertAsync(Tablet tablet)
{
return await _sessionPool.InsertAlignedTabletAsync(tablet);
}
/// <summary>
/// 查询数据
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
return await _sessionPool.ExecuteQueryStatementAsync(sql);
}
public void Dispose()
{
_sessionPool?.Close().ConfigureAwait(false).GetAwaiter().GetResult();
}
}
}

72
src/JiShe.CollectBus.IoTDBProvider/Provider/TableSessionPoolAdapter.cs Normal file
View File

@ -0,0 +1,72 @@
using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using JiShe.CollectBus.IoTDBProvider.Interface;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
namespace JiShe.CollectBus.IoTDBProvider.Provider
{
/// <summary>
/// 表模型Session连接池
/// </summary>
public class TableSessionPoolAdapter : IIoTDBSessionPool
{
private readonly TableSessionPool _sessionPool;
private readonly IoTDBOptions _options;
public TableSessionPoolAdapter(IoTDBOptions options)
{
_options = options;
_sessionPool = new TableSessionPool.Builder()
.SetNodeUrls(options.ClusterList)
.SetUsername(options.UserName)
.SetPassword(options.Password)
.SetFetchSize(options.FetchSize)
.SetPoolSize(options.PoolSize)
.SetDatabase(options.DataBaseName)
.Build();
}
/// <summary>
/// 打开连接池
/// </summary>
/// <returns></returns>
public async Task OpenAsync()
{
await _sessionPool.Open(false);
if (_options.OpenDebugMode)
{
_sessionPool.OpenDebugMode(builder => builder.AddConsole());
}
}
/// <summary>
/// 批量插入
/// </summary>
/// <param name="tablet"></param>
/// <returns></returns>
public async Task<int> InsertAsync(Tablet tablet)
{
return await _sessionPool.InsertAsync(tablet);
}
/// <summary>
/// 查询数据
/// </summary>
/// <param name="sql"></param>
/// <returns></returns>
public async Task<SessionDataSet> ExecuteQueryStatementAsync(string sql)
{
return await _sessionPool.ExecuteQueryStatementAsync(sql);
}
public void Dispose()
{
_sessionPool?.Close().ConfigureAwait(false).GetAwaiter().GetResult();
}
}
}

19
src/JiShe.CollectBus.MongoDB/Repository/MeterReadingRecord/MeterReadingRecordRepository.cs
View File

@ -109,6 +109,21 @@ namespace JiShe.CollectBus.Repository.MeterReadingRecord
return entity;
}
/// <summary>
/// 单个获取
/// </summary>
/// <param name="entity"></param>
/// <param name="dateTime"></param>
/// <returns></returns>
/// <exception cref="NotImplementedException"></exception>
public async Task<MeterReadingRecords> FirOrDefaultAsync(MeterReadingRecords entity, DateTime? dateTime)
{
var collection = await GetShardedCollection(dateTime);
//await collection.findon
throw new NotImplementedException();
}
/// <summary>
/// 多集合数据查询
/// </summary>
@ -156,9 +171,5 @@ namespace JiShe.CollectBus.Repository.MeterReadingRecord
return database.GetCollection<MeterReadingRecords>(collectionName);
}
public Task<MeterReadingRecords> FirOrDefaultAsync(MeterReadingRecords entity, DateTime? dateTime)
{
throw new NotImplementedException();
}
}
}

9
src/JiShe.CollectBus.MongoDB/ShardingStrategy/DayShardingStrategy.cs
View File

@ -1,4 +1,5 @@
using System;
using JiShe.CollectBus.Common.Extensions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
@ -21,7 +22,7 @@ namespace JiShe.CollectBus.ShardingStrategy
public string GetCollectionName(DateTime dateTime)
{
var baseName = typeof(TEntity).Name;
return $"{baseName}_{dateTime:yyyyMMddHHmm}";
return $"{baseName}_{dateTime.GetDataTableShardingStrategy()}";
}
/// <summary>
@ -31,7 +32,7 @@ namespace JiShe.CollectBus.ShardingStrategy
public string GetCurrentCollectionName()
{
var baseName = typeof(TEntity).Name;
return $"{baseName}_{DateTime.Now:yyyyMMddHHmm}";
return $"{baseName}_{DateTime.Now.GetDataTableShardingStrategy()}";
}
/// <summary>
@ -49,7 +50,7 @@ namespace JiShe.CollectBus.ShardingStrategy
while (current <= end)
{
months.Add($"{baseName}_{current:yyyyMMddHHmm}");
months.Add($"{baseName}_{current.GetDataTableShardingStrategy()}");
current = current.AddMonths(1);
}

22
src/JiShe.CollectBus.Protocol.Contracts/Abstracts/BaseProtocolPlugin.cs
View File

@ -11,12 +11,13 @@ using JiShe.CollectBus.Protocol.Contracts.AnalysisData;
using Microsoft.Extensions.DependencyInjection;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.Protocols;
using MassTransit;
namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
{
public abstract class BaseProtocolPlugin : IProtocolPlugin
{
private readonly ICapPublisher _capBus;
private readonly ICapPublisher _producerBus;
private readonly ILogger<BaseProtocolPlugin> _logger;
private readonly IRepository<ProtocolInfo, Guid> _protocolInfoRepository;
@ -36,7 +37,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
_logger = serviceProvider.GetRequiredService<ILogger<BaseProtocolPlugin>>();
_protocolInfoRepository = serviceProvider.GetRequiredService<IRepository<ProtocolInfo, Guid>>();
_capBus = serviceProvider.GetRequiredService<ICapPublisher>();
_producerBus = serviceProvider.GetRequiredService<ICapPublisher>();
}
public abstract ProtocolInfo Info { get; }
@ -55,7 +56,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
//await _protocolInfoCache.Get()
}
public abstract Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) where T : TB3761FN;
public abstract Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) where T : TB3761;
/// <summary>
/// 登录帧解析
@ -86,7 +87,8 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
};
var bytes = Build3761SendData.BuildSendCommandBytes(reqParam);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
await _producerBus.PublishAsync(ProtocolConst.SubscriberLoginIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
//await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Login, MessageId = messageReceived.MessageId });
}
/// <summary>
@ -124,7 +126,9 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
Fn = 1
};
var bytes = Build3761SendData.BuildSendCommandBytes(reqParam);
await _capBus.PublishAsync(ProtocolConst.SubscriberIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
await _producerBus.PublishAsync(ProtocolConst.SubscriberHeartbeatIssuedEventName, new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
//await _producerBus.Publish(new IssuedEventMessage { ClientId = messageReceived.ClientId, DeviceNo = messageReceived.DeviceNo, Message = bytes, Type = IssuedEventType.Heartbeat, MessageId = messageReceived.MessageId });
}
}
@ -602,7 +606,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
/// <param name="messageReceived"></param>
/// <param name="sendAction"></param>
/// <returns></returns>
public virtual TB3761FN AnalyzeReadingDataAsync(MessageReceived messageReceived,
public virtual TB3761 AnalyzeReadingDataAsync(MessageReceived messageReceived,
Action<byte[]>? sendAction = null)
{
var hexStringList = messageReceived.MessageHexString.StringToPairs();
@ -664,7 +668,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
}
}
return tb3761Fn;
return tb3761;
}
/// <summary>
@ -673,7 +677,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
/// <param name="messageReceived"></param>
/// <param name="sendAction"></param>
/// <returns></returns>
public virtual TB3761FN AnalyzeReadingTdcDataAsync(MessageReceived messageReceived,
public virtual TB3761 AnalyzeReadingTdcDataAsync(MessageReceived messageReceived,
Action<byte[]>? sendAction = null)
{
@ -717,7 +721,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Abstracts
}
}
return tb3761Fn;
return tb3761;
//var freezeDensity = (FreezeDensity)Convert.ToInt32(hexDatas.Skip(5).Take(1));
//var addMinute = 0;
//switch (freezeDensity)

63
src/JiShe.CollectBus.Protocol.Contracts/Extensions/ProtocolConstExtensions.cs Normal file
View File

@ -0,0 +1,63 @@
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Extensions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Protocol.Contracts
{
public class ProtocolConstExtensions
{
/// <summary>
/// 自动获取 ProtocolConst 类中所有下行 Kafka 主题名称
/// （通过反射筛选 public const string 且字段名以 "EventName" 结尾的常量）
/// </summary>
public static List<string> GetAllTopicNamesByIssued(string serverTagName)
{
List<string> topics = typeof(ProtocolConst)
.GetFields(BindingFlags.Public | BindingFlags.Static)
.Where(f =>
f.IsLiteral &&
!f.IsInitOnly &&
f.FieldType == typeof(string) &&
f.Name.EndsWith("IssuedEventName")) // 通过命名规则过滤主题字段
//.Select(f => $"{serverTagName}.{(string)f.GetRawConstantValue()!}")
.Select(f => (string)f.GetRawConstantValue()!)
.ToList();
return topics;
}
/// <summary>
/// 自动获取 ProtocolConst 类中所有下行 Kafka 主题名称
/// （通过反射筛选 public const string 且字段名以 "EventName" 结尾的常量）
/// </summary>
public static List<string> GetAllTopicNamesByReceived(string serverTagName)
{
//固定的上报主题
var topicList = typeof(ProtocolConst)
.GetFields(BindingFlags.Public | BindingFlags.Static)
.Where(f =>
f.IsLiteral &&
!f.IsInitOnly &&
f.FieldType == typeof(string) &&
f.Name.EndsWith("ReceivedEventName")) // 通过命名规则过滤主题字段
.Select(f => (string)f.GetRawConstantValue()!)
.ToList();
//动态上报主题，需根据协议的AFN功能码动态获取
var afnList = EnumExtensions.ToNameValueDictionary<AFN>();
foreach (var item in afnList)
{
topicList.Add(string.Format(ProtocolConst.AFNTopicNameFormat, item.Value.ToString().PadLeft(2, '0')));
}
//return topicList.Select(f => $"{serverTagName}.{f}").ToList();
return topicList;
}
}
}

2
src/JiShe.CollectBus.Protocol.Contracts/Interfaces/IProtocolPlugin.cs
View File

@ -14,7 +14,7 @@ namespace JiShe.CollectBus.Protocol.Contracts.Interfaces
Task AddAsync();
Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) where T : TB3761FN;
Task<T> AnalyzeAsync<T>(MessageReceived messageReceived, Action<byte[]>? sendAction = null) where T : TB3761;
Task LoginAsync(MessageReceivedLogin messageReceived);

74
src/JiShe.CollectBus.Protocol.Contracts/ProtocolConst.cs
View File

@ -9,62 +9,104 @@ namespace JiShe.CollectBus.Protocol.Contracts
public class ProtocolConst
{
public const string SubscriberGroup = "jishe.collectbus";
public const string SubscriberIssuedEventName = "issued.event";
/// <summary>
/// 心跳下行消息主题
/// </summary>
public const string SubscriberHeartbeatIssuedEventName = "issued.heartbeat.event";
/// <summary>
/// 登录下行消息主题
/// </summary>
public const string SubscriberLoginIssuedEventName = "issued.login.event";
/// <summary>
/// 上行消息主题
/// </summary>
public const string SubscriberReceivedEventName = "received.event";
public const string SubscriberReceivedHeartbeatEventName = "received.heartbeat.event";
public const string SubscriberReceivedLoginEventName = "received.login.event";
/// <summary>
/// 心跳上行消息主题
/// </summary>
public const string SubscriberHeartbeatReceivedEventName = "received.heartbeat.event";
/// <summary>
/// 登录上行消息主题
/// </summary>
public const string SubscriberLoginReceivedEventName = "received.login.event";
#region 电表消息主题
/// <summary>
/// 1分钟采集电表数据下行消息主题
/// </summary>
public const string AmmeterSubscriberWorkerOneMinuteIssuedEventName = "issued.one.ammeter.event";
public const string AmmeterSubscriberWorkerOneMinuteIssuedEventName = "issued.auto.one.ammeter.event";
/// <summary>
/// 5分钟采集电表数据下行消息主题
/// </summary>
public const string AmmeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.five.ammeter.event";
public const string AmmeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.auto.five.ammeter.event";
/// <summary>
/// 15分钟采集电表数据下行消息主题
/// </summary>
public const string AmmeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.ammeter.event";
public const string AmmeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.auto.fifteen.ammeter.event";
/// <summary>
/// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号、定时阀控等
/// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号控等
/// </summary>
public const string AmmeterSubscriberWorkerOtherIssuedEventName = "issued.other.ammeter.event";
public const string AmmeterSubscriberWorkerOtherIssuedEventName = "issued.auto.other.ammeter.event";
/// <summary>
/// 电表自动阀控
/// </summary>
public const string AmmeterSubscriberWorkerAutoValveControlIssuedEventName = "issued.auto.control.ammeter.event";
/// <summary>
/// 电表手动阀控
/// </summary>
public const string AmmeterSubscriberWorkerManualValveControlIssuedEventName = "issued.control.ammeter.event";
public const string AmmeterSubscriberWorkerManualValveControlIssuedEventName = "issued.manual.control.ammeter.event";
/// <summary>
/// 电表手动抄读
/// </summary>
public const string AmmeterSubscriberWorkerManualValveReadingIssuedEventName = "issued.manual.reading.ammeter.event";
#endregion
#region 水表消息主题
/// <summary>
/// 1分钟采集水表数据下行消息主题
/// </summary>
public const string WatermeterSubscriberWorkerOneMinuteIssuedEventName = "issued.one.watermeter.event";
public const string WatermeterSubscriberWorkerOneMinuteIssuedEventName = "issued.auto.one.watermeter.event";
/// <summary>
/// 5分钟采集水表数据下行消息主题
/// </summary>
public const string WatermeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.five.watermeter.event";
public const string WatermeterSubscriberWorkerFiveMinuteIssuedEventName = "issued.auto.five.watermeter.event";
/// <summary>
/// 15分钟采集水表数据下行消息主题
/// </summary>
public const string WatermeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.fifteen.watermeter.event";
public const string WatermeterSubscriberWorkerFifteenMinuteIssuedEventName = "issued.auto.fifteen.watermeter.event";
/// <summary>
/// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号、定时阀控等
/// 其他采集数据下行消息主题，日冻结，月冻结、集中器版本号等
/// </summary>
public const string WatermeterSubscriberWorkerOtherIssuedEventName = "issued.other.watermeter.event";
public const string WatermeterSubscriberWorkerOtherIssuedEventName = "issued.auto.other.watermeter.event";
/// <summary>
/// 水表自动阀控
/// </summary>
public const string WatermeterSubscriberWorkerAutoValveControlIssuedEventName = "issued.auto.control.watermeter.event";
/// <summary>
/// 水表手动阀控
/// </summary>
public const string WatermeterSubscriberWorkerManualValveControlIssuedEventName = "issued.control.watermeter.event";
public const string WatermeterSubscriberWorkerManualValveControlIssuedEventName = "issued.manual.control.watermeter.event";
/// <summary>
/// 水表手动抄读
/// </summary>
public const string WatermeterSubscriberWorkerManualValveReadingIssuedEventName = "issued.manual.reading.watermeter.event";
#endregion
/// <summary>
/// AFN上行主题格式
/// </summary>
public const string AFNTopicNameFormat = "received.afn{0}.event";
}
}