diff --git a/src/SJZY.Track.Api/Program.cs b/src/SJZY.Track.Api/Program.cs index 8d7f64af..ecfd4dff 100644 --- a/src/SJZY.Track.Api/Program.cs +++ b/src/SJZY.Track.Api/Program.cs @@ -2,10 +2,21 @@ 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); + 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.Api/SjzyTrackApiModule.cs b/src/SJZY.Track.Api/SjzyTrackApiModule.cs index bc20b7f5..8c0ff673 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,6 +100,18 @@ public class SjzyTrackApiModule : AbpModule #region 启动提示 + // 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"); + // Console.WriteLine("libPath:"+path); + // 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.Contract/Services/IBroadcastingService.cs b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs index 238f015b..91d34dec 100644 --- a/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs +++ b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs @@ -20,9 +20,14 @@ public interface IBroadcastingService : ITransientDependency /// /// /// - /// void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode); + /// + /// 轨迹推送完成广播（推送完成回调） + /// + /// + void TrackPushCompletedPublicTopic(List list); + /// /// 轨迹推送完成广播（推送完成回调） /// diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index d07a72d4..33d65d4f 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -5,10 +5,11 @@ using SJZY.Expand.ABP.Core.PublicMethords; 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; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Extensions; using SJZY.Track.Domain.Queries; @@ -24,6 +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 IServiceProvider _serviceProvider; public TrackDeleteEventHandler( @@ -34,6 +36,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, IServiceProvider serviceProvider + //WarningSignalPublisher warningSignalPublisher ) { _provideQueries = provideQueries; @@ -43,6 +46,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _serviceProvider = serviceProvider; + //_warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] @@ -367,7 +371,17 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_DeletedTrack(List delTracks) { - var signals = TrackSignal.FromTrack(delTracks, WarningSignalType.轨迹删除); + if (_BUGOptions.EnablePref) + { + var scope = _serviceProvider.CreateScope(); + var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); + _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks); + scope.Dispose(); + return; + } + + //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks); + var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除); _ = Task.Run(async () => { diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 1892b3c2..65070319 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -5,10 +5,11 @@ using SJZY.Expand.ABP.Core.PublicMethords; 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; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; @@ -25,7 +26,10 @@ 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; public TrackUpsertEventHandler( @@ -36,7 +40,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger, - IServiceProvider serviceProvider) + IServiceProvider serviceProvider + //WarningSignalPublisher warningSignalPublisher + ) { _orderRepository = orderRepository; _capPublisher = capPublisher; @@ -46,6 +52,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _broadcastingService = broadcastingService; _logger = logger; _serviceProvider = serviceProvider; + //_warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -200,15 +207,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe // todo:引发预警计算 改MQ批量+分流 WarningCalc_AddedTrack(insertTracks); - // 添加轨迹触发预警计算（异步） - await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent - { - SubSystemId = @event.SubSystemId, - CustomId = @event.CustomId, - OrderIds = insertTracks.Select(x => x.OrderId).ToArray(), - TrackStatusId = trackContent.TrackStatusId, - TrackTime = @event.TrackTime, - }); + // 添加轨迹触发预警计算（异步）todo:弃用 + // await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent + // { + // SubSystemId = @event.SubSystemId, + // CustomId = @event.CustomId, + // OrderIds = insertTracks.Select(x => x.OrderId).ToArray(), + // TrackStatusId = trackContent.TrackStatusId, + // TrackTime = @event.TrackTime, + // }); } // 触发订单主表更新事件（异步） @@ -511,8 +518,17 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List tracks) { - var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹新增); + if (_BUGOptions.EnablePref) + { + var scope = _serviceProvider.CreateScope(); + var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); + _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks); + scope.Dispose(); + return; + } + //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks); + var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹新增); _ = Task.Run(async () => { var scope = _serviceProvider.CreateAsyncScope(); @@ -539,7 +555,17 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_UpdatedTrack(List tracks) { - var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹更新); + if (_BUGOptions.EnablePref) + { + var scope = _serviceProvider.CreateScope(); + var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); + _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks); + scope.Dispose(); + return; + } + + //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks); + var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新); _ = Task.Run(async () => { diff --git a/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs b/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs new file mode 100644 index 00000000..c1f0e04e --- /dev/null +++ b/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs @@ -0,0 +1,287 @@ +using System.Diagnostics; +using Confluent.Kafka; +using Confluent.Kafka.Core.Consumer; +using Confluent.Kafka.Core.Producer; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; +using SJZY.Expand.ABP.Core.Logger; +using SJZY.OTel; +using SJZY.Track.Application.Options; +using SJZY.Track.Application.Services.Warning.Handlers; +using SJZY.Track.Application.Services.Warning.Signals; +using SJZY.Track.Domain.Extensions; + +namespace SJZY.Track.Application.Kafka; + +/// +/// 预警计算后台任务 +/// 预警信号从kafka摄取来进行批处理 +/// +public sealed class WarningCalcBackgroundService : BackgroundService +{ + private readonly IServiceScopeFactory _serviceScopeFactory; + private readonly IKafkaConsumer _consumer; + private readonly IKafkaProducer _producer; + private readonly IKafkaProducer _dlxProducer; + private TrackWarningTaskOptions _options; + + public WarningCalcBackgroundService( + IOptionsMonitor optionsMonitor, + IServiceScopeFactory serviceScopeFactory, + IKafkaConsumer consumer, + IKafkaProducer producer, + IKafkaProducer dlxProducer + ) + { + _serviceScopeFactory = serviceScopeFactory; + _consumer = consumer; + _producer = producer; + _dlxProducer = dlxProducer; + _options = optionsMonitor.CurrentValue; + optionsMonitor.OnChange(x => _options = x); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + while (!stoppingToken.IsCancellationRequested) + { + await DoExecuteAsync(stoppingToken); + await Task.Delay(_options.WarningCalcInterval, stoppingToken); + } + } + + private Activity? _activity; + private readonly Stopwatch _stopwatch = Stopwatch.StartNew(); + + private async ValueTask DoExecuteAsync(CancellationToken ct) + { + try + { + await FetchAndHandlerAsync(ct); + } + catch (Exception e) + { + if (e is KafkaException kafkaException) + { + // kafka重启后，集群恢复需要重新选举，本地消费者存储leader节点信息可能和重新选举后的leader不一致 + if (kafkaException.Error.Code == ErrorCode.NotController) + return; + } + + LogHelper.Error(e); + _activity?.SetTag("Exception", e.Message); + } + finally + { + _activity?.SetTag("HandleTime", _stopwatch.ElapsedMilliseconds); + _activity?.Dispose(); + } + } + + private async ValueTask FetchAndHandlerAsync(CancellationToken ct) + { + // 500ms或者提取到1000条消息任意以条件到达既可返回 + var list = _consumer.ConsumeBatch(_options.WarningCalcBatchSize, TimeSpan.FromMilliseconds(500)).ToList(); + + // 没有提取到任何消息跳过 + if (list.Count == 0) return; + + _stopwatch.Restart(); + _activity = ActivityHelper.CreateSpanActivity("WarningCalcStart"); + _activity.SetTag("HandleCount", list.Count); + + // 处理消息 + await MessageHandler(list.Select(x => x.Message), ct); + + // 提交消息 + foreach (var message in list) + { + // 偏移量首先提交到内存，如果设置EnableAutoOffsetStore=true则会根据AutoCommitIntervalMs周期的提交到服务器 + _consumer.StoreOffset(message); + } + + // 提交内存中所有的偏移到服务器，适用于EnableAutoCommit为false的手动提交场景 + _consumer.Commit(); + _activity.Dispose(); + _activity = null; + } + + private async ValueTask MessageHandler(IEnumerable> messages, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + + // 1.根据类型分组信号 + var typeSignalDict = new Dictionary>(); + var needRetryMetadata = new List(); // 需要重试的消息 + var needEnqueueDlxMetadata = new List(); // 需要投递到死信队列的 + var needHandleMessage = new List>(); + foreach (var m in messages) + { + // tips:不支持的消息需要及时上报 + var mType = m.Key.Type; + var signal = m.Value; + if (!CheckSupportMessage(mType)) + { + needEnqueueDlxMetadata.Add(WarningSignalMetadata.FormMessageOrCreate(m, false, true, errorReason: $"不支持的消息类型：{mType}")); + continue; + } + + var list = typeSignalDict.GetValueOrDefault(mType, new List()); + list.Add(signal); + typeSignalDict[mType] = list; + needHandleMessage.Add(m); + } + + // tips:上报异常(类型异常) + Report_AggregateNoSupport(needEnqueueDlxMetadata); + + // 2.处理消息 + if (typeSignalDict.Count > 0) + { + var (success, error) = await TrackSignalHandler(typeSignalDict, ct); + + if (!success) + { + // 取消异常，这里只认为服务器下线时引发的，直接中断即可，等待重分配 + if (error is TaskCanceledException) + { + // 被取消不需要增加重试次数 + //needRetryMetadata.AddRange(needHandleMessage.Select(x => TrackSignalMetadata.FormMessageOrCreate(x, false, false, error: error))); + LogHelper.Warn($"处理被取消，消息会重新分配到其他Host,影响消息条数[{needHandleMessage.Count}]"); + throw error; + } + + // tips：处理消息出错，批量上报异常指明出错的详细原因 + foreach (var metadata in needHandleMessage.Select(x => WarningSignalMetadata.FormMessageOrCreate(x, true, false, error: error!))) + { + // 超过了最大的重试次数，此时应该投递到死信队列 + if (metadata.RetryCount >= _options.WarningCalcMaxRetryCount) + { + needEnqueueDlxMetadata.Add(metadata); + continue; + } + + needRetryMetadata.Add(metadata); + } + } + } + + // 3.需要重试的消息,重新投递到原队列 + if (needRetryMetadata.Count > 0) + { + _activity?.SetTag("StartPublicRetryMessage", $"开始重新投递[{needRetryMetadata.Count}]条消息"); + LogHelper.Warn($"开始重新投递[{needRetryMetadata.Count}]条消息"); + + var tasks = needRetryMetadata.Select(async metadata => + { + metadata.SetTrackSignalMessageHeader(); + await _producer.ProduceAsync(metadata.OwenMessage, CancellationToken.None); + }).ToArray(); + await Task.WhenAll(tasks); + + _activity?.SetTag("EndPublicRetryMessage", $"已重新投递[{needRetryMetadata.Count}]条消息"); + LogHelper.Warn($"已重新投递[{needRetryMetadata.Count}]条消息"); + } + + // 4.进入死信的消息，投入到死信队列，等待人工处理 + if (needEnqueueDlxMetadata.Count > 0) + { + _activity?.SetTag("StartPublicDLXMessage", $"开始投递死信[{needEnqueueDlxMetadata.Count}]条消息"); + LogHelper.Warn($"开始投递死信[{needEnqueueDlxMetadata.Count}]条消息"); + + var tasks = needEnqueueDlxMetadata.Select(async metadata => + { + metadata.SetTrackSignalMessageHeader(); + var msg = metadata.OwenMessage; + await _dlxProducer.ProduceAsync(new Message + { + Key = msg.Value.OrderId, + Value = msg.Value, + Headers = msg.Headers + }, CancellationToken.None); + }).ToArray(); + await Task.WhenAll(tasks); + + _activity?.SetTag("StartPublicDLXMessage", $"已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息"); + LogHelper.Warn($"已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息"); + } + } + + private static bool CheckSupportMessage(WarningSignalType type) + { + return type switch + { + WarningSignalType.轨迹新增 => true, + WarningSignalType.轨迹更新 => true, + WarningSignalType.轨迹删除 => true, + _ => false + }; + } + + private async Task<(bool Success, Exception? error)> TrackSignalHandler(Dictionary> typeSignalDict, CancellationToken ct) + { + var scope = _serviceScopeFactory.CreateAsyncScope(); + try + { + ct.ThrowIfCancellationRequested(); + var handler = scope.ServiceProvider.GetRequiredService(); + var handler2 = scope.ServiceProvider.GetRequiredService(); + var handler3 = scope.ServiceProvider.GetRequiredService(); + + // tips:处理失败视为整体失败，消息会被重新投递，考虑减小处理批次大小？ + var addSignalr = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹新增, Empty.List); + var updateSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹更新, Empty.List); + var deleteSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹删除, Empty.List); + + // 增 + if (addSignalr.Count > 0) + await handler.TrackAddedHandleAsync(addSignalr); + // 删 + if (deleteSignals.Count > 0) + { + await handler.TrackDeletedHandleAsync(deleteSignals); + await handler2.TrackDeletedHandleAsync(deleteSignals); + } + + // 增+改 + var addOrUpdateSignal = new List(addSignalr.Count + updateSignals.Count); + addOrUpdateSignal.AddRange(addSignalr); + addOrUpdateSignal.AddRange(updateSignals); + if (addOrUpdateSignal.Count > 0) + await handler2.TrackAddedOrUpdatedHandleAsync(addOrUpdateSignal); + + // 改+删 + var updateOrDeleteSignal = new List(updateSignals.Count + deleteSignals.Count); + updateOrDeleteSignal.AddRange(updateSignals); + updateOrDeleteSignal.AddRange(deleteSignals); + if (updateOrDeleteSignal.Count > 0) + await handler3.TrackUpdatedOrDeletedHandleAsync(updateOrDeleteSignal); + + return (true, null); + } + catch (Exception e) + { + return (false, e); + } + finally + { + await scope.DisposeAsync(); + } + } + + private void Report_AggregateNoSupport(List ms) + { + if (ms.Count == 0) return; + var reasons = string.Join(",", ms.GroupBy(x => x.LastErrorReason).Select(x => x.Key)); + _activity?.SetTag("AggregateNoSupport", reasons); + LogHelper.Error($"预警信号类型不支持：{reasons}"); + } + + public override void Dispose() + { + base.Dispose(); + _consumer.Close(); + _consumer.Dispose(); + _producer.Dispose(); + } +} diff --git a/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs new file mode 100644 index 00000000..485e6ab8 --- /dev/null +++ b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs @@ -0,0 +1,176 @@ +//#define ProtoBuf +#define NewtonsoftJson +using System.Globalization; +using System.Net; +using System.Text.Json; +using System.Text.Json.Serialization; +using Confluent.Kafka; +using Confluent.Kafka.Core.Consumer; +using Confluent.Kafka.Core.Producer; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using Newtonsoft.Json.Converters; +using Newtonsoft.Json.Serialization; +using SJZY.Expand.ABP.Core.Logger; +using SJZY.Track.Application.Helpers; +using SJZY.Track.Application.Services.Warning.Signals; + +namespace SJZY.Track.Application.Kafka; + +public static class WarningCalcServiceExtension +{ + public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf) + { + var env = "-" + conf.GetValue("Environment")!.ToLower(); + // 预警信号批处理后台任务 + services.AddHostedService(); + + // 预警信号 + RegisterConsumer(services, conf, GlobalConst.WarningKafkaGroup + env, new[] { GlobalConst.WarningKafkaTopic + env }); + RegisterProducer(services, conf, GlobalConst.WarningKafkaTopic + env); + + // 预警信号DLX + RegisterConsumer(services, conf, GlobalConst.WarningDLXKafkaGroup + env, new[] { GlobalConst.WarningDLXKafkaTopic + env }); + RegisterProducer(services, conf, GlobalConst.WarningDLXKafkaTopic + env); + } + + private static void RegisterConsumer(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic) + { + services.AddSingleton>(sp => + { + var bb = new KafkaConsumerBuilder() + .WithConsumerConfiguration(o => + { + o.WithClientId(Dns.GetHostName()) + .WithGroupId(groupId) + .WithBootstrapServers(conf.GetValue("Kafka:BootstrapServers")) + .WithAutoOffsetReset(AutoOffsetReset.Earliest) + //偏移提交控制 + .WithEnablePartitionEof(false) + .WithEnableAutoOffsetStore(false) + .WithEnableAutoCommit(false) + //链接控制 + .WithSessionTimeoutMs(conf.GetValue("Kafka:Consumer:SessionTimeoutMs")) + .WithMaxPollIntervalMs(conf.GetValue("Kafka:Consumer:MaxPollIntervalMs")) + .WithHeartbeatIntervalMs(conf.GetValue("Kafka:Consumer:HeartbeatIntervalMs")) + //吞吐量 + .WithFetchMinBytes(conf.GetValue("Kafka:Consumer:FetchMinBytes")) + .WithFetchMaxBytes(conf.GetValue("Kafka:Consumer:FetchMaxBytes")) + .WithFetchWaitMaxMs(conf.GetValue("Kafka:Consumer:FetchWaitMaxMs")) + //日志 + .WithDebug("consumer,cgrp,protocol") + .WithTopicSubscriptions(subTopic); + }); + + bb.WithLoggerFactory(sp.GetRequiredService()); + DebounceHelper.DebounceWrapper? reportFuncWrapper = null; + bb.WithErrorHandler((consumer, error) => + { + if (error.IsFatal) + { + LogHelper.Error($"Kafka致命错误${error}"); + return; + } + + // tips: 防抖以防止高频上报错误 + reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => + { + if (error.Code == ErrorCode.Local_Transport) + LogHelper.Error($"Kafka链接错误{error}"); + else if (error.Code == ErrorCode.Local_AllBrokersDown) + LogHelper.Error($"Kafka所有Broker都挂了{error}"); + }); + reportFuncWrapper.Debounce(); + }); + +#if ProtoBuf + bb.WithProtobufNetDeserializer(); + +#elif NewtonsoftJson + bb.WithNewtonsoftJsonDeserializer(builder => + builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values + .WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops + .WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names + ); +#else + bb.WithJsonCoreDeserializer(bc => + bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references + .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming + .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull) + ); +#endif + return (bb.Build() as IKafkaConsumer)!; + }); + } + + private static void RegisterProducer(IServiceCollection services, IConfiguration conf, string? defaultTopic = null) + { + services.AddSingleton>(sp => + { + var bb = new KafkaProducerBuilder() + .WithProducerConfiguration(o => + { + o.WithClientId(Dns.GetHostName()) + .WithBootstrapServers(conf.GetValue("Kafka:BootstrapServers")) + // 确保消息发送可靠 + .WithAcks(Acks.All) + .WithEnableIdempotence(true) // 幂等发送 + .WithMessageSendMaxRetries(conf.GetValue("Kafka:Producer:MessageSendMaxRetries")) + .WithRequestTimeoutMs(conf.GetValue("Kafka:Producer:RequestTimeoutMs")) + // 消息体大小 + .WithCompressionType(conf.GetValue("Kafka:Producer:CompressionType")) + .WithCompressionLevel(conf.GetValue("Kafka:Producer:CompressionLevel")) //Gzip 0-9 LZ4 0-12 数值越大压缩效果越好cpu占用越多 + // 消息发送速率控制(任意一条件先达到就发送) + .WithLingerMs(conf.GetValue("Kafka:Producer:LingerMs")) + .WithBatchSize(conf.GetValue("Kafka:Producer:BatchSize")) + .WithBatchNumMessages(conf.GetValue("Kafka:Producer:BatchNumMessages")) + // 发送缓冲内存限制 + .WithQueueBufferingMaxMessages(conf.GetValue("Kafka:Producer:QueueBufferingMaxMessages")) + .WithQueueBufferingMaxKbytes(conf.GetValue("Kafka:Producer:QueueBufferingMaxKbytes")) + // 分区器 一致性hash/随机 + .WithPartitioner(conf.GetValue("Kafka:Producer:Partitioner")) + .WithDefaultTopic(defaultTopic); + }); + + bb.WithLoggerFactory(sp.GetRequiredService()); + DebounceHelper.DebounceWrapper? reportFuncWrapper = null; + bb.WithErrorHandler((producer, error) => + { + if (error.IsFatal) + { + LogHelper.Error($"Kafka致命错误${error}"); + return; + } + + // tips: 防抖以防止高频上报错误 + reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => + { + if (error.Code == ErrorCode.Local_Transport) + LogHelper.Error($"Kafka链接错误{error}"); + else if (error.Code == ErrorCode.Local_AllBrokersDown) + LogHelper.Error($"Kafka所有Broker都挂了{error}"); + }); + reportFuncWrapper.Debounce(); + }); + +#if ProtoBuf + // 2000w条消息数据包压缩从 2g -> 1g + bb.WithProtobufNetSerializer(); +#elif NewtonsoftJson + bb.WithNewtonsoftJsonSerializer(builder => + builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values + .WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops + .WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names + ); +#else + bb.WithJsonCoreSerializer(bc => + bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references + .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming + .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull) + ); +#endif + return (bb.Build() as IKafkaProducer)!; + }); + } +} diff --git a/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs b/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs new file mode 100644 index 00000000..95e0336f --- /dev/null +++ b/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs @@ -0,0 +1,35 @@ +using Confluent.Kafka; +using Confluent.Kafka.Core.Producer; +using SJZY.Track.Application.Services.Warning.Signals; +using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.Kafka; + +/// +/// 预警信号发布者 +/// +public sealed class WarningSignalPublisher : ITransientDependency +{ + private readonly IKafkaProducer _producer; + + public WarningSignalPublisher(IKafkaProducer producer) + { + _producer = producer; + } + + public void PublishWarningSignal(WarningSignalType type, List tracks) + { + foreach (var track in tracks) + PublishWarningSignal(type, track); + } + + public void PublishWarningSignal(WarningSignalType type, TrackDetailsIndex track) + { + _producer.Produce(new Message + { + Key = new WarningSignalKey(type, track.OrderId), + Value = WarningSignal.FromTrack(track, type) + }); + } +} diff --git a/src/SJZY.Track.Application/Options/TrackWarningTaskOptions.cs b/src/SJZY.Track.Application/Options/TrackWarningTaskOptions.cs index 61ccfc0d..0c873b50 100644 --- a/src/SJZY.Track.Application/Options/TrackWarningTaskOptions.cs +++ b/src/SJZY.Track.Application/Options/TrackWarningTaskOptions.cs @@ -2,11 +2,18 @@ namespace SJZY.Track.Application.Options; public sealed class TrackWarningTaskOptions { - /// - /// 所属子系统id(业务中台id) - /// + /// 所属子系统id(业务中台id) public int MiddleSystemId { get; set; } + /// 预警计算每批处理的大小 + public int WarningCalcBatchSize { get; set; } + + /// 预警计算间隔ms + public int WarningCalcInterval { get; set; } + + /// 预警计算最大重试次数，超过这个次数会投递到死信队列当中 + public int WarningCalcMaxRetryCount { get; set; } + /// 通知任务扫描优先级 public int ScanWarningNotifyTaskLevel { get; set; } @@ -19,7 +26,7 @@ public sealed class TrackWarningTaskOptions /// 回传重试推送扫描优先级 public int ScanReturnWarningRetryPushTaskLevel { get; set; } - /// 回传重试推送任务扫描间隔（分钟 + /// 回传重试推送任务扫描间隔（分钟） public int ScanReturnWarningRetryPushTaskMinuteInterval { get; set; } /// 运输预警明细生成任务扫描优先级 diff --git a/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs b/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs new file mode 100644 index 00000000..55ff1ae3 --- /dev/null +++ b/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs @@ -0,0 +1,81 @@ +using System.Diagnostics; +using BSI.Stack.Rabbit; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.Rabbit; + +/// +/// 支持发布确认模式 +/// +public class RabbitClientBusExt : RabbitClientBus, ISingletonDependency +{ + public RabbitClientBusExt(IServiceProvider sp, ILogger logger) + : base(sp, logger, sp.GetRequiredService>().CurrentValue) + { + } + + public void PublishByConfirmSelect( + string exchange, + string routingKey, + byte[] data, + Action? options = null + ) + { + using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey); + activity?.SetTag("kind", "publish"); + var key = exchange + ":" + routingKey; + Lazy orAdd = Channels.GetOrAdd(key, (Func>) (_ => new Lazy(CreateModel))); + var basicProperties1 = orAdd.Value.CreateBasicProperties(); + + // 配置属性 + if (options != null) + options(basicProperties1); + var model = orAdd.Value; + + // 启用发布确认 + model.ConfirmSelect(); + var readOnlyMemory = (ReadOnlyMemory) data; + model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory); + model.WaitForConfirmsOrDie(); + } + + public void PublishByConfirmSelect( + string exchange, + string routingKey, + IEnumerable datas, + Action? options = null + ) + { + using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey); + activity?.SetTag("kind", "publish"); + var key = exchange + ":" + routingKey; + Lazy orAdd = Channels.GetOrAdd(key, (Func>) (_ => new Lazy(CreateModel))); + var basicProperties1 = orAdd.Value.CreateBasicProperties(); + + // 配置属性 + if (options != null) + options(basicProperties1); + var model = orAdd.Value; + + // 启用发布确认 + model.ConfirmSelect(); + foreach (var data in datas) + { + var readOnlyMemory = (ReadOnlyMemory) data; + model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory); + } + model.WaitForConfirmsOrDie(); + } + + private IConnection GetConnection() + { + var key = Connections.OrderBy((Func, int>) (q => q.Value)).Select((Func, IConnection>) (q => q.Key)).FirstOrDefault(); + Connections.AddOrUpdate(key, (Func) (_ => 1), (Func) ((_, v) => v + 1)); + return key; + } + + private IModel CreateModel() => GetConnection().CreateModel(); +} diff --git a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs index 8561d827..1ba4da4b 100644 --- a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -84,7 +84,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl // 推送广播 if (logs.First().IsToMq) - logs.ForEach(log => _broadcastingService.TrackPushCompletedPublicTopic(BuildDto(log))); + _broadcastingService.TrackPushCompletedPublicTopic(logs.Select(BuildDto).ToList()); // 写入推送日志 await _trackPushRecordEsRepository.AddManyAsync(logs); diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index 237fb654..fcb4b392 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -8,14 +8,16 @@ - + + + + + - - diff --git a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs index 7af7e83b..1723a226 100644 --- a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs +++ b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs @@ -1,9 +1,9 @@ using System.Text; -using BSI.Stack.Rabbit; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; +using SJZY.Track.Application.Rabbit; namespace SJZY.Track.Application.Services.Broadcasting; @@ -12,9 +12,9 @@ namespace SJZY.Track.Application.Services.Broadcasting; /// public class BroadcastingService : IBroadcastingService { - private readonly RabbitClientBus _rabbitClientBus; + private readonly RabbitClientBusExt _rabbitClientBus; - public BroadcastingService(RabbitClientBus rabbitClientBus) + public BroadcastingService(RabbitClientBusExt rabbitClientBus) { _rabbitClientBus = rabbitClientBus; } @@ -64,15 +64,29 @@ public class BroadcastingService : IBroadcastingService else data.Option = 3; - _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", - Encoding.UTF8.GetBytes(data.ToJson())); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", + Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; }); } public void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode) { eto.Option = 2; - _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", - Encoding.UTF8.GetBytes(eto.ToJson())); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", + Encoding.UTF8.GetBytes(eto.ToJson()), prop => { prop.Persistent = true; }); + } + + /// + /// 轨迹推送完成广播（批量） + /// + /// + /// + public void TrackPushCompletedPublicTopic(List list) + { + var datas = list + .Select(TrackPushRecordDtoToEvent) + .Select(x => Encoding.UTF8.GetBytes(x.ToJson())); + var log = list.First(); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", datas, prop => { prop.Persistent = true; }); } /// @@ -82,7 +96,24 @@ public class BroadcastingService : IBroadcastingService /// public void TrackPushCompletedPublicTopic(TrackPushRecordDto log) { - var data = new TrackPushCompletedEvent + var data = TrackPushRecordDtoToEvent(log); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", + Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; }); + } + + /// + /// 轨迹推送失败广播 + /// + /// + public void TrackPushFailedPubilcTopic(TrackPushRecordDto log) + { + //暂时使用推送完成模型，后续可能会出现有区分度的模型 + TrackPushCompletedPublicTopic(log); + } + + private static TrackPushCompletedEvent TrackPushRecordDtoToEvent(TrackPushRecordDto log) + { + return new TrackPushCompletedEvent { RequestId = log.RequestId, IsSuccess = log.IsSuccess, @@ -101,14 +132,5 @@ public class BroadcastingService : IBroadcastingService TrackCode = log.TrackCode, Extra = log.Extra }; - - _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", - Encoding.UTF8.GetBytes(data.ToJson())); - } - - public void TrackPushFailedPubilcTopic(TrackPushRecordDto log) - { - //暂时使用推送完成模型，后续可能会出现有区分度的模型 - TrackPushCompletedPublicTopic(log); } } diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs index 1d95537e..edc087ff 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs @@ -1,9 +1,9 @@ using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate.Complete; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Dtos.TrackDetails; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; @@ -31,7 +31,7 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient #region 预警信号批处理-删除轨迹 - public async Task TrackDeletedHandleAsync(List deletedSignals) + public async Task TrackDeletedHandleAsync(List deletedSignals) { // 1.只处理未删除的订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); @@ -118,7 +118,7 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient #region todo: 预警信号批处理-新增轨迹 - public async Task TrackAddedHandleAsync(List addedSignals) + public async Task TrackAddedHandleAsync(List addedSignals) { // 1.只处理未删除的订单 var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet(); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs index a9444cf2..3756c523 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs @@ -1,16 +1,16 @@ using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.WarningAggregate; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; namespace SJZY.Track.Application.Services.Warning.Handlers; public interface ICompleteWarningHandler { // 批量处理轨迹 删除信号 - Task TrackDeletedHandleAsync(List deletedSignals); + Task TrackDeletedHandleAsync(List deletedSignals); // 批量处理轨迹 新增信号 - Task TrackAddedHandleAsync(List addedSignals); + Task TrackAddedHandleAsync(List addedSignals); // 完整预警主体禁用 Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs index 91f8ee79..69843a6e 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs @@ -1,15 +1,15 @@ using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; namespace SJZY.Track.Application.Services.Warning.Handlers; public interface IReturnWarningHandler { // 批量处理轨迹 修改/删除 - Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals); + Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals); // 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传） Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List records, List pushInputs, List updateInputs); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs index 61b8c4b8..16a035a7 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs @@ -1,6 +1,6 @@ using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.WarningAggregate; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Aggregates.WarningAggregate.Transport; namespace SJZY.Track.Application.Services.Warning.Handlers; @@ -11,10 +11,10 @@ namespace SJZY.Track.Application.Services.Warning.Handlers; public interface ITransportWarningHandler { // 批量处理轨迹删除信号 - Task TrackDeletedHandleAsync(List deletedSignals); + Task TrackDeletedHandleAsync(List deletedSignals); // 批量处理轨迹新增/修改 - Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals); + Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals); // 运输预警主体禁用 Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs index 53a88be9..e8f2db19 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs @@ -3,11 +3,11 @@ using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Options; using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate.Return; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; @@ -384,7 +384,7 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe #region todo-test: 预警信号批处理-修改或删除轨迹 - public async Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals) + public async Task TrackUpdatedOrDeletedHandleAsync(List updateOrDeleteSignals) { // 1.查询订单 var orderIdSet = updateOrDeleteSignals.Select(x => x.OrderId).ToHashSet(); diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs index d5cad8ca..a472f129 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs @@ -1,11 +1,11 @@ using System.Linq.Expressions; using SJZY.Track.Application.Services.Warning.Model; +using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.KPIAreaAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackStatusAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate; -using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Aggregates.WarningAggregate.Transport; using SJZY.Track.Domain.Dtos.Forewarning; using SJZY.Track.Domain.Dtos.TrackDetails; @@ -42,7 +42,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen // tips:删除轨迹信号在运输预警处理只关注被删除的轨迹的状态 // 可能影响【运输预警明细】【运输预警明细生成任务】 - public async Task TrackDeletedHandleAsync(List deletedSignals) + public async Task TrackDeletedHandleAsync(List deletedSignals) { // 1.查询订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); @@ -199,7 +199,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen #region 预警信号批处理-新增修改轨迹 // tips:新增修改轨迹信号，在运输处理关注最新的轨迹 - public async Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals) + public async Task TrackAddedOrUpdatedHandleAsync(List addOrUpdateSignals) { // 1.查询订单 var orderIdSet = addOrUpdateSignals.Select(x => x.OrderId).ToHashSet(); diff --git a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs new file mode 100644 index 00000000..28479223 --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs @@ -0,0 +1,51 @@ +using Newtonsoft.Json; +using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; + +namespace SJZY.Track.Application.Services.Warning.Signals; + +/// +/// 预警信号，目前只是操作轨迹时产生 +/// 未来可以通过扩展类型+扩展字段的形式增加更多种预警触发源 +/// +public readonly struct WarningSignal +{ + public WarningSignalType Type { get; } + + public int OrderId { get; } + public string EsId { get; } + public int TrackStatusId { get; } + public int TrackContentId { get; } + + // public string OrderNo { get; set; } + // public int OrderType { get; set; } + // public int SubSystemId { get; set; } + + [System.Text.Json.Serialization.JsonConstructor] + [JsonConstructor] + public WarningSignal(WarningSignalType type, int orderId, string esId, int trackStatusId, int trackContentId) + { + Type = type; + OrderId = orderId; + EsId = esId; + TrackStatusId = trackStatusId; + TrackContentId = trackContentId; + } + + public static List FromTrack(List tracks, WarningSignalType type) + { + return tracks + .Select(x => FromTrack(x, type)) + .ToList(); + } + + public static WarningSignal FromTrack(TrackDetailsIndex track, WarningSignalType type) + { + return new WarningSignal( + type, + track.OrderId, + track.Id, + track.TrackStatusId, + track.TrackContentId + ); + } +} diff --git a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalKey.cs b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalKey.cs new file mode 100644 index 00000000..32d7e9b1 --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalKey.cs @@ -0,0 +1,45 @@ +using Newtonsoft.Json; + +namespace SJZY.Track.Application.Services.Warning.Signals; + +/// +/// 预警信号key，用于充当分区key +/// +public readonly struct WarningSignalKey : IEquatable +{ + public WarningSignalType Type { get; } + public int OrderId { get; } + + [System.Text.Json.Serialization.JsonConstructor] + [JsonConstructor] + public WarningSignalKey(WarningSignalType type, int orderId) + { + Type = type; + OrderId = orderId; + } + + public bool Equals(WarningSignalKey other) + { + return Type == other.Type && OrderId == other.OrderId; + } + + public override bool Equals(object? obj) + { + return obj is WarningSignalKey other && Equals(other); + } + + public override int GetHashCode() + { + return HashCode.Combine((int) Type, OrderId); + } + + public static bool operator ==(WarningSignalKey left, WarningSignalKey right) + { + return left.Equals(right); + } + + public static bool operator !=(WarningSignalKey left, WarningSignalKey right) + { + return !(left == right); + } +} diff --git a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalMetadata.cs b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalMetadata.cs new file mode 100644 index 00000000..41f39e4b --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalMetadata.cs @@ -0,0 +1,104 @@ +using Confluent.Kafka; +using Newtonsoft.Json; +using JsonSerializer = System.Text.Json.JsonSerializer; + +namespace SJZY.Track.Application.Services.Warning.Signals; + +/// +/// 预警信号元数据 +/// +public struct WarningSignalMetadata : IEquatable +{ + public const string TrackSignalMetadataHeaderKey = "track_signal_metadata"; + + /// 已经重试的次数，被重新入队时+1 + public int RetryCount { get; set; } + + /// 是否是无法处理的内容，需要直接报告异常 + public bool Unhandled { get; set; } + + /// 最后一次出错的原因 + public string LastErrorReason { get; set; } + + [System.Text.Json.Serialization.JsonIgnore] + [JsonIgnore] + public Message OwenMessage { get; set; } + + public WarningSignalMetadata() + { + RetryCount = 0; + LastErrorReason = string.Empty; + Unhandled = false; + OwenMessage = null!; + } + + public void SetTrackSignalMessageHeader() + { + if (OwenMessage == null) + throw new InvalidOperationException("OwenMessage is null!"); + OwenMessage.Headers.Add(TrackSignalMetadataHeaderKey, ToBytes()); + } + + // 从消息提取或者构建一个元数据并为其设置属性 + public static WarningSignalMetadata FormMessageOrCreate( + Message msg, + bool retryInc, + bool unhandled, + string? errorReason = null, + Exception? error = null + ) + { + msg.Headers.TryGetLastBytes(TrackSignalMetadataHeaderKey, out var metadataBytes); + + var m = metadataBytes == null + ? new WarningSignalMetadata() + : Form(metadataBytes); + + m.OwenMessage = msg; + m.RetryCount = retryInc ? m.RetryCount + 1 : m.RetryCount; + m.LastErrorReason = errorReason ?? error?.Message ?? string.Empty; + m.Unhandled = unhandled || error is NotSupportedException; + + return m; + } + + public static WarningSignalMetadata Form(byte[] bytes) + { + return JsonSerializer.Deserialize(bytes); + } + + public byte[] ToBytes() + { + return JsonSerializer.SerializeToUtf8Bytes(this); + } + + public static implicit operator byte[](WarningSignalMetadata metadata) + { + return metadata.ToBytes(); + } + + public static bool operator ==(WarningSignalMetadata left, WarningSignalMetadata right) + { + return left.Equals(right); + } + + public static bool operator !=(WarningSignalMetadata left, WarningSignalMetadata right) + { + return !(left == right); + } + + public bool Equals(WarningSignalMetadata other) + { + return RetryCount == other.RetryCount && LastErrorReason == other.LastErrorReason && OwenMessage == other.OwenMessage; + } + + public override bool Equals(object? obj) + { + return obj is WarningSignalMetadata other && Equals(other); + } + + public override int GetHashCode() + { + return HashCode.Combine(RetryCount, LastErrorReason); + } +} diff --git a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalType.cs b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalType.cs new file mode 100644 index 00000000..5e93398d --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalType.cs @@ -0,0 +1,15 @@ +namespace SJZY.Track.Application.Services.Warning.Signals; + +/// +/// 预警信号类型 +/// +public enum WarningSignalType +{ + 未知的 = 0, + + 轨迹新增 = 1, + + 轨迹删除 = 2, + + 轨迹更新 = 3, +} diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 53284c7a..a3020ec5 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -15,14 +15,13 @@ using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.CAP; using SJZY.Track.Application.Contract; using SJZY.Track.Application.Hangfire; +using SJZY.Track.Application.Kafka; using SJZY.Track.Application.Localization; using SJZY.Track.Application.Options; using SJZY.Track.Application.Track2InterfaceRelay; using SJZY.UserCenter.Contracts; using SJZY.UserCenter.SharedModel; using SJZY.UserCenter.SharedModel.Pub; -using StackExchange.Redis.Extensions.Core.Configuration; -using StackExchange.Redis.Extensions.System.Text.Json; using Volo.Abp.AspNetCore.Mvc; using Volo.Abp.AutoMapper; using Volo.Abp.FluentValidation; @@ -140,8 +139,7 @@ public class SjzyTrackApplicationModule : AbpModule Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列"); } - - context.Services.AddRabbitBoot(options => + Action action = options => { options.ConnectionString = $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; @@ -180,7 +178,10 @@ public class SjzyTrackApplicationModule : AbpModule } } }; - }, typeof(SjzyTrackApplicationModule).Assembly); + }; + + context.Services.Configure(action); + context.Services.AddRabbitBoot(action, typeof(SjzyTrackApplicationModule).Assembly); context.Services.Configure(configuration.GetSection("TrackApp")); context.Services.Configure( @@ -200,21 +201,12 @@ public class SjzyTrackApplicationModule : AbpModule $",password={configuration["Redis:Password"]}" + $",defaultDatabase={configuration["Redis:Database"]}" + $",abortConnect=false"; - // 注册Redis服务 - context.Services.AddStackExchangeRedisExtensions(new RedisConfiguration - { - KeyPrefix = configuration["Redis:Prefix"]!, - Database = int.Parse(configuration["Redis:Database"]!), - ConnectionString = connectionString, - AbortOnConnectFail = false, - AllowAdmin = true, - //ConnectTimeout = 5000, - PoolSize = 5, - ConnectRetry = 3, - IsDefault = false, - }); - // 注册后台服务 + // 注册预警计算服务 + if (_BUGOptions.EnablePref) + context.Services.AddWarningCalcServices(configuration); + + // 注册预警任务后台服务 context.Services.AddHostedService(); // 注册HangFire服务 diff --git a/src/SJZY.Track.Application/_BUGOptions.cs b/src/SJZY.Track.Application/_BUGOptions.cs new file mode 100644 index 00000000..3418cbe9 --- /dev/null +++ b/src/SJZY.Track.Application/_BUGOptions.cs @@ -0,0 +1,6 @@ +namespace SJZY.Track.Application; + +public static class _BUGOptions +{ + public static bool EnablePref = false; +} diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs deleted file mode 100644 index 00389cd9..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs +++ /dev/null @@ -1,43 +0,0 @@ -using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; - -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -/// -/// 轨迹信号，操作轨迹时产生 -/// 用于进行预警计算 -/// -public readonly struct TrackSignal -{ - public WarningSignalType Type { get; } - - public int OrderId { get; } - // public string OrderNo { get; set; } - // public int OrderType { get; set; } - // public int SubSystemId { get; set; } - - public string EsId { get; } - public int TrackStatusId { get; } - public int TrackContentId { get; } - - public TrackSignal(WarningSignalType type, int orderId, string esId, int trackStatusId, int trackContentId) - { - Type = type; - OrderId = orderId; - EsId = esId; - TrackStatusId = trackStatusId; - TrackContentId = trackContentId; - } - - public static List FromTrack(List tracks, WarningSignalType type) - { - return tracks - .Select(x => new TrackSignal( - type, - x.OrderId, - x.Id, - x.TrackStatusId, - x.TrackContentId - )) - .ToList(); - } -} diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs deleted file mode 100644 index 31f8dec1..00000000 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; - -public enum WarningSignalType -{ - 轨迹新增 = 0, - - 轨迹删除 = 1, - - 轨迹更新 = 2, -} diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 93b5b277..3fab82d2 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -61,4 +61,25 @@ public partial class GlobalConst /// 轨迹推送通知交换机 /// public const string TrackPushExchange = "sjzy.track.on_track_push_ex"; + + // todo:以下需要在kafka创建主题 + /// + /// 预警信号kafka 主题 + /// + public const string WarningKafkaTopic = "sjzy_trackapi_warning"; + + /// + /// 预警信号消费组 + /// + public const string WarningKafkaGroup = "sjzy_trackapi_warning_group"; + + /// + /// 预警信号 DLX kafka 主题 + /// + public const string WarningDLXKafkaTopic = "sjzy_trackapi_warning_dlx"; + + /// + /// 预警信号 DLX kafka消费组 + /// + public const string WarningDLXKafkaGroup = "sjzy_trackapi_warning_dlx_group"; } diff --git a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs index c4933fcd..e2ef66ee 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs @@ -223,14 +223,14 @@ public class OrderRepository : BasicRepository, IOrderRepository, I x.OrderTypeName = typeDisplayDict.TryGetValue(GlobalConst.MainDataKey_OrderType, out var orderTypeDisplayDict) ? orderTypeDisplayDict.TryGetValue(x.OrderType, out var orderTypeDisplay) ? orderTypeDisplay - : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}] " - : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}] EnumKey:[{x.OrderType}]"; + : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}],EnumKey:[{x.OrderType}]" + : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}]"; x.QuestionTypeName = typeDisplayDict.TryGetValue(GlobalConst.MainDataKey_TrackQuestionType, out var questDisplayDict) ? questDisplayDict.TryGetValue(x.QuestionType, out var questDisplay) ? questDisplay - : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}]" - : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_TrackQuestionType}] EnumKey:[{x.OrderType}]"; + : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_OrderType}],EnumKey:[{x.QuestionType}]" + : $"未在主数据中心找到Property:[{GlobalConst.MainDataKey_TrackQuestionType}]"; }); return pageList;