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

调整项目结构依赖，移除Mongo、Hangfire、Cassandra

Browse Source
This commit is contained in:
ChenYi 2025-05-26 11:20:32 +08:00
parent 04da31f423
commit 8f3dea6c2c
32 changed files with 373 additions and 777 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
protocols/JiShe.CollectBus.Protocol.T1882018/JiShe.CollectBus.Protocol.T1882018.csproj
View File

@ -9,6 +9,8 @@
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.2" />
<PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
</ItemGroup>
<ItemGroup>

3
protocols/JiShe.CollectBus.Protocol.T37612012/T37612012ProtocolPlugin.cs
View File

@ -27,7 +27,6 @@ namespace JiShe.CollectBus.Protocol.T37612012
private readonly IProducerService _producerService;
private readonly IRepository<Device, Guid> _deviceRepository;
private readonly ITcpService _tcpService;
public readonly Dictionary<string, Telemetry3761PacketBuilder.T3761Delegate> T3761AFNHandlers;
@ -41,7 +40,6 @@ namespace JiShe.CollectBus.Protocol.T37612012
_logger = logger;
//_logger = serviceProvider.GetRequiredService<ILogger<StandardProtocolPlugin>>();
_producerService = serviceProvider.GetRequiredService<IProducerService>();
_deviceRepository = serviceProvider.GetRequiredService<IRepository<Device, Guid>>();
_tcpService = tcpService;
T3761AFNHandlers = Telemetry3761PacketBuilder.T3761AFNHandlers;
}
@ -133,7 +131,6 @@ namespace JiShe.CollectBus.Protocol.T37612012
else
{
_logger.LogError($"不支持的上报kafka主题：{topicName}");
await _producerService.ProduceAsync(ProtocolConst.SubscriberReceivedEventName, messageReceivedAnalysis);
}
}

4
protocols/JiShe.CollectBus.Protocol/Abstracts/ProtocolPlugin.cs
View File

@ -20,13 +20,11 @@ namespace JiShe.CollectBus.Protocol.Abstracts
public const string errorData = "EE";
private readonly ILogger _logger;
private readonly IRepository<ProtocolInfo, Guid> _protocolInfoRepository;
private readonly IFreeRedisProvider _redisProvider;
public ProtocolPlugin(IServiceProvider serviceProvider, ILogger logger)
{
_logger = logger;
_protocolInfoRepository = serviceProvider.GetRequiredService<IRepository<ProtocolInfo, Guid>>();
_redisProvider = serviceProvider.GetRequiredService<IFreeRedisProvider>();
}
@ -42,8 +40,6 @@ namespace JiShe.CollectBus.Protocol.Abstracts
throw new ArgumentNullException(nameof(Info));
}
await _protocolInfoRepository.DeleteDirectAsync(a => a.Name == Info.Name);
await _protocolInfoRepository.InsertAsync(Info);
await _redisProvider.Instance.HDelAsync($"{RedisConst.ProtocolKey}", Info.Name);
await _redisProvider.Instance.HSetAsync($"{RedisConst.ProtocolKey}", Info.Name, Info);
}

1
services/JiShe.CollectBus.Application.Contracts/Subscribers/ISubscriberAppService.cs
View File

@ -12,7 +12,6 @@ namespace JiShe.CollectBus.Subscribers
{
Task<ISubscribeAck> LoginIssuedEvent(List<IssuedEventMessage> issuedEventMessage);
Task<ISubscribeAck> HeartbeatIssuedEvent(List<IssuedEventMessage> issuedEventMessage);
Task<ISubscribeAck> ReceivedEvent(MessageProtocolAnalysis<TB3761> receivedMessage);
Task<ISubscribeAck> ReceivedHeartbeatEvent(List<MessageReceivedHeartbeat> receivedHeartbeatMessage);
Task<ISubscribeAck> ReceivedLoginEvent(List<MessageReceivedLogin> receivedLoginMessage);
}

47
services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
View File

