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

Merge branch 'S07' into test

Browse Source
This commit is contained in:
hardstylewyl 2025-05-16 16:21:09 +08:00
commit 555baa259d
29 changed files with 994 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/SJZY.Track.Api/Program.cs
View File

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

14
src/SJZY.Track.Api/SjzyTrackApiModule.cs
View File

@ -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($@"
+-------------------------------------------------------+
| |

7
src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs
View File

@ -20,9 +20,14 @@ public interface IBroadcastingService : ITransientDependency
/// </summary>
/// <param name="eto"></param>
/// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param>
void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode);
/// <summary>
/// 轨迹推送完成广播（推送完成回调）
/// </summary>
/// <param name="list"></param>
void TrackPushCompletedPublicTopic(List<TrackPushRecordDto> list);
/// <summary>
/// 轨迹推送完成广播（推送完成回调）
/// </summary>

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

@ -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<TrackDetailsIndex> delTracks)
{
var signals = TrackSignal.FromTrack(delTracks, WarningSignalType.轨迹删除);
if (_BUGOptions.EnablePref)
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks);
scope.Dispose();
return;
}
//_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks);
var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除);
_ = Task.Run(async () =>
{

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

@ -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<TrackUpsertEventHandler> _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<TrackUpsertEventHandler> 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<TrackDetailsIndex> tracks)
{
var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹新增);
if (_BUGOptions.EnablePref)
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_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<TrackDetailsIndex> tracks)
{
var signals = TrackSignal.FromTrack(tracks, WarningSignalType.轨迹更新);
if (_BUGOptions.EnablePref)
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks);
scope.Dispose();
return;
}
//_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks);
var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新);
_ = Task.Run(async () =>
{

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

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

@ -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<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();
});
#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<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();
});
#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<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>

81
src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs Normal file
View File

@ -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;
/// <summary>
/// 支持发布确认模式
/// </summary>
public class RabbitClientBusExt : RabbitClientBus, ISingletonDependency
{
public RabbitClientBusExt(IServiceProvider sp, ILogger<RabbitClient> logger)
: base(sp, logger, sp.GetRequiredService<IOptionsMonitor<RabbitOptions>>().CurrentValue)
{
}
public void PublishByConfirmSelect(
string exchange,
string routingKey,
byte[] data,
Action<IBasicProperties>? options = null
)
{
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
activity?.SetTag("kind", "publish");
var key = exchange + ":" + routingKey;
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(CreateModel)));
var basicProperties1 = orAdd.Value.CreateBasicProperties();
// 配置属性
if (options != null)
options(basicProperties1);
var model = orAdd.Value;
// 启用发布确认
model.ConfirmSelect();
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
model.WaitForConfirmsOrDie();
}
public void PublishByConfirmSelect(
string exchange,
string routingKey,
IEnumerable<byte[]> datas,
Action<IBasicProperties>? options = null
)
{
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
activity?.SetTag("kind", "publish");
var key = exchange + ":" + routingKey;
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(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<byte>) data;
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
}
model.WaitForConfirmsOrDie();
}
private IConnection GetConnection()
{
var key = Connections.OrderBy((Func<KeyValuePair<IConnection, int>, int>) (q => q.Value)).Select((Func<KeyValuePair<IConnection, int>, IConnection>) (q => q.Key)).FirstOrDefault();
Connections.AddOrUpdate(key, (Func<IConnection, int>) (_ => 1), (Func<IConnection, int, int>) ((_, v) => v + 1));
return key;
}
private IModel CreateModel() => GetConnection().CreateModel();
}

2
src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs
View File

@ -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);

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

@ -8,14 +8,16 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="BSI.Stack.Rabbit" Version="2.1.3" />
<PackageReference Include="BSI.Stack.Rabbit" Version="2.1.4" />
<PackageReference Include="Confluent.Kafka.Core" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.JsonCore" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.NewtonsoftJson" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.ProtobufNet" Version="1.0.0" />
<PackageReference Include="Hangfire.AspNetCore" Version="1.8.18" />
<PackageReference Include="Hangfire.Redis.StackExchange" Version="1.9.4" />
<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" />
<PackageReference Include="StackExchange.Redis.Extensions.AspNetCore" Version="10.2.0" />
<PackageReference Include="StackExchange.Redis.Extensions.System.Text.Json" Version="10.2.0" />
<PackageReference Include="Volo.Abp.Http.Client" Version="7.0.2" />
<PackageReference Include="Volo.Abp.FluentValidation" Version="7.0.2" />
<PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.1.8.21" />

56
src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs
View File

@ -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;
/// </summary>
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; });
}
/// <summary>
/// 轨迹推送完成广播（批量）
/// </summary>
/// <param name="list"></param>
/// <returns></returns>
public void TrackPushCompletedPublicTopic(List<TrackPushRecordDto> 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; });
}
/// <summary>
@ -82,7 +96,24 @@ public class BroadcastingService : IBroadcastingService
/// <returns></returns>
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; });
}
/// <summary>
/// 轨迹推送失败广播
/// </summary>
/// <param name="log"></param>
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);
}
}

6
src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs
View File

