diff --git a/src/SJZY.Track.Api/SjzyTrackApiModule.cs b/src/SJZY.Track.Api/SjzyTrackApiModule.cs index 7a6b5e1a..9413b8fb 100644 --- a/src/SJZY.Track.Api/SjzyTrackApiModule.cs +++ b/src/SJZY.Track.Api/SjzyTrackApiModule.cs @@ -1,3 +1,5 @@ +using System.Reflection; +using System.Runtime.InteropServices; using FluentValidation.AspNetCore; using Hangfire; using Microsoft.AspNetCore.Mvc; @@ -98,7 +100,17 @@ public class SjzyTrackApiModule : AbpModule #region 启动提示 - Console.WriteLine("\r\nKafka.LibVersion:" + Confluent.Kafka.Library.VersionString); + var isLinux = RuntimeInformation.IsOSPlatform(OSPlatform.Linux); + var arch = RuntimeInformation.ProcessArchitecture; + var os = isLinux ? "linux-" : "win-"; + var rid = os + arch; + Console.WriteLine("\r\nrid:" + rid); + // 获取当前程序集路径 + var assemblyPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location); + var path = Path.Combine(assemblyPath, "runtimes", rid, "native", isLinux ? "librdkafka.so" : "librdkafka.dll"); + var loadSuccess = Confluent.Kafka.Library.Load(path); + Console.WriteLine($"KafkaLib 加载{loadSuccess}"); + Console.WriteLine("Kafka.LibVersion:" + Confluent.Kafka.Library.VersionString); Console.WriteLine($@" +-------------------------------------------------------+ | | diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 16c4af9b..646096a3 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -25,7 +25,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( @@ -35,8 +35,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + IServiceProvider serviceProvider, + WarningSignalPublisher warningSignalPublisher ) { _provideQueries = provideQueries; @@ -46,7 +46,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] @@ -371,31 +371,31 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_DeletedTrack(List delTracks) { - //_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.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(); + // } + // }); } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index eccd7bbe..ec670528 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -27,7 +27,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private readonly IBroadcastingService _broadcastingService; private readonly ILogger _logger; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - //private readonly WarningSignalPublisher _warningSignalPublisher; + private readonly WarningSignalPublisher _warningSignalPublisher; private readonly IServiceProvider _serviceProvider; public TrackUpsertEventHandler( @@ -38,8 +38,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + IServiceProvider serviceProvider, + WarningSignalPublisher warningSignalPublisher ) { _orderRepository = orderRepository; @@ -50,7 +50,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _broadcastingService = broadcastingService; _logger = logger; _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -516,57 +516,57 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List tracks) { - //_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.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(); + // } + // }); } private void WarningCalc_UpdatedTrack(List tracks) { - //_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.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(); + // } + // }); } } diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index f946830b..b24448c9 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -200,7 +200,7 @@ public class SjzyTrackApplicationModule : AbpModule $",abortConnect=false"; // 注册预警计算服务 - //context.Services.AddWarningCalcServices(configuration); + context.Services.AddWarningCalcServices(configuration); // 注册预警任务后台服务 context.Services.AddHostedService();