@ -1,13 +1,10 @@
using Cassandra.Mapping;
using JiShe.CollectBus.Cassandra;
using JiShe.CollectBus.DataChannels;
using JiShe.CollectBus.DataChannels;
using JiShe.CollectBus.FreeRedis;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.Interceptors;
using JiShe.CollectBus.IoTDB;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Mappers;
using JiShe.CollectBus.Protocol;
using JiShe.CollectBus.ScheduledMeterReading;
using Microsoft.Extensions.DependencyInjection;
@ -19,12 +16,8 @@ using System.Threading.Channels;
using System.Threading.Tasks;
using Volo.Abp;
using Volo.Abp.Application;
using Volo.Abp.AuditLogging;
using Volo.Abp.Autofac;
using Volo.Abp.AutoMapper;
using Volo.Abp.BackgroundJobs;
using Volo.Abp.BackgroundWorkers;
using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.Modularity;
namespace JiShe.CollectBus;
@ -34,16 +27,12 @@ namespace JiShe.CollectBus;
typeof(CollectBusApplicationContractsModule),
typeof(AbpDddApplicationModule),
typeof(AbpAutoMapperModule),
typeof(AbpAutofacModule),
typeof(AbpBackgroundWorkersHangfireModule),
typeof(CollectBusFreeRedisModule),
typeof(CollectBusFreeSqlModule),
typeof(CollectBusKafkaModule),
typeof(CollectBusIoTDbModule),
typeof(AbpAutofacModule),
typeof(CollectBusDomainSharedModule),
typeof(AbpAuditLoggingDomainModule),
typeof(AbpBackgroundJobsDomainModule),
typeof(CollectBusCassandraModule),
typeof(CollectBusProtocolModule)
)]
public class CollectBusApplicationModule : AbpModule
@ -55,8 +44,8 @@ public class CollectBusApplicationModule : AbpModule
context.Services.AddAutoMapperObjectMapper<CollectBusApplicationModule>();
Configure<AbpAutoMapperOptions>(options => { options.AddMaps<CollectBusApplicationModule>(true); });
context.Services.AddSingleton(new MappingConfiguration()
.Define(new CollectBusMapping()));
//context.Services.AddSingleton(new MappingConfiguration()
// .Define(new CollectBusMapping()));
// 注册拦截器
context.Services.OnRegistered(ctx =>
@ -70,33 +59,17 @@ public class CollectBusApplicationModule : AbpModule
public override async Task OnApplicationInitializationAsync(
ApplicationInitializationContext context)
{
var assembly = Assembly.GetExecutingAssembly();
var types = assembly.GetTypes().Where(t => typeof(ICollectWorker).IsAssignableFrom(t) && !t.IsInterface)
.ToList();
foreach (var type in types) await context.AddBackgroundWorkerAsync(type);
//Task.Run(() =>
//var assembly = Assembly.GetExecutingAssembly();
//var types = assembly.GetTypes().Where(t => typeof(ICollectWorker).IsAssignableFrom(t) && !t.IsInterface)
// .ToList();
//foreach (var type in types)
//{
// //默认初始化表计信息
// var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
// dbContext.InitAmmeterCacheData();
// //await dbContext.InitWatermeterCacheData();
//}).ConfigureAwait(false);
// await context.AddBackgroundWorkerAsync(type);
//}
//下发任务通道构建
DataChannelManage.TaskDataChannel = Channel.CreateUnbounded<ValueTuple<string, List<MeterReadingTelemetryPacketInfo>>>();
// 日志存储通道构建
DataChannelManage.LogSaveChannel = Channel.CreateUnbounded<object>();
// 日志刷新通道构建
DataChannelManage.LogRefreshChannel = Channel.CreateUnbounded<object>();
// 启动通道任务
var _dataChannelManage = context.ServiceProvider.GetRequiredService<DataChannelManageService>();
_ = _dataChannelManage.LogSaveAsync(DataChannelManage.LogSaveChannel.Reader);
//默认初始化表计信息
var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
await dbContext.InitAmmeterCacheData("V4-Gather-8890");

236
services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs
View File

@ -1,6 +1,4 @@
using Amazon.Runtime.Internal.Transform;
using DnsClient.Protocol;
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface;
@ -10,7 +8,6 @@ using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Dto;
using JiShe.CollectBus.Protocol.Models;
using JiShe.CollectBus.Repository.LogRecord;
using Mapster;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
@ -36,22 +33,19 @@ namespace JiShe.CollectBus.DataChannels
private readonly IProducerService _producerService;
private readonly KafkaOptionConfig _kafkaOptions;
private readonly ServerApplicationOptions _applicationOptions;
private readonly ILogRecordRepository _logRecordRepository;
public DataChannelManageService(
ILogger<DataChannelManageService> logger,
IIoTDbProvider dbProvider,
IProducerService producerService,
IOptions<KafkaOptionConfig> kafkaOptions,
IOptions<ServerApplicationOptions> applicationOptions,
ILogRecordRepository logRecordRepository)
IOptions<ServerApplicationOptions> applicationOptions)
{
_logger = logger;
_dbProvider = dbProvider;
_producerService = producerService;
_kafkaOptions = kafkaOptions.Value;
_applicationOptions = applicationOptions.Value;
_logRecordRepository= logRecordRepository;
}
/// <summary>
@ -207,126 +201,126 @@ namespace JiShe.CollectBus.DataChannels
/// <summary>
/// 日志保存
/// </summary>
/// <param name="channelReader"></param>
/// <returns></returns>
public async Task LogSaveAsync(ChannelReader<object> channelReader)
{
const int BatchSize = 1000;
const int EmptyWaitMilliseconds = 1000;
var timeout = TimeSpan.FromSeconds(2);
var timer = Stopwatch.StartNew();
long timeoutMilliseconds = 0;
try
{
while (true)
{
var batch = new List<object>();
var canRead = channelReader.Count;
if (canRead <= 0)
{
if (timeoutMilliseconds > 0)
{
_logger.LogError($"{nameof(LogSaveAsync)} 通道处理数据耗时{timeoutMilliseconds}毫秒");
}
timeoutMilliseconds = 0;
//无消息时短等待1秒
await Task.Delay(EmptyWaitMilliseconds);
continue;
}
///// <summary>
///// 日志保存
///// </summary>
///// <param name="channelReader"></param>
///// <returns></returns>
//public async Task LogSaveAsync(ChannelReader<object> channelReader)
//{
// const int BatchSize = 1000;
// const int EmptyWaitMilliseconds = 1000;
// var timeout = TimeSpan.FromSeconds(2);
// var timer = Stopwatch.StartNew();
// long timeoutMilliseconds = 0;
// try
// {
// while (true)
// {
// var batch = new List<object>();
// var canRead = channelReader.Count;
// if (canRead <= 0)
// {
// if (timeoutMilliseconds > 0)
// {
// _logger.LogError($"{nameof(LogSaveAsync)} 通道处理数据耗时{timeoutMilliseconds}毫秒");
// }
// timeoutMilliseconds = 0;
// //无消息时短等待1秒
// await Task.Delay(EmptyWaitMilliseconds);
// continue;
// }
timer.Restart();
var startTime = DateTime.Now;
// timer.Restart();
// var startTime = DateTime.Now;
try
{
// 异步批量读取数据
while (batch != null && batch.Count < BatchSize && (DateTime.Now - startTime) < timeout)
{
try
{
if (channelReader.TryRead(out var dataItem))
{
batch.Add(dataItem);
}
}
catch (Exception)
{
throw;
}
}
}
catch (Exception)
{
throw;
}
// try
// {
// // 异步批量读取数据
// while (batch != null && batch.Count < BatchSize && (DateTime.Now - startTime) < timeout)
// {
// try
// {
// if (channelReader.TryRead(out var dataItem))
// {
// batch.Add(dataItem);
// }
// }
// catch (Exception)
// {
// throw;
// }
// }
// }
// catch (Exception)
// {
// throw;
// }
if (batch == null || batch.Count == 0)
{
await Task.Delay(EmptyWaitMilliseconds);
continue;
}
try
{
// if (batch == null || batch.Count == 0)
// {
// await Task.Delay(EmptyWaitMilliseconds);
// continue;
// }
// try
// {
// 按小时分组
var hourGroups = new Dictionary<DateTime, List<LogRecords>>();
DateTime? dateTime = null;
List<LogRecords> batchList = new List<LogRecords>();
int index = 1;
foreach (var item in batch)
{
var records = item.Adapt<LogRecords>();
// // 按小时分组
// var hourGroups = new Dictionary<DateTime, List<LogRecords>>();
// DateTime? dateTime = null;
// List<LogRecords> batchList = new List<LogRecords>();
// int index = 1;
// foreach (var item in batch)
// {
// var records = item.Adapt<LogRecords>();
if (!records.ReceivedTime.HasValue)
records.ReceivedTime = DateTime.Now;
var curDateTime = new DateTime(records.ReceivedTime.Value.Year, records.ReceivedTime.Value.Month, records.ReceivedTime.Value.Hour, records.ReceivedTime.Value.Day, records.ReceivedTime.Value.Hour, 0, 0);
if (!dateTime.HasValue || curDateTime != dateTime)
{
dateTime = curDateTime;
if (batchList.Count > 0)
{
var immutableList = ImmutableList.CreateRange(batchList);
hourGroups.Add(dateTime.Value, immutableList.ToList());
batchList.Clear();
}
}
batchList.Add(records);
// 最后一批
if(index== batch.Count)
{
var immutableList = ImmutableList.CreateRange(batchList);
hourGroups.Add(dateTime.Value, immutableList.ToList());
batchList.Clear();
}
index++;
}
foreach (var (time, records) in hourGroups)
{
// 批量写入数据库
await _logRecordRepository.InsertManyAsync(records, time);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "数据通道处理日志数据时发生异常");
}
batch.Clear();
timer.Stop();
// if (!records.ReceivedTime.HasValue)
// records.ReceivedTime = DateTime.Now;
// var curDateTime = new DateTime(records.ReceivedTime.Value.Year, records.ReceivedTime.Value.Month, records.ReceivedTime.Value.Hour, records.ReceivedTime.Value.Day, records.ReceivedTime.Value.Hour, 0, 0);
// if (!dateTime.HasValue || curDateTime != dateTime)
// {
// dateTime = curDateTime;
// if (batchList.Count > 0)
// {
// var immutableList = ImmutableList.CreateRange(batchList);
// hourGroups.Add(dateTime.Value, immutableList.ToList());
// batchList.Clear();
// }
// }
// batchList.Add(records);
// // 最后一批
// if(index== batch.Count)
// {
// var immutableList = ImmutableList.CreateRange(batchList);
// hourGroups.Add(dateTime.Value, immutableList.ToList());
// batchList.Clear();
// }
// index++;
// }
// foreach (var (time, records) in hourGroups)
// {
// // 批量写入数据库
// await _logRecordRepository.InsertManyAsync(records, time);
// }
// }
// catch (Exception ex)
// {
// _logger.LogError(ex, "数据通道处理日志数据时发生异常");
// }
// batch.Clear();
// timer.Stop();
timeoutMilliseconds = timeoutMilliseconds + timer.ElapsedMilliseconds;
// timeoutMilliseconds = timeoutMilliseconds + timer.ElapsedMilliseconds;
startTime = DateTime.Now;
}
}
catch (Exception ex)
{
_logger.LogCritical(ex, "日志处理发生致命错误");
throw;
}
}
// startTime = DateTime.Now;
// }
// }
// catch (Exception ex)
// {
// _logger.LogCritical(ex, "日志处理发生致命错误");
// throw;
// }
//}
}
}

10
services/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
View File

@ -17,20 +17,22 @@
<ItemGroup>
<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="3.1.2" />
<PackageReference Include="TouchSocket.Hosting" Version="3.1.2" />
<PackageReference Include="Volo.Abp.EventBus.Kafka" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
<ProjectReference Include="..\..\protocols\JiShe.CollectBus.Protocol\JiShe.CollectBus.Protocol.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.Cassandra\JiShe.CollectBus.Cassandra.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.FreeRedis\JiShe.CollectBus.FreeRedis.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.Kafka\JiShe.CollectBus.Kafka.csproj" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
</ItemGroup>
<ItemGroup>
<Folder Include="Mappers\" />
<Folder Include="Workers\" />
</ItemGroup>
</Project>

17
services/JiShe.CollectBus.Application/Mappers/CollectBusMapping.cs
View File

@ -1,17 +0,0 @@
using Cassandra.Mapping;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
namespace JiShe.CollectBus.Mappers
{
public class CollectBusMapping: Mappings
{
public CollectBusMapping()
{
For<MessageIssued>()
.Column(e => e.Type, cm => cm.WithName("type").WithDbType<int>());
For<Device>()
.Column(e => e.Status, cm => cm.WithName("status").WithDbType<int>());
}
}
}

143
services/JiShe.CollectBus.Application/Samples/TestAppService.cs
View File

@ -1,6 +1,4 @@
using Cassandra;
using JiShe.CollectBus.Cassandra;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.DynamicModule;
using JiShe.CollectBus.Interceptors;
using JiShe.CollectBus.IotSystems.MessageIssueds;
@ -22,8 +20,8 @@ namespace JiShe.CollectBus.Samples;
public class TestAppService : CollectBusAppService
{
private readonly ILogger<TestAppService> _logger;
private readonly ICassandraRepository<MessageIssued, string> _messageReceivedCassandraRepository;
private readonly ICassandraProvider _cassandraProvider;
//private readonly ICassandraRepository<MessageIssued, string> _messageReceivedCassandraRepository;
//private readonly ICassandraProvider _cassandraProvider;
private readonly IProtocolService _protocolService;
private readonly IServiceProvider _serviceProvider;
private readonly IDynamicModuleManager _dynamicModuleManager;
@ -31,93 +29,94 @@ public class TestAppService : CollectBusAppService
public TestAppService(
ILogger<TestAppService> logger,
ICassandraRepository<MessageIssued, string> messageReceivedCassandraRepository,
ICassandraProvider cassandraProvider, IProtocolService protocolService,IServiceProvider serviceProvider, IDynamicModuleManager dynamicModuleManager)
//ICassandraRepository<MessageIssued, string> messageReceivedCassandraRepository,
//ICassandraProvider cassandraProvider,
IProtocolService protocolService,IServiceProvider serviceProvider, IDynamicModuleManager dynamicModuleManager)
{
_logger = logger;
_messageReceivedCassandraRepository = messageReceivedCassandraRepository;
_cassandraProvider = cassandraProvider;
//_messageReceivedCassandraRepository = messageReceivedCassandraRepository;
//_cassandraProvider = cassandraProvider;
_protocolService = protocolService;
_serviceProvider = serviceProvider;
_dynamicModuleManager = dynamicModuleManager;
}
public async Task AddMessageOfCassandra()
{
var stopwatch = Stopwatch.StartNew();
for (int i = 1; i <= 10000; i++)
{
var str = Guid.NewGuid().ToString();
await _messageReceivedCassandraRepository.InsertAsync(new MessageIssued
{
ClientId = str,
DeviceNo = i.ToString(),
MessageId = str,
Type = IssuedEventType.Data,
Id = str,
Message = str.GetBytes()
});
}
stopwatch.Stop();
_logger.LogWarning($"插入 {10000} 条记录完成，耗时: {stopwatch.ElapsedMilliseconds} 毫秒");
//var stopwatch = Stopwatch.StartNew();
//for (int i = 1; i <= 10000; i++)
//{
// var str = Guid.NewGuid().ToString();
// await _messageReceivedCassandraRepository.InsertAsync(new MessageIssued
// {
// ClientId = str,
// DeviceNo = i.ToString(),
// MessageId = str,
// Type = IssuedEventType.Data,
// Id = str,
// Message = str.GetBytes()
// });
//}
//stopwatch.Stop();
//_logger.LogWarning($"插入 {10000} 条记录完成，耗时: {stopwatch.ElapsedMilliseconds} 毫秒");
}
public async Task AddMessageOfBulkInsertCassandra()
{
var records = new List<MessageIssued>();
var prepared = await _cassandraProvider.Session.PrepareAsync(
$"INSERT INTO {_cassandraProvider.CassandraConfig.Keyspace}.{nameof(MessageIssued)} (id, clientid, message, deviceno,type,messageid) VALUES (?, ?, ?, ?, ?, ?)");
//var records = new List<MessageIssued>();
//var prepared = await _cassandraProvider.Session.PrepareAsync(
// $"INSERT INTO {_cassandraProvider.CassandraConfig.Keyspace}.{nameof(MessageIssued)} (id, clientid, message, deviceno,type,messageid) VALUES (?, ?, ?, ?, ?, ?)");
for (int i = 1; i <= 100000; i++)
{
var str = Guid.NewGuid().ToString();
records.Add(new MessageIssued
{
ClientId = str,
DeviceNo = i.ToString(),
MessageId = str,
Type = IssuedEventType.Data,
Id = str,
Message = str.GetBytes()
});
}
var stopwatch = Stopwatch.StartNew();
await BulkInsertAsync(_cassandraProvider.Session, prepared, records);
stopwatch.Stop();
_logger.LogWarning($"插入 {100000} 条记录完成，耗时: {stopwatch.ElapsedMilliseconds} 毫秒");
//for (int i = 1; i <= 100000; i++)
//{
// var str = Guid.NewGuid().ToString();
// records.Add(new MessageIssued
// {
// ClientId = str,
// DeviceNo = i.ToString(),
// MessageId = str,
// Type = IssuedEventType.Data,
// Id = str,
// Message = str.GetBytes()
// });
//}
//var stopwatch = Stopwatch.StartNew();
//await BulkInsertAsync(_cassandraProvider.Session, prepared, records);
//stopwatch.Stop();
//_logger.LogWarning($"插入 {100000} 条记录完成，耗时: {stopwatch.ElapsedMilliseconds} 毫秒");
}
private static async Task BulkInsertAsync(ISession session, PreparedStatement prepared, List<MessageIssued> records)
{
var tasks = new List<Task>();
var batch = new BatchStatement();
//private static async Task BulkInsertAsync(ISession session, PreparedStatement prepared, List<MessageIssued> records)
//{
// var tasks = new List<Task>();
// var batch = new BatchStatement();
for (int i = 0; i < records.Count; i++)
{
var record = records[i];
var boundStatement = prepared.Bind(
record.Id,
record.ClientId,
record.Message,
record.DeviceNo,
(int)record.Type,
record.MessageId);
// for (int i = 0; i < records.Count; i++)
// {
// var record = records[i];
// var boundStatement = prepared.Bind(
// record.Id,
// record.ClientId,
// record.Message,
// record.DeviceNo,
// (int)record.Type,
// record.MessageId);
// 设置一致性级别为ONE以提高性能
boundStatement.SetConsistencyLevel(ConsistencyLevel.One);
// // 设置一致性级别为ONE以提高性能
// boundStatement.SetConsistencyLevel(ConsistencyLevel.One);
batch.Add(boundStatement);
// batch.Add(boundStatement);
// 当达到批处理大小时执行
if (batch.Statements.Count() >= 1000 || i == records.Count - 1)
{
tasks.Add(session.ExecuteAsync(batch));
batch = new BatchStatement();
}
}
// // 当达到批处理大小时执行
// if (batch.Statements.Count() >= 1000 || i == records.Count - 1)
// {
// tasks.Add(session.ExecuteAsync(batch));
// batch = new BatchStatement();
// }
// }
// 等待所有批处理完成
await Task.WhenAll(tasks);
}
// // 等待所有批处理完成
// await Task.WhenAll(tasks);
//}
[LogIntercept]
public virtual Task<string> LogInterceptorTest(string str)

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

@ -334,8 +334,6 @@ namespace JiShe.CollectBus.ScheduledMeterReading
// 创建取消令牌源
//var cts = new CancellationTokenSource();
await _dbProvider.GetSessionPool(true).InitTableSessionModelAsync();
_ = _dataChannelManage.ScheduledMeterTaskReadingAsync(DataChannelManage.TaskDataChannel.Reader);
// //此处代码不要删除

5
services/JiShe.CollectBus.Application/Subscribers/SubscriberAnalysisAppService.cs
View File

@ -7,7 +7,6 @@ using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Interfaces;
using JiShe.CollectBus.Protocol3761;
using JiShe.CollectBus.Repository.MeterReadingRecord;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System;
@ -24,7 +23,6 @@ namespace JiShe.CollectBus.Subscribers
private readonly ILogger<SubscriberAnalysisAppService> _logger;
private readonly ITcpService _tcpService;
private readonly IServiceProvider _serviceProvider;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
private readonly IIoTDbProvider _dbProvider;
private readonly IProtocolService _protocolService;
@ -32,12 +30,11 @@ namespace JiShe.CollectBus.Subscribers
ITcpService tcpService,
IServiceProvider serviceProvider,
IIoTDbProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordsRepository, IProtocolService protocolService)
IProtocolService protocolService)
{
_logger = logger;
_tcpService = tcpService;
_serviceProvider = serviceProvider;
_meterReadingRecordsRepository = meterReadingRecordsRepository;
_dbProvider = dbProvider;
_protocolService = protocolService;
}

172
services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -1,172 +0,0 @@
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.Interceptors;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Protocol.Interfaces;
using JiShe.CollectBus.Protocol3761;
using JiShe.CollectBus.Repository.MeterReadingRecord;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using TouchSocket.Sockets;
using Volo.Abp.Domain.Repositories;
namespace JiShe.CollectBus.Subscribers
{
public class SubscriberAppService : CollectBusAppService, ISubscriberAppService, IKafkaSubscribe
{
private readonly ILogger<SubscriberAppService> _logger;
private readonly ITcpService _tcpService;
private readonly IServiceProvider _serviceProvider;
private readonly IRepository<MessageReceivedLogin, Guid> _messageReceivedLoginEventRepository;
private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
private readonly IIoTDbProvider _dbProvider;
private readonly IProtocolService _protocolService;
/// <summary>
/// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
/// </summary>
/// <param name="logger">The logger.</param>
/// <param name="tcpService">The TCP service.</param>
/// <param name="serviceProvider">The service provider.</param>
/// <param name="messageReceivedLoginEventRepository">The message received login event repository.</param>
/// <param name="messageReceivedHeartbeatEventRepository">The message received heartbeat event repository.</param>
/// <param name="meterReadingRecordsRepository">The device repository.</param>
public SubscriberAppService(ILogger<SubscriberAppService> logger,
ITcpService tcpService,
IServiceProvider serviceProvider,
IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
IIoTDbProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordsRepository, IProtocolService protocolService)
{
_logger = logger;
_tcpService = tcpService;
_serviceProvider = serviceProvider;
_messageReceivedLoginEventRepository = messageReceivedLoginEventRepository;
_messageReceivedHeartbeatEventRepository = messageReceivedHeartbeatEventRepository;
_meterReadingRecordsRepository = meterReadingRecordsRepository;
_dbProvider = dbProvider;
_protocolService = protocolService;
}
[LogIntercept]
[KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName, EnableBatch = true)]
public async Task<ISubscribeAck> LoginIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
{
bool isAck = true;
foreach (var issuedEventMessage in issuedEventMessages)
{
//var loginEntity = await _messageReceivedLoginEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
//if (loginEntity == null)
//{
// isAck=false;
// break;
//}
//loginEntity.AckTime = Clock.Now;
//loginEntity.IsAck = true;
//await _messageReceivedLoginEventRepository.UpdateAsync(loginEntity);
}
// TODO:暂时ACK，等后续处理是否放到私信队列中
return isAck? SubscribeAck.Success(): SubscribeAck.Fail();
}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName, EnableBatch = true)]
public async Task<ISubscribeAck> HeartbeatIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
{
bool isAck = true;
//foreach (var issuedEventMessage in issuedEventMessages)
//{
// var heartbeatEntity = await _messageReceivedHeartbeatEventRepository.FirstOrDefaultAsync(a => a.MessageId == issuedEventMessage.MessageId);
// if (heartbeatEntity == null)
// {
// isAck = false;
// break;
// }
// heartbeatEntity.AckTime = Clock.Now;
// heartbeatEntity.IsAck = true;
// await _messageReceivedHeartbeatEventRepository.UpdateAsync(heartbeatEntity);
// }
// TODO:暂时ACK，等后续处理是否放到私信队列中
return isAck ? SubscribeAck.Success() : SubscribeAck.Fail();
}
[KafkaSubscribe(ProtocolConst.SubscriberReceivedEventName)]
public async Task<ISubscribeAck> ReceivedEvent(MessageProtocolAnalysis<TB3761> receivedMessage)
{
var currentTime = Clock.Now;
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
{
_logger.LogError("协议不存在！");
}
else
{
//todo 会根据不同的协议进行解析，然后做业务处理
//TB3761? tB3761 = protocolPlugin.Analysis3761(receivedMessage.MessageHexString);
//if (tB3761 == null)
//{
// Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
// return SubscribeAck.Success();
//}
//if (tB3761.DT == null || tB3761.AFN_FC == null)
//{
// Logger.LogError($"数据解析失败：{receivedMessage.Serialize()}");
// return SubscribeAck.Success();
//}
//报文入库
var entity = new MeterReadingRecords()
{
ReceivedMessageHexString = receivedMessage.MessageHexString,
AFN = (AFN)receivedMessage.Data?.AFN_FC.AFN!,
Fn = receivedMessage.Data.DT.Fn,
Pn = 0,
FocusAddress = "",
MeterAddress = "",
};
//如果没数据，则插入，有数据则更新
var updateEntity = await _meterReadingRecordsRepository.FirOrDefaultAsync(entity, currentTime);
if (updateEntity == null)
{
await _meterReadingRecordsRepository.InsertAsync(entity, currentTime);
}
//_dbProvider.InsertAsync();
//todo 查找是否有下发任务
//await _messageReceivedEventRepository.InsertAsync(receivedMessage);
}
return SubscribeAck.Success();
}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName, EnableBatch = true)]
public async Task<ISubscribeAck> ReceivedHeartbeatEvent(List<MessageReceivedHeartbeat> receivedHeartbeatMessages)
{
await _messageReceivedHeartbeatEventRepository.InsertManyAsync(receivedHeartbeatMessages);
return SubscribeAck.Success();
}
[KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName,EnableBatch =true)]
public async Task<ISubscribeAck> ReceivedLoginEvent(List<MessageReceivedLogin> receivedLoginMessages)
{
await _messageReceivedLoginEventRepository.InsertManyAsync(receivedLoginMessages);
return SubscribeAck.Success();
}
}
}

43
services/JiShe.CollectBus.Application/Workers/CreateToBeIssueTaskWorker.cs
View File

@ -1,43 +0,0 @@
//using System;
//using System.Threading;
//using System.Threading.Tasks;
//using Hangfire;
//using JiShe.CollectBus.Common.Consts;
//using JiShe.CollectBus.ScheduledMeterReading;
//using Microsoft.Extensions.Logging;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//using Volo.Abp.Uow;
//namespace JiShe.CollectBus.Workers
//{
// /// <summary>
// /// 构建待处理的下发指令任务处理
// /// </summary>
// public class CreateToBeIssueTaskWorker : HangfireBackgroundWorkerBase, ITransientDependency, ICollectWorker
// {
// private readonly ILogger<CreateToBeIssueTaskWorker> _logger;
// private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
// /// <summary>
// /// Initializes a new instance of the <see cref="CreateToBeIssueTaskWorker"/> class.
// /// </summary>
// /// <param name="logger">The logger.</param>
// /// <param name="scheduledMeterReadingService">定时任务</param>
// public CreateToBeIssueTaskWorker(ILogger<CreateToBeIssueTaskWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
// {
// _logger = logger;
// RecurringJobId = nameof(CreateToBeIssueTaskWorker);
// CronExpression = "0 0/1 * * * *";
// TimeZone = TimeZoneInfo.Local;
// this._scheduledMeterReadingService = scheduledMeterReadingService;
// }
// public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// _logger.LogError($"{DateTime.Now}");
// // await _scheduledMeterReadingService.CreateToBeIssueTasks();
// }
// }
//}

39
services/JiShe.CollectBus.Application/Workers/DataDetectionFifteenMinuteWorker.cs
View File

@ -1,39 +0,0 @@
//using JiShe.CollectBus.ScheduledMeterReading;
//using Microsoft.Extensions.Logging;
//using System;
//using System.Collections.Generic;
//using System.Linq;
//using System.Text;
//using System.Threading;
//using System.Threading.Tasks;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//namespace JiShe.CollectBus.Workers
//{
// /// <summary>
// /// 定时数据检测1小时一次
// /// </summary>
// public class DataDetectionFifteenMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency, ICollectWorker
// {
// private readonly ILogger<CreateToBeIssueTaskWorker> _logger;
// private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
// public DataDetectionFifteenMinuteWorker(ILogger<CreateToBeIssueTaskWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
// {
// _logger = logger;
// RecurringJobId = nameof(CreateToBeIssueTaskWorker);
// CronExpression = "0 0 0/1 * * ?";
// TimeZone = TimeZoneInfo.Local;
// this._scheduledMeterReadingService = scheduledMeterReadingService;
// }
// public override Task DoWorkAsync(CancellationToken cancellationToken = default)
// {
// //throw new NotImplementedException();
// return Task.CompletedTask;
// }
// }
//}

39
services/JiShe.CollectBus.Application/Workers/EpiCollectWorker.cs
View File

@ -1,39 +0,0 @@
//using System;
//using System.Threading;
//using System.Threading.Tasks;
//using Hangfire;
//using JiShe.CollectBus.Common.Attributes;
//using Microsoft.Extensions.Logging;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//using Volo.Abp.Uow;
//namespace JiShe.CollectBus.Workers
//{
// public class EpiCollectWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
// {
// private readonly ILogger<EpiCollectWorker> _logger;
// /// <summary>
// /// Initializes a new instance of the <see cref="EpiCollectWorker"/> class.
// /// </summary>
// /// <param name="logger">The logger.</param>
// public EpiCollectWorker(ILogger<EpiCollectWorker> logger)
// {
// _logger = logger;
// RecurringJobId = nameof(EpiCollectWorker);
// CronExpression = Cron.Daily();
// }
// public override Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// using (var uow = LazyServiceProvider.LazyGetRequiredService<IUnitOfWorkManager>().Begin())
// {
// Logger.LogInformation("Executed MyLogWorker..!");
// return Task.CompletedTask;
// }
// }
// }
//}

48
services/JiShe.CollectBus.Application/Workers/SubscriberFifteenMinuteWorker.cs
View File

@ -1,48 +0,0 @@
//using System;
//using System.Threading;
//using System.Threading.Tasks;
//using Hangfire;
//using JiShe.CollectBus.ScheduledMeterReading;
//using Microsoft.Extensions.Logging;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//using Volo.Abp.Uow;
//namespace JiShe.CollectBus.Workers
//{
// /// <summary>
// /// 15分钟采集数据
// /// </summary>
// public class SubscriberFifteenMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency, ICollectWorker
// {
// private readonly ILogger<SubscriberFifteenMinuteWorker> _logger;
// private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
// /// <summary>
// /// Initializes a new instance of the <see cref="SubscriberFifteenMinuteWorker"/> class.
// /// </summary>
// /// <param name="logger">The logger.</param>
// /// <param name="scheduledMeterReadingService">定时任务</param>
// public SubscriberFifteenMinuteWorker(ILogger<SubscriberFifteenMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
// {
// _logger = logger;
// RecurringJobId = nameof(SubscriberFifteenMinuteWorker);
// CronExpression = "0 0/15 * * * *";
// TimeZone = TimeZoneInfo.Local;
// this._scheduledMeterReadingService = scheduledMeterReadingService;
// }
// public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// //await _scheduledMeterReadingService.AmmeterScheduledMeterFifteenMinuteReading();
// //await _scheduledMeterReadingService.WatermeterScheduledMeterFifteenMinuteReading();
// //using (var uow = LazyServiceProvider.LazyGetRequiredService<IUnitOfWorkManager>().Begin())
// //{
// // Logger.LogInformation("Executed MyLogWorker..!");
// // return Task.CompletedTask;
// //}
// }
// }
//}

42
services/JiShe.CollectBus.Application/Workers/SubscriberFiveMinuteWorker.cs
View File

@ -1,42 +0,0 @@
//using System;
//using System.Threading;
//using System.Threading.Tasks;
//using Hangfire;
//using JiShe.CollectBus.ScheduledMeterReading;
//using Microsoft.Extensions.Logging;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//using Volo.Abp.Uow;
//namespace JiShe.CollectBus.Workers
//{
// /// <summary>
// /// 5分钟采集数据
// /// </summary>
// public class SubscriberFiveMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
// {
// private readonly ILogger<SubscriberFiveMinuteWorker> _logger;
// private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
// /// <summary>
// /// Initializes a new instance of the <see cref="SubscriberFiveMinuteWorker"/> class.
// /// </summary>
// /// <param name="logger">The logger.</param>
// /// <param name="scheduledMeterReadingService">定时任务</param>
// public SubscriberFiveMinuteWorker(ILogger<SubscriberFiveMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
// {
// _logger = logger;
// RecurringJobId = nameof(SubscriberFiveMinuteWorker);
// CronExpression = "0 0/5 * * * *";
// TimeZone = TimeZoneInfo.Local;
// this._scheduledMeterReadingService = scheduledMeterReadingService;
// }
// public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// //await _scheduledMeterReadingService.AmmeterScheduledMeterFiveMinuteReading();
// //await _scheduledMeterReadingService.WatermeterScheduledMeterFiveMinuteReading();
// }
// }
//}

44
services/JiShe.CollectBus.Application/Workers/SubscriberOneMinuteWorker.cs
View File

@ -1,44 +0,0 @@
//using System;
//using System.Threading;
//using System.Threading.Tasks;
//using Hangfire;
//using JiShe.CollectBus.ScheduledMeterReading;
//using Microsoft.Extensions.Logging;
//using Volo.Abp.BackgroundWorkers.Hangfire;
//using Volo.Abp.DependencyInjection;
//using Volo.Abp.Uow;
//namespace JiShe.CollectBus.Workers
//{
// /// <summary>
// /// 1分钟采集数据
// /// </summary>
// public class SubscriberOneMinuteWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
// {
// private readonly ILogger<SubscriberOneMinuteWorker> _logger;
// private readonly IScheduledMeterReadingService _scheduledMeterReadingService;
// /// <summary>
// /// Initializes a new instance of the <see cref="SubscriberOneMinuteWorker"/> class.
// /// </summary>
// /// <param name="logger">The logger.</param>
// /// <param name="scheduledMeterReadingService">定时任务</param>
// public SubscriberOneMinuteWorker(ILogger<SubscriberOneMinuteWorker> logger, IScheduledMeterReadingService scheduledMeterReadingService)
// {
// _logger = logger;
// RecurringJobId = nameof(SubscriberOneMinuteWorker);
// CronExpression = "0 0/1 * * * *";
// TimeZone = TimeZoneInfo.Local;
// this._scheduledMeterReadingService = scheduledMeterReadingService;
// }
// public override async Task DoWorkAsync(CancellationToken cancellationToken = new CancellationToken())
// {
// //await _scheduledMeterReadingService.AmmeterScheduledMeterOneMinuteReading();
// //await _scheduledMeterReadingService.WatermeterScheduledMeterOneMinuteReading();
// }
// }
//}

4
services/JiShe.CollectBus.DbMigrator/CollectBusDbMigratorModule.cs
View File

@ -1,4 +1,4 @@
using JiShe.CollectBus.MongoDB;
using JiShe.CollectBus.IoTDB;
using Volo.Abp.Autofac;
using Volo.Abp.Modularity;
@ -6,7 +6,7 @@ namespace JiShe.CollectBus.DbMigrator;
[DependsOn(
typeof(AbpAutofacModule),
typeof(CollectBusMongoDbModule),
typeof(CollectBusIoTDbModule),
typeof(CollectBusApplicationContractsModule)
)]
public class CollectBusDbMigratorModule : AbpModule

11
services/JiShe.CollectBus.DbMigrator/DbMigratorHostedService.cs
View File

@ -7,6 +7,7 @@ using JiShe.CollectBus.Data;
using Serilog;
using Volo.Abp;
using Volo.Abp.Data;
using JiShe.CollectBus.IoTDB.Interface;
namespace JiShe.CollectBus.DbMigrator;
@ -33,10 +34,16 @@ public class DbMigratorHostedService : IHostedService
{
await application.InitializeAsync();
//await application
// .ServiceProvider
// .GetRequiredService<CollectBusDbMigrationService>()
// .MigrateAsync();
//初始化IoTDB表模型
await application
.ServiceProvider
.GetRequiredService<CollectBusDbMigrationService>()
.MigrateAsync();
.GetRequiredService<IIoTDbProvider>().GetSessionPool(true)
.InitTableSessionModelAsync();
await application.ShutdownAsync();

1
services/JiShe.CollectBus.DbMigrator/JiShe.CollectBus.DbMigrator.csproj
View File

@ -18,7 +18,6 @@
<ItemGroup>
<PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
<ProjectReference Include="..\..\modules\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.csproj" />
</ItemGroup>

9
services/JiShe.CollectBus.DbMigrator/appsettings.json
View File

@ -1,5 +1,14 @@
{
"ConnectionStrings": {
"Default": "mongodb://admin:collectbus_mongodb_jishe@118.190.144.92:37017/JiSheCollectBus?authSource=admin"
},
"IoTDBOptions": {
"UserName": "root",
"Password": "root",
"ClusterList": [ "192.168.5.9:6667" ],
"PoolSize": 32,
"DataBaseName": "energy",
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
}
}

4
shared/JiShe.CollectBus.Common/JiShe.CollectBus.Common.csproj
View File

@ -18,12 +18,16 @@
<ItemGroup>
<PackageReference Include="JetBrains.Annotations" Version="2024.2.0" />
<PackageReference Include="Mapster" Version="7.4.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.2" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Logging" Version="8.0.1" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="Serilog" Version="4.1.0" />
<PackageReference Include="System.Linq.Dynamic.Core" Version="1.6.4" />
<PackageReference Include="System.Text.Json" Version="8.0.5" />
<PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Ddd.Domain" Version="8.3.3" />
</ItemGroup>

82
shared/JiShe.CollectBus.Common/WorkService/SystemBackGroundWorkService.cs Normal file
View File

@ -0,0 +1,82 @@
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Common.WorkService
{
/// <summary>
/// 系统后台定时服务
/// </summary>
public abstract class SystemBackGroundWorkService : BackgroundService
{
/// <summary>
/// 日志记录
/// </summary>
public ILogger<SystemBackGroundWorkService> Logger { get; set; }
/// <summary>
/// 创建一个取消标记源
/// </summary>
private readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
/// <summary>
/// 任务执行间隔时间
/// </summary>
private TimeSpan interval;
protected SystemBackGroundWorkService(ILogger<SystemBackGroundWorkService> logger)
{
Logger = logger;
interval = GetInterval();
}
/// <summary>
/// 执行时间间隔
/// </summary>
/// <returns></returns>
protected abstract TimeSpan GetInterval();
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await Task.CompletedTask;// 等待其他任务执行完成，避免阻塞应用程序启动
Logger.LogInformation($"任务每隔{interval.TotalSeconds}秒执行一次");
await InitAsync(cancellationTokenSource.Token);
}
protected virtual async Task InitAsync(CancellationToken cancellationToken)
{
while (!cancellationToken.IsCancellationRequested)
{
try
{
await DoWorkAsync(cancellationToken);
}
catch (Exception ex)
{
Logger.LogError(ex, "后台任务执行发生异常");
}
await Task.Delay(interval, cancellationToken);
}
}
public override Task StopAsync(CancellationToken cancellationToken)
{
Logger.LogInformation("后台服务停止……");
cancellationTokenSource.Cancel();
return base.StopAsync(cancellationToken);
}
/// <summary>
/// 抛出方法入口以便于其服务实现
/// </summary>
/// <param name="cancellationToken"></param>
/// <returns></returns>
protected abstract Task DoWorkAsync(CancellationToken cancellationToken);
}
}

33
web/JiShe.CollectBus.Host/CollectBusHostModule.Configure.cs
View File

@ -19,7 +19,6 @@ using TouchSocket.Sockets;
using JiShe.CollectBus.Plugins;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using JiShe.CollectBus.Cassandra;
namespace JiShe.CollectBus.Host
@ -39,18 +38,18 @@ namespace JiShe.CollectBus.Host
Configure<AbpBackgroundJobOptions>(options => { options.IsJobExecutionEnabled = false; });
context.Services.AddHangfire(config =>
{
config.UseRedisStorage(
context.Services.GetConfiguration().GetValue<string>("Redis:Configuration"), redisStorageOptions)
.WithJobExpirationTimeout(TimeSpan.FromDays(7));
var delaysInSeconds = new[] { 10, 60, 60 * 3 }; // 重试时间间隔
const int Attempts = 3; // 重试次数
config.UseFilter(new AutomaticRetryAttribute() { Attempts = Attempts, DelaysInSeconds = delaysInSeconds });
//config.UseFilter(new AutoDeleteAfterSuccessAttribute(TimeSpan.FromDays(7)));
config.UseFilter(new JobRetryLastFilter(Attempts));
});
context.Services.AddHangfireServer();
//context.Services.AddHangfire(config =>
//{
// config.UseRedisStorage(
// context.Services.GetConfiguration().GetValue<string>("Redis:Configuration"), redisStorageOptions)
// .WithJobExpirationTimeout(TimeSpan.FromDays(7));
// var delaysInSeconds = new[] { 10, 60, 60 * 3 }; // 重试时间间隔
// const int Attempts = 3; // 重试次数
// config.UseFilter(new AutomaticRetryAttribute() { Attempts = Attempts, DelaysInSeconds = delaysInSeconds });
// //config.UseFilter(new AutoDeleteAfterSuccessAttribute(TimeSpan.FromDays(7)));
// config.UseFilter(new JobRetryLastFilter(Attempts));
//});
//context.Services.AddHangfireServer();
}
/// <summary>
@ -194,9 +193,9 @@ namespace JiShe.CollectBus.Host
options =>
{
options.IgnoredUrls.Add("/AuditLogs/page");
options.IgnoredUrls.Add("/hangfire/stats");
//options.IgnoredUrls.Add("/hangfire/stats");
options.IgnoredUrls.Add("/hangfire/recurring/trigger");
options.IgnoredUrls.Add("/cap");
//options.IgnoredUrls.Add("/cap");
options.IgnoredUrls.Add("/");
});
}
@ -253,10 +252,8 @@ namespace JiShe.CollectBus.Host
private void ConfigureHealthChecks(ServiceConfigurationContext context, IConfiguration configuration)
{
if (!configuration.GetValue<bool>("HealthChecks:IsEnable")) return;
var cassandraConfig = new CassandraConfig();
configuration.GetSection("Cassandra").Bind(cassandraConfig);
context.Services.AddHealthChecks()
.AddMongoDb(configuration.GetConnectionString("Default"), "MongoDB", HealthStatus.Unhealthy)
//.AddMongoDb(configuration.GetConnectionString("Default"), "MongoDB", HealthStatus.Unhealthy)
.AddRedis(configuration.GetValue<string>("Redis:Configuration") ?? string.Empty, "Redis",
HealthStatus.Unhealthy)
//.AddKafka(new Confluent.Kafka.ProducerConfig

18
web/JiShe.CollectBus.Host/CollectBusHostModule.cs
View File

@ -5,14 +5,12 @@ using JiShe.CollectBus.Host.Extensions;
using JiShe.CollectBus.Host.HealthChecks;
using JiShe.CollectBus.Host.Swaggers;
using JiShe.CollectBus.IoTDB.Options;
using JiShe.CollectBus.MongoDB;
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
using Swashbuckle.AspNetCore.SwaggerUI;
using Volo.Abp;
using Volo.Abp.AspNetCore.Authentication.JwtBearer;
using Volo.Abp.AspNetCore.Serilog;
using Volo.Abp.Autofac;
using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.Caching.StackExchangeRedis;
using Volo.Abp.Modularity;
using Volo.Abp.Swashbuckle;
@ -30,9 +28,7 @@ namespace JiShe.CollectBus.Host
typeof(AbpSwashbuckleModule),
typeof(AbpTimingModule),
typeof(CollectBusApplicationModule),
typeof(CollectBusMongoDbModule),
typeof(AbpCachingStackExchangeRedisModule),
typeof(AbpBackgroundWorkersHangfireModule)
typeof(AbpCachingStackExchangeRedisModule)
)]
public partial class CollectBusHostModule : AbpModule
{
@ -47,8 +43,8 @@ namespace JiShe.CollectBus.Host
ConfigureSwaggerServices(context, configuration);
ConfigureNetwork(context, configuration);
ConfigureJwtAuthentication(context, configuration);
ConfigureHangfire(context);
ConfigureAuditLog(context);
//ConfigureHangfire(context);
//ConfigureAuditLog(context);
ConfigureCustom(context, configuration);
ConfigureHealthChecks(context, configuration);
Configure<AbpClockOptions>(options => { options.Kind = DateTimeKind.Local; });
@ -92,10 +88,10 @@ namespace JiShe.CollectBus.Host
app.UseAuditing();
app.UseAbpSerilogEnrichers();
app.UseUnitOfWork();
app.UseHangfireDashboard("/hangfire", new DashboardOptions
{
IgnoreAntiforgeryToken = true
});
//app.UseHangfireDashboard("/hangfire", new DashboardOptions
//{
// IgnoreAntiforgeryToken = true
//});
app.UseConfiguredEndpoints(endpoints =>
{
if (!configuration.GetValue<bool>("HealthChecks:IsEnable")) return;

47
web/JiShe.CollectBus.Host/HealthChecks/CassandraHealthCheck.cs
View File

@ -1,5 +1,5 @@
using Cassandra;
using JiShe.CollectBus.Cassandra;
//using Cassandra;
//using JiShe.CollectBus.Cassandra;
using Microsoft.Extensions.Diagnostics.HealthChecks;
namespace JiShe.CollectBus.Host.HealthChecks
@ -31,27 +31,28 @@ namespace JiShe.CollectBus.Host.HealthChecks
/// </returns>
public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
{
var cassandraConfig = new CassandraConfig();
_configuration.GetSection("Cassandra").Bind(cassandraConfig);
try
{
var clusterBuilder = Cluster.Builder();
foreach (var node in cassandraConfig.Nodes)
{
clusterBuilder.AddContactPoint(node.Host)
.WithPort(node.Port);
}
clusterBuilder.WithCredentials(cassandraConfig.Username, cassandraConfig.Password);
var cluster = clusterBuilder.Build();
using var session = await cluster.ConnectAsync();
var result = await Task.FromResult(session.Execute("SELECT release_version FROM system.local"));
var version = result.First().GetValue<string>("release_version");
return HealthCheckResult.Healthy($"Cassandra is healthy. Version: {version}");
}
catch (Exception ex)
{
return new HealthCheckResult(context.Registration.FailureStatus, $"Cassandra is unhealthy: {ex.Message}", ex);
}
return HealthCheckResult.Healthy("Cassandra is unhealthy.");
//var cassandraConfig = new CassandraConfig();
//_configuration.GetSection("Cassandra").Bind(cassandraConfig);
//try
//{
// var clusterBuilder = Cluster.Builder();
// foreach (var node in cassandraConfig.Nodes)
// {
// clusterBuilder.AddContactPoint(node.Host)
// .WithPort(node.Port);
// }
// clusterBuilder.WithCredentials(cassandraConfig.Username, cassandraConfig.Password);
// var cluster = clusterBuilder.Build();
// using var session = await cluster.ConnectAsync();
// var result = await Task.FromResult(session.Execute("SELECT release_version FROM system.local"));
// var version = result.First().GetValue<string>("release_version");
// return HealthCheckResult.Healthy($"Cassandra is healthy. Version: {version}");
//}
//catch (Exception ex)
//{
// return new HealthCheckResult(context.Registration.FailureStatus, $"Cassandra is unhealthy: {ex.Message}", ex);
//}
}
}
}

1
web/JiShe.CollectBus.Host/HealthChecks/IoTDBHealthCheck.cs
View File

@ -1,5 +1,4 @@
using System.Net.Sockets;
using JiShe.CollectBus.Cassandra;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Options;
using JiShe.CollectBus.IoTDB.Provider;

6
web/JiShe.CollectBus.Host/JiShe.CollectBus.Host.csproj
View File

@ -25,7 +25,6 @@
<ItemGroup>
<PackageReference Include="AspNetCore.HealthChecks.Kafka" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.MongoDb" Version="8.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.Redis" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.UI" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.UI.Client" Version="9.0.0" />
@ -47,13 +46,9 @@
<PackageReference Include="Volo.Abp.AspNetCore.Authentication.JwtBearer" Version="8.3.3" />
<PackageReference Include="Volo.Abp.AspNetCore.Serilog" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
<PackageReference Include="Volo.Abp.BackgroundWorkers.Hangfire" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Caching.StackExchangeRedis" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Swashbuckle" Version="8.3.3" />
<!--<PackageReference Include="Hangfire.HttpJob" Version="3.8.5" />
<PackageReference Include="Hangfire.MySqlStorage" Version="2.0.3" />
<PackageReference Include="Hangfire.Dashboard.BasicAuthorization" Version="1.0.2" />-->
</ItemGroup>
<ItemGroup>
@ -61,7 +56,6 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\modules\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
<ProjectReference Include="..\..\services\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.csproj" />
<ProjectReference Include="..\..\services\JiShe.CollectBus.Application\JiShe.CollectBus.Application.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.HttpApi\JiShe.CollectBus.HttpApi.csproj" />

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

@ -140,7 +140,7 @@
}
],
"ServerApplicationOptions": {
"ServerTagName": "JiSheCollectBus99",
"ServerTagName": "JiSheCollectBus77",
"SystemType": "Energy",
"FirstCollectionTime": "2025-04-28 15:07:00",
"AutomaticVerificationTime": "16:07:00",
@ -153,5 +153,8 @@
"PlugInFolder": "",
"TCP": {
"ClientPort": 10500
},
"BackgroundJobs": {
"IsJobExecutionEnabled": false // 关闭任务执行
}
}

2
web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
View File

@ -2,7 +2,6 @@
using JiShe.CollectBus.Common;
using JiShe.CollectBus.Migration.Host.HealthChecks;
using JiShe.CollectBus.Migration.Host.Swaggers;
using JiShe.CollectBus.MongoDB;
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
using Swashbuckle.AspNetCore.SwaggerUI;
using Volo.Abp;
@ -27,7 +26,6 @@ namespace JiShe.CollectBus.Migration.Host
typeof(AbpAspNetCoreSerilogModule),
typeof(AbpSwashbuckleModule),
typeof(AbpTimingModule),
typeof(CollectBusMongoDbModule),
typeof(CollectBusMigrationApplicationModule),
typeof(AbpCachingStackExchangeRedisModule)
)]

6
web/JiShe.CollectBus.Migration.Host/JiShe.CollectBus.Migration.Host.csproj
View File

@ -21,7 +21,6 @@
<ItemGroup>
<PackageReference Include="AspNetCore.HealthChecks.Kafka" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.MongoDb" Version="8.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.Redis" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.UI" Version="9.0.0" />
<PackageReference Include="AspNetCore.HealthChecks.UI.Client" Version="9.0.0" />
@ -43,19 +42,14 @@
<PackageReference Include="Volo.Abp.AspNetCore.Authentication.JwtBearer" Version="8.3.3" />
<PackageReference Include="Volo.Abp.AspNetCore.Serilog" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
<!--<PackageReference Include="Volo.Abp.BackgroundWorkers.Hangfire" Version="8.3.3" />-->
<PackageReference Include="Volo.Abp.Caching.StackExchangeRedis" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
<PackageReference Include="Volo.Abp.Swashbuckle" Version="8.3.3" />
<!--<PackageReference Include="Hangfire.HttpJob" Version="3.8.5" />
<PackageReference Include="Hangfire.MySqlStorage" Version="2.0.3" />
<PackageReference Include="Hangfire.Dashboard.BasicAuthorization" Version="1.0.2" />-->
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\modules\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />
<ProjectReference Include="..\..\services\JiShe.CollectBus.Migration.Application.Contracts\JiShe.CollectBus.Migration.Application.Contracts.csproj" />
<ProjectReference Include="..\..\services\JiShe.CollectBus.Migration.Application\JiShe.CollectBus.Migration.Application.csproj" />
<ProjectReference Include="..\JiShe.CollectBus.Migration.HttpApi\JiShe.CollectBus.Migration.HttpApi.csproj" />