diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 5b2fc8fc..646096a3 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -5,6 +5,7 @@ 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; @@ -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( @@ -33,7 +35,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, - IServiceProvider serviceProvider + 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,30 +371,31 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_DeletedTrack(List 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 81d0c79c..ec670528 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -5,6 +5,7 @@ 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; @@ -26,6 +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 IServiceProvider _serviceProvider; public TrackUpsertEventHandler( @@ -36,7 +38,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger, - IServiceProvider serviceProvider) + IServiceProvider serviceProvider, + WarningSignalPublisher warningSignalPublisher + ) { _orderRepository = orderRepository; _capPublisher = capPublisher; @@ -46,6 +50,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _broadcastingService = broadcastingService; _logger = logger; _serviceProvider = serviceProvider; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -200,15 +205,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,56 +516,57 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List 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) { - 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/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..0583380b --- /dev/null +++ b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs @@ -0,0 +1,150 @@ +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 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(); + }); + + //bb.WithProtobufNetDeserializer(); + bb.WithJsonCoreDeserializer(bc => + bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references + .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming + .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull)); + 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(); + }); + + // 2000w条消息数据包压缩从 2g -> 1g + // bb.WithProtobufNetSerializer(); + + bb.WithJsonCoreSerializer(bc => + bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references + .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming + .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull)); + 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/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index 237fb654..075f9d94 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -11,6 +11,8 @@ + + diff --git a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs index f7e8b2b2..963268ca 100644 --- a/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs @@ -33,13 +33,18 @@ public readonly struct WarningSignal public static List FromTrack(List tracks, WarningSignalType type) { return tracks - .Select(x => new WarningSignal( - type, - x.OrderId, - x.Id, - x.TrackStatusId, - x.TrackContentId - )) + .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..46499c9d --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalKey.cs @@ -0,0 +1,44 @@ +using System.Text.Json.Serialization; + +namespace SJZY.Track.Application.Services.Warning.Signals; + +/// +/// 预警信号key，用于充当分区key +/// +public readonly struct WarningSignalKey : IEquatable +{ + public WarningSignalType Type { get; } + public int OrderId { get; } + + [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..9b5b8a26 --- /dev/null +++ b/src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalMetadata.cs @@ -0,0 +1,103 @@ +using System.Text.Json; +using System.Text.Json.Serialization; +using Confluent.Kafka; + +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; } + + [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/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 53284c7a..06e6dc9c 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -15,6 +15,7 @@ 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; @@ -214,7 +215,10 @@ public class SjzyTrackApplicationModule : AbpModule IsDefault = false, }); - // 注册后台服务 + // 注册预警计算服务 + context.Services.AddWarningCalcServices(configuration); + + // 注册预警任务后台服务 context.Services.AddHostedService(); // 注册HangFire服务 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"; }