ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf:接入kafka优化预警计算批处理，使其具有高吞吐高效处理的优点

Browse Source
This commit is contained in:
hardstylewyl 2025-05-12 15:45:49 +08:00
parent aed09e5347
commit 5dd423f136
12 changed files with 763 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -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<TrackDetailsIndex> delTracks)
{
var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除);
_ = Task.Run(async () =>
{
var scope = _serviceProvider.CreateAsyncScope();
try
{
var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
var handler3 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
await Task.Delay(1500); //确保落盘
await handler.TrackDeletedHandleAsync(signals);
await handler2.TrackDeletedHandleAsync(signals);
await handler3.TrackUpdatedOrDeletedHandleAsync(signals);
}
catch (Exception ex)
{
LogHelper.Error(ex);
}
finally
{
await scope.DisposeAsync();
}
});
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks);
// var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除);
//
// _ = Task.Run(async () =>
// {
// var scope = _serviceProvider.CreateAsyncScope();
// try
// {
// var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
// var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
// var handler3 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
//
// await Task.Delay(1500); //确保落盘
// await handler.TrackDeletedHandleAsync(signals);
// await handler2.TrackDeletedHandleAsync(signals);
// await handler3.TrackUpdatedOrDeletedHandleAsync(signals);
// }
// catch (Exception ex)
// {
// LogHelper.Error(ex);
// }
// finally
// {
// await scope.DisposeAsync();
// }
// });
}
}

118
src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -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<TrackUpsertEventHandler> _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<TrackUpsertEventHandler> 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<TrackDetailsIndex> tracks)
{
var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹新增);
_ = Task.Run(async () =>
{
var scope = _serviceProvider.CreateAsyncScope();
try
{
var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
await Task.Delay(1500); //确保落盘
await handler.TrackAddedHandleAsync(signals);
await handler2.TrackAddedOrUpdatedHandleAsync(signals);
}
catch (Exception ex)
{
LogHelper.Error(ex);
}
finally
{
await scope.DisposeAsync();
}
});
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks);
// var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹新增);
// _ = Task.Run(async () =>
// {
// var scope = _serviceProvider.CreateAsyncScope();
// try
// {
// var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
// var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
//
// await Task.Delay(1500); //确保落盘
// await handler.TrackAddedHandleAsync(signals);
// await handler2.TrackAddedOrUpdatedHandleAsync(signals);
// }
// catch (Exception ex)
// {
// LogHelper.Error(ex);
// }
// finally
// {
// await scope.DisposeAsync();
// }
// });
}
private void WarningCalc_UpdatedTrack(List<TrackDetailsIndex> tracks)
{
var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新);
_ = Task.Run(async () =>
{
var scope = _serviceProvider.CreateAsyncScope();
try
{
var handler = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
var handler2 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
await Task.Delay(1500); //确保落盘
await handler.TrackAddedOrUpdatedHandleAsync(signals);
await handler2.TrackUpdatedOrDeletedHandleAsync(signals);
}
catch (Exception ex)
{
LogHelper.Error(ex);
}
finally
{
await scope.DisposeAsync();
}
});
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks);
// var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新);
//
// _ = Task.Run(async () =>
// {
// var scope = _serviceProvider.CreateAsyncScope();
// try
// {
// var handler = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
// var handler2 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
//
// await Task.Delay(1500); //确保落盘
// await handler.TrackAddedOrUpdatedHandleAsync(signals);
// await handler2.TrackUpdatedOrDeletedHandleAsync(signals);
// }
// catch (Exception ex)
// {
// LogHelper.Error(ex);
// }
// finally
// {
// await scope.DisposeAsync();
// }
// });
}
}

287
src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs Normal file
View File