@ -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<TrackSignal> deletedSignals)
public async Task TrackDeletedHandleAsync(List<WarningSignal> 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<TrackSignal> addedSignals)
public async Task TrackAddedHandleAsync(List<WarningSignal> addedSignals)
{
// 1.只处理未删除的订单
var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet();

6
src/SJZY.Track.Application/Services/Warning/Handlers/ICompleteWarningHandler.cs
View File

@ -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<TrackSignal> deletedSignals);
Task TrackDeletedHandleAsync(List<WarningSignal> deletedSignals);
// 批量处理轨迹 新增信号
Task TrackAddedHandleAsync(List<TrackSignal> addedSignals);
Task TrackAddedHandleAsync(List<WarningSignal> addedSignals);
// 完整预警主体禁用
Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule);

4
src/SJZY.Track.Application/Services/Warning/Handlers/IReturnWarningHandler.cs
View File

@ -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<TrackSignal> updateOrDeleteSignals);
Task TrackUpdatedOrDeletedHandleAsync(List<WarningSignal> updateOrDeleteSignals);
// 收到轨迹推送记录（接口通回调）（人工介入/重试 阶段的回传）
Task ReceiveTrackPushRecordHandleAsync(bool isSuccess, string requestParams, List<TrackPushRecordIndex> records, List<TrackPushInput> pushInputs, List<TrackUpdateInput> updateInputs);

6
src/SJZY.Track.Application/Services/Warning/Handlers/ITransportWarningHandler.cs
View File

@ -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<TrackSignal> deletedSignals);
Task TrackDeletedHandleAsync(List<WarningSignal> deletedSignals);
// 批量处理轨迹新增/修改
Task TrackAddedOrUpdatedHandleAsync(List<TrackSignal> addOrUpdateSignals);
Task TrackAddedOrUpdatedHandleAsync(List<WarningSignal> addOrUpdateSignals);
// 运输预警主体禁用
Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule);

4
src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs
View File

@ -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<TrackSignal> updateOrDeleteSignals)
public async Task TrackUpdatedOrDeletedHandleAsync(List<WarningSignal> updateOrDeleteSignals)
{
// 1.查询订单
var orderIdSet = updateOrDeleteSignals.Select(x => x.OrderId).ToHashSet();

6
src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs
View File

@ -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<TrackSignal> deletedSignals)
public async Task TrackDeletedHandleAsync(List<WarningSignal> 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<TrackSignal> addOrUpdateSignals)
public async Task TrackAddedOrUpdatedHandleAsync(List<WarningSignal> addOrUpdateSignals)
{
// 1.查询订单
var orderIdSet = addOrUpdateSignals.Select(x => x.OrderId).ToHashSet();

51
src/SJZY.Track.Application/Services/Warning/Signals/WarningSignal.cs Normal file
View File

@ -0,0 +1,51 @@
using Newtonsoft.Json;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
namespace SJZY.Track.Application.Services.Warning.Signals;
/// <summary>
/// 预警信号，目前只是操作轨迹时产生
/// 未来可以通过扩展类型+扩展字段的形式增加更多种预警触发源
/// </summary>
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<WarningSignal> FromTrack(List<TrackDetailsIndex> 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
);
}
}

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

@ -0,0 +1,45 @@
using Newtonsoft.Json;
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; }
[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);
}
}

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

@ -0,0 +1,104 @@
using Confluent.Kafka;
using Newtonsoft.Json;
using JsonSerializer = System.Text.Json.JsonSerializer;
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; }
[System.Text.Json.Serialization.JsonIgnore]
[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);
}
}

15
src/SJZY.Track.Application/Services/Warning/Signals/WarningSignalType.cs Normal file
View File

@ -0,0 +1,15 @@
namespace SJZY.Track.Application.Services.Warning.Signals;
/// <summary>
/// 预警信号类型
/// </summary>
public enum WarningSignalType
{
未知的 = 0,
轨迹新增 = 1,
轨迹删除 = 2,
轨迹更新 = 3,
}

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

@ -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<RabbitOptions> 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<TrackAppOptions>(configuration.GetSection("TrackApp"));
context.Services.Configure<InterfaceRelayParamsConfigOptions>(
@ -200,21 +201,12 @@ public class SjzyTrackApplicationModule : AbpModule
$",password={configuration["Redis:Password"]}" +
$",defaultDatabase={configuration["Redis:Database"]}" +
$",abortConnect=false";
// 注册Redis服务
context.Services.AddStackExchangeRedisExtensions<SystemTextJsonSerializer>(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<WarningTaskComponentBackgroundService>();
// 注册HangFire服务

6
src/SJZY.Track.Application/_BUGOptions.cs Normal file
View File

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

43
src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/TrackSignal.cs
View File

@ -1,43 +0,0 @@
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
/// <summary>
/// 轨迹信号，操作轨迹时产生
/// 用于进行预警计算
/// </summary>
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<TrackSignal> FromTrack(List<TrackDetailsIndex> tracks, WarningSignalType type)
{
return tracks
.Select(x => new TrackSignal(
type,
x.OrderId,
x.Id,
x.TrackStatusId,
x.TrackContentId
))
.ToList();
}
}

10
src/SJZY.Track.Domain/Aggregates/WarningAggregate/Signals/WarningSignalType.cs
View File

@ -1,10 +0,0 @@
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
public enum WarningSignalType
{
轨迹新增 = 0,
轨迹删除 = 1,
轨迹更新 = 2,
}

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";
}

8
src/SJZY.Track.Repository/Repositories/OrderRepository.cs
View File

@ -223,14 +223,14 @@ public class OrderRepository : BasicRepository<OrderEntity>, 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;