diff --git a/src/SJZY.Track.Api/Program.cs b/src/SJZY.Track.Api/Program.cs index 4997c503..8d7f64af 100644 --- a/src/SJZY.Track.Api/Program.cs +++ b/src/SJZY.Track.Api/Program.cs @@ -2,22 +2,10 @@ using SJZY.Expand.ABP.Core.Configuration; using SJZY.Expand.ABP.Core.Configuration.AutoRegistOptions; using SJZY.Expand.ABP.Core.Logger; using SJZY.Track.Api; -using SJZY.Track.Application; try { var builder = WebApplication.CreateBuilder(args); - _BUGOptions.EnablePref = true; - if (args.Length > 0) - { - _BUGOptions.EnablePref = "on".Equals(args[0]); - Console.WriteLine($"是否启动优化模式：{_BUGOptions.EnablePref}"); - if (_BUGOptions.EnablePref) - { - var loadSuccess = Confluent.Kafka.Library.Load(null); - Console.WriteLine($"加载{loadSuccess},版本号：{Confluent.Kafka.Library.VersionString}"); - } - } builder .AddConfigurationContainer() .AddOptionsFromConfiguration() diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index f1c7ad81..3b607dd1 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -6,7 +6,6 @@ using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Kafka; -using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -25,8 +24,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; - // private readonly WarningSignalPublisher _warningSignalPublisher; - private readonly IServiceProvider _serviceProvider; + private readonly WarningSignalPublisher _warningSignalPublisher; public TrackDeleteEventHandler( IProvideQueries provideQueries, @@ -35,8 +33,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + WarningSignalPublisher warningSignalPublisher ) { _provideQueries = provideQueries; @@ -45,8 +42,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _trackDetailsEsRepository = trackDetailsEsRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; - _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] @@ -368,43 +364,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe } } - private void WarningCalc_DeletedTrack(List delTracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks); - var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除); - - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - var handler3 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackDeletedHandleAsync(signals); - await handler2.TrackDeletedHandleAsync(signals); - await handler3.TrackUpdatedOrDeletedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 11cd7f91..61cd543b 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -6,7 +6,6 @@ using SJZY.OTel; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Kafka; -using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -26,11 +25,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private readonly ILogger _logger; - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - - //private readonly WarningSignalPublisher _warningSignalPublisher; - private readonly IServiceProvider _serviceProvider; + private readonly WarningSignalPublisher _warningSignalPublisher; public TrackUpsertEventHandler( IOrderRepository orderRepository, @@ -40,8 +36,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + WarningSignalPublisher warningSignalPublisher ) { _orderRepository = orderRepository; @@ -51,8 +46,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _logger = logger; - _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -102,11 +96,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) { - var mapCode = (await _provideQueries - .MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) + var mapCode = (await _provideQueries.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) .FirstOrDefault(); - if (mapCode == default) + if (mapCode == null) { // 跨子系统没做映射的轨迹不处理 continue; @@ -518,75 +511,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List tracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, tracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks); - var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹新增); - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackAddedHandleAsync(signals); - await handler2.TrackAddedOrUpdatedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, tracks); } private void WarningCalc_UpdatedTrack(List tracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks); - var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新); - - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackAddedOrUpdatedHandleAsync(signals); - await handler2.TrackUpdatedOrDeletedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); } } diff --git a/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs index 1d6ec63c..8d9f2cf7 100644 --- a/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs +++ b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs @@ -15,6 +15,7 @@ public static class WarningCalcServiceExtension { public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf) { + PrintKafkaLibVersion(); var env = "-" + conf.GetValue("Environment")!.ToLower(); // 预警信号批处理后台任务 services.AddHostedService(); @@ -163,6 +164,20 @@ public static class WarningCalcServiceExtension return list; } + + public static void PrintKafkaLibVersion() + { + var loadSuccess = Library.Load(null); + Console.WriteLine($@" + +-------------------------------------------------------+ + | | + | LibrdKafka | + | 加载：{loadSuccess}，版本号：{Library.VersionString} | + | | + +-------------------------------------------------------+ + "); + + } } internal sealed class JsonCoreSerializer : ISerializer, IDeserializer diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index a3020ec5..5d3c8f99 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -203,8 +203,7 @@ public class SjzyTrackApplicationModule : AbpModule $",abortConnect=false"; // 注册预警计算服务 - if (_BUGOptions.EnablePref) - context.Services.AddWarningCalcServices(configuration); + context.Services.AddWarningCalcServices(configuration); // 注册预警任务后台服务 context.Services.AddHostedService(); diff --git a/src/SJZY.Track.Application/_BUGOptions.cs b/src/SJZY.Track.Application/_BUGOptions.cs deleted file mode 100644 index 3418cbe9..00000000 --- a/src/SJZY.Track.Application/_BUGOptions.cs +++ /dev/null @@ -1,6 +0,0 @@ -namespace SJZY.Track.Application; - -public static class _BUGOptions -{ - public static bool EnablePref = false; -}