@ -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;
/// <summary>
/// 预警计算后台任务
/// 预警信号从kafka摄取来进行批处理
/// </summary>
public sealed class WarningCalcBackgroundService : BackgroundService
{
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IKafkaConsumer<WarningSignalKey, WarningSignal> _consumer;
private readonly IKafkaProducer<WarningSignalKey, WarningSignal> _producer;
private readonly IKafkaProducer<int, WarningSignal> _dlxProducer;
private TrackWarningTaskOptions _options;
public WarningCalcBackgroundService(
IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor,
IServiceScopeFactory serviceScopeFactory,
IKafkaConsumer<WarningSignalKey, WarningSignal> consumer,
IKafkaProducer<WarningSignalKey, WarningSignal> producer,
IKafkaProducer<int, WarningSignal> 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<Message<WarningSignalKey, WarningSignal>> messages, CancellationToken ct)
{
ct.ThrowIfCancellationRequested();
// 1.根据类型分组信号
var typeSignalDict = new Dictionary<WarningSignalType, List<WarningSignal>>();
var needRetryMetadata = new List<WarningSignalMetadata>(); // 需要重试的消息
var needEnqueueDlxMetadata = new List<WarningSignalMetadata>(); // 需要投递到死信队列的
var needHandleMessage = new List<Message<WarningSignalKey, WarningSignal>>();
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<WarningSignal>());
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<int, WarningSignal>
{
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<WarningSignalType, List<WarningSignal>> typeSignalDict, CancellationToken ct)
{
var scope = _serviceScopeFactory.CreateAsyncScope();
try
{
ct.ThrowIfCancellationRequested();
var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
var handler3 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
// tips:处理失败视为整体失败，消息会被重新投递，考虑减小处理批次大小？
var addSignalr = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹新增, Empty<WarningSignal>.List);
var updateSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹更新, Empty<WarningSignal>.List);
var deleteSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹删除, Empty<WarningSignal>.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<WarningSignal>(addSignalr.Count + updateSignals.Count);
addOrUpdateSignal.AddRange(addSignalr);
addOrUpdateSignal.AddRange(updateSignals);
if (addOrUpdateSignal.Count > 0)
await handler2.TrackAddedOrUpdatedHandleAsync(addOrUpdateSignal);
// 改+删
var updateOrDeleteSignal = new List<WarningSignal>(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<WarningSignalMetadata> 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();
}
}

150
src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs Normal file
View File

@ -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<string>("Environment")!.ToLower();
// 预警信号批处理后台任务
services.AddHostedService<WarningCalcBackgroundService>();
// 预警信号
RegisterConsumer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaGroup + env, new[] { GlobalConst.WarningKafkaTopic + env });
RegisterProducer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaTopic + env);
// 预警信号DLX
RegisterConsumer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaGroup + env, new[] { GlobalConst.WarningDLXKafkaTopic + env });
RegisterProducer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaTopic + env);
}
private static void RegisterConsumer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic)
{
services.AddSingleton<IKafkaConsumer<TKey, TValue>>(sp =>
{
var bb = new KafkaConsumerBuilder<TKey, TValue>()
.WithConsumerConfiguration(o =>
{
o.WithClientId(Dns.GetHostName())
.WithGroupId(groupId)
.WithBootstrapServers(conf.GetValue<string>("Kafka:BootstrapServers"))
.WithAutoOffsetReset(AutoOffsetReset.Earliest)
//偏移提交控制
.WithEnablePartitionEof(false)
.WithEnableAutoOffsetStore(false)
.WithEnableAutoCommit(false)
//链接控制
.WithSessionTimeoutMs(conf.GetValue<int>("Kafka:Consumer:SessionTimeoutMs"))
.WithMaxPollIntervalMs(conf.GetValue<int>("Kafka:Consumer:MaxPollIntervalMs"))
.WithHeartbeatIntervalMs(conf.GetValue<int>("Kafka:Consumer:HeartbeatIntervalMs"))
//吞吐量
.WithFetchMinBytes(conf.GetValue<int>("Kafka:Consumer:FetchMinBytes"))
.WithFetchMaxBytes(conf.GetValue<int>("Kafka:Consumer:FetchMaxBytes"))
.WithFetchWaitMaxMs(conf.GetValue<int>("Kafka:Consumer:FetchWaitMaxMs"))
//日志
.WithDebug("consumer,cgrp,protocol")
.WithTopicSubscriptions(subTopic);
});
bb.WithLoggerFactory(sp.GetRequiredService<ILoggerFactory>());
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<TKey, TValue>)!;
});
}
private static void RegisterProducer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string? defaultTopic = null)
{
services.AddSingleton<IKafkaProducer<TKey, TValue>>(sp =>
{
var bb = new KafkaProducerBuilder<TKey, TValue>()
.WithProducerConfiguration(o =>
{
o.WithClientId(Dns.GetHostName())
.WithBootstrapServers(conf.GetValue<string>("Kafka:BootstrapServers"))
// 确保消息发送可靠
.WithAcks(Acks.All)
.WithEnableIdempotence(true) // 幂等发送
.WithMessageSendMaxRetries(conf.GetValue<int>("Kafka:Producer:MessageSendMaxRetries"))
.WithRequestTimeoutMs(conf.GetValue<int>("Kafka:Producer:RequestTimeoutMs"))
// 消息体大小
.WithCompressionType(conf.GetValue<CompressionType>("Kafka:Producer:CompressionType"))
.WithCompressionLevel(conf.GetValue<int>("Kafka:Producer:CompressionLevel")) //Gzip 0-9 LZ4 0-12 数值越大压缩效果越好cpu占用越多
// 消息发送速率控制(任意一条件先达到就发送)
.WithLingerMs(conf.GetValue<int>("Kafka:Producer:LingerMs"))
.WithBatchSize(conf.GetValue<int>("Kafka:Producer:BatchSize"))
.WithBatchNumMessages(conf.GetValue<int>("Kafka:Producer:BatchNumMessages"))
// 发送缓冲内存限制
.WithQueueBufferingMaxMessages(conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxMessages"))
.WithQueueBufferingMaxKbytes(conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxKbytes"))
// 分区器 一致性hash/随机
.WithPartitioner(conf.GetValue<Partitioner>("Kafka:Producer:Partitioner"))
.WithDefaultTopic(defaultTopic);
});
bb.WithLoggerFactory(sp.GetRequiredService<ILoggerFactory>());
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<TKey, TValue>)!;
});
}
}

35
src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs Normal file
View File

@ -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;
/// <summary>
/// 预警信号发布者
/// </summary>
public sealed class WarningSignalPublisher : ITransientDependency
{
private readonly IKafkaProducer<WarningSignalKey, WarningSignal> _producer;
public WarningSignalPublisher(IKafkaProducer<WarningSignalKey, WarningSignal> producer)
{
_producer = producer;
}
public void PublishWarningSignal(WarningSignalType type, List<TrackDetailsIndex> tracks)
{
foreach (var track in tracks)
PublishWarningSignal(type, track);
}
public void PublishWarningSignal(WarningSignalType type, TrackDetailsIndex track)
{
_producer.Produce(new Message<WarningSignalKey, WarningSignal>
{
Key = new WarningSignalKey(type, track.OrderId),
Value = WarningSignal.FromTrack(track, type)
});
}
}

15
src/SJZY.Track.Application/Options/TrackWarningTaskOptions.cs
View File

@ -2,11 +2,18 @@ namespace SJZY.Track.Application.Options;
public sealed class TrackWarningTaskOptions
{
/// <summary>
/// 所属子系统id(业务中台id)
/// </summary>
/// <summary> 所属子系统id(业务中台id) </summary>
public int MiddleSystemId { get; set; }
/// <summary> 预警计算每批处理的大小 </summary>
public int WarningCalcBatchSize { get; set; }
/// <summary> 预警计算间隔ms </summary>
public int WarningCalcInterval { get; set; }
/// <summary> 预警计算最大重试次数，超过这个次数会投递到死信队列当中 </summary>
public int WarningCalcMaxRetryCount { get; set; }
/// <summary> 通知任务扫描优先级 </summary>
public int ScanWarningNotifyTaskLevel { get; set; }
@ -19,7 +26,7 @@ public sealed class TrackWarningTaskOptions
/// <summary> 回传重试推送扫描优先级 </summary>
public int ScanReturnWarningRetryPushTaskLevel { get; set; }
/// <summary> 回传重试推送任务扫描间隔（分钟 </summary>
/// <summary> 回传重试推送任务扫描间隔（分钟） </summary>
public int ScanReturnWarningRetryPushTaskMinuteInterval { get; set; }
/// <summary> 运输预警明细生成任务扫描优先级 </summary>

2
src/SJZY.Track.Application/SJZY.Track.Application.csproj
View File

@ -11,6 +11,8 @@
<PackageReference Include="BSI.Stack.Rabbit" Version="2.1.3" />
<PackageReference Include="Hangfire.AspNetCore" Version="1.8.18" />
<PackageReference Include="Hangfire.Redis.StackExchange" Version="1.9.4" />
<PackageReference Include="Confluent.Kafka.Core" Version="1.0.5" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.JsonCore" Version="1.0.5" />
<PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" Version="8.2.1"/>
<PackageReference Include="SJZY.Expand.AliyunOSS" Version="7.0.17" />
<PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.5-alpha2" />

19
src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs
View File

@ -33,13 +33,18 @@ public readonly struct WarningSignal
public static List<WarningSignal> FromTrack(List<TrackDetailsIndex> 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
);
}
}

44
src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalKey.cs Normal file
View File

@ -0,0 +1,44 @@
using System.Text.Json.Serialization;
namespace SJZY.Track.Application.Services.Warning.Signals;
/// <summary>
/// 预警信号key，用于充当分区key
/// </summary>
public readonly struct WarningSignalKey : IEquatable<WarningSignalKey>
{
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);
}
}

103
src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalMetadata.cs Normal file
View File

@ -0,0 +1,103 @@
using System.Text.Json;
using System.Text.Json.Serialization;
using Confluent.Kafka;
namespace SJZY.Track.Application.Services.Warning.Signals;
/// <summary>
/// 预警信号元数据
/// </summary>
public struct WarningSignalMetadata : IEquatable<WarningSignalMetadata>
{
public const string TrackSignalMetadataHeaderKey = "track_signal_metadata";
/// <summary> 已经重试的次数，被重新入队时+1 </summary>
public int RetryCount { get; set; }
/// <summary> 是否是无法处理的内容，需要直接报告异常 </summary>
public bool Unhandled { get; set; }
/// <summary> 最后一次出错的原因 </summary>
public string LastErrorReason { get; set; }
[JsonIgnore]
public Message<WarningSignalKey, WarningSignal> 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<WarningSignalKey, WarningSignal> 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<WarningSignalMetadata>(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);
}
}

6
src/SJZY.Track.Application/SjzyTrackApplicationModule.cs
View File

@ -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<WarningTaskComponentBackgroundService>();
// 注册HangFire服务

21
src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
View File

@ -61,4 +61,25 @@ public partial class GlobalConst
/// 轨迹推送通知交换机
/// </summary>
public const string TrackPushExchange = "sjzy.track.on_track_push_ex";
// todo:以下需要在kafka创建主题
/// <summary>
/// 预警信号kafka 主题
/// </summary>
public const string WarningKafkaTopic = "sjzy_trackapi_warning";
/// <summary>
/// 预警信号消费组
/// </summary>
public const string WarningKafkaGroup = "sjzy_trackapi_warning_group";
/// <summary>
/// 预警信号 DLX kafka 主题
/// </summary>
public const string WarningDLXKafkaTopic = "sjzy_trackapi_warning_dlx";
/// <summary>
/// 预警信号 DLX kafka消费组
/// </summary>
public const string WarningDLXKafkaGroup = "sjzy_trackapi_warning_dlx_group";
}