clean:清理原本地异步模式的预警计算代码

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()

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 WarningSignalPublisher _warningSignalPublisher;
-    private readonly IServiceProvider _serviceProvider;
 
     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<TrackDetailsIndex> delTracks)
     {
-        if (_BUGOptions.EnablePref)
+        _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks);
-        {
-            var scope = _serviceProvider.CreateScope();
-            var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
-            _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<ICompleteWarningHandler>();
-                var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
-                var handler3 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
-
-                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();
-            }
-        });
     }
 }

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<TrackUpsertEventHandler> _logger;
-
     private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
-
+    private readonly WarningSignalPublisher _warningSignalPublisher;
-    //private readonly WarningSignalPublisher _warningSignalPublisher;
-    private readonly IServiceProvider _serviceProvider;
 
     public TrackUpsertEventHandler(
         IOrderRepository orderRepository,
@@ -40,8 +36,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
         IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
         IBroadcastingService broadcastingService,
         ILogger<TrackUpsertEventHandler> 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
+                    var mapCode = (await _provideQueries.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode))
-                            .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<TrackDetailsIndex> tracks)
     {
-        if (_BUGOptions.EnablePref)
+        _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, tracks);
-        {
-            var scope = _serviceProvider.CreateScope();
-            var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
-            _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<ICompleteWarningHandler>();
-                var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
-
-                await Task.Delay(1500); //确保落盘
-                await handler.TrackAddedHandleAsync(signals);
-                await handler2.TrackAddedOrUpdatedHandleAsync(signals);
-            }
-            catch (Exception ex)
-            {
-                LogHelper.Error(ex);
-            }
-            finally
-            {
-                await scope.DisposeAsync();
-            }
-        });
     }
 
 
     private void WarningCalc_UpdatedTrack(List<TrackDetailsIndex> tracks)
     {
-        if (_BUGOptions.EnablePref)
+        _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks);
-        {
-            var scope = _serviceProvider.CreateScope();
-            var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
-            _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<ITransportWarningHandler>();
-                var handler2 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
-
-                await Task.Delay(1500); //确保落盘
-                await handler.TrackAddedOrUpdatedHandleAsync(signals);
-                await handler2.TrackUpdatedOrDeletedHandleAsync(signals);
-            }
-            catch (Exception ex)
-            {
-                LogHelper.Error(ex);
-            }
-            finally
-            {
-                await scope.DisposeAsync();
-            }
-        });
     }
 }

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<string>("Environment")!.ToLower();
         // 预警信号批处理后台任务
         services.AddHostedService<WarningCalcBackgroundService>();
@@ -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<T> : ISerializer<T>, IDeserializer<T>

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<WarningTaskComponentBackgroundService>();

src/SJZY.Track.Application/_BUGOptions.cs

@@ -1,6 +0,0 @@
-namespace SJZY.Track.Application;
-
-public static class _BUGOptions
-{
-    public static bool EnablePref = false;
-}