diff --git a/SJZY.Track.Api.sln.DotSettings b/SJZY.Track.Api.sln.DotSettings deleted file mode 100644 index 55bb2ed6..00000000 --- a/SJZY.Track.Api.sln.DotSettings +++ /dev/null @@ -1,3 +0,0 @@ - - True - True \ No newline at end of file diff --git a/TrackCenter.sln b/TrackCenter.sln index 85473f2d..e85884b0 100644 --- a/TrackCenter.sln +++ b/TrackCenter.sln @@ -18,7 +18,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "solution items", "solution NuGet.Config = NuGet.Config .gitignore = .gitignore .editorconfig = .editorconfig - global.json = global.json EndProjectSection EndProject Global diff --git a/src/TrackCenter.Api/Dockerfile b/src/TrackCenter.Api/Dockerfile index 5b3298d2..0fe08490 100644 --- a/src/TrackCenter.Api/Dockerfile +++ b/src/TrackCenter.Api/Dockerfile @@ -1,9 +1,17 @@ -FROM harbor.sjzy.local/images/dotnet:7.0 -COPY publish/ /app/ -COPY version.log /app/ -COPY --chown=appuser:appuser publish/ /app/ -COPY --chown=appuser:appuser version.log /app/ -ENV ASPNETCORE_HOSTINGSTARTUPASSEMBLIES SJZY.Expand.ABP.Core.HostingStartup;SkyAPM.Agent.AspNetCore -ENV SKYWALKING__SERVICENAME Package_Track_Dev -WORKDIR /app -CMD ["/usr/bin/dotnet","SJZY.Track.Api.dll"] \ No newline at end of file +FROM mcr.microsoft.com/dotnet/sdk:8.0.303 AS base +ARG BUILD_CONFIGURATION=Release +WORKDIR /src +COPY . /src +RUN dotnet restore "src/TrackCenter.Api/TrackCenter.Api.csproj" +WORKDIR "/src/src/TrackCenter.Api" +RUN dotnet build "TrackCenter.Api.csproj" -c $BUILD_CONFIGURATION -o /app/build + +FROM base AS publish +ARG BUILD_CONFIGURATION=Release +RUN dotnet publish "TrackCenter.Api.csproj" -c $BUILD_CONFIGURATION -o /app/publish /p:UseAppHost=false + +# 使用运行时镜像 +FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS runtime +WORKDIR /app +COPY --from=publish /app/publish . +ENTRYPOINT ["dotnet", "TrackCenter.Api.dll"] diff --git a/src/TrackCenter.Api/Properties/launchSettings.json b/src/TrackCenter.Api/Properties/launchSettings.json index 58c7579f..a5aa1de9 100644 --- a/src/TrackCenter.Api/Properties/launchSettings.json +++ b/src/TrackCenter.Api/Properties/launchSettings.json @@ -1,15 +1,15 @@ { + "$schema": "https://json.schemastore.org/launchsettings.json", "profiles": { - "SJZY.Track.Api": { + "http": { "commandName": "Project", - "environmentVariables": { - "ASPNETCORE_ENVIRONMENT": "Development", - "ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "SJZY.Expand.ABP.Core", - "SKYWALKING__SERVICENAME": "Track" - }, "dotnetRunMessages": true, - "applicationUrl": "http://*:5056" + "launchBrowser": true, + "launchUrl": "swagger/index.html", + "applicationUrl": "http://localhost:8889", + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + } } - }, - "$schema": "https://json.schemastore.org/launchsettings.json" + } } diff --git a/src/TrackCenter.Api/TrackCenter.Api.csproj b/src/TrackCenter.Api/TrackCenter.Api.csproj index 0a941dd9..32b9d74e 100644 --- a/src/TrackCenter.Api/TrackCenter.Api.csproj +++ b/src/TrackCenter.Api/TrackCenter.Api.csproj @@ -24,8 +24,11 @@ + - + + .dockerignore + true diff --git a/src/TrackCenter.Api/TrackCenterApiModule.cs b/src/TrackCenter.Api/TrackCenterApiModule.cs index 6b38afbd..a4ffc9a6 100644 --- a/src/TrackCenter.Api/TrackCenterApiModule.cs +++ b/src/TrackCenter.Api/TrackCenterApiModule.cs @@ -13,6 +13,7 @@ using TrackCenter.Repository; namespace TrackCenter.Api; [DependsOn(typeof(FakeAutofacModule))] +[DependsOn(typeof(FakeAspNetCoreModule))] [DependsOn(typeof(TrackCenterApplicationModule))] [DependsOn(typeof(TrackCenterRepositoryModule))] public class TrackCenterApiModule : FakeModule @@ -54,7 +55,13 @@ public class TrackCenterApiModule : FakeModule public override void ConfigureApplication(ApplicationConfigureContext context) { var app = context.GetWebApplication(); - var configuration = context.GetConfiguration(); + app.Use(async (httpContext, next) => + { + httpContext.Request.EnableBuffering(); + await next.Invoke(); + }); + + app.UseFakeExceptionHandling(); app.MapControllers(); // Configure the HTTP request pipeline. diff --git a/src/TrackCenter.Api/kafkalib/librdkafka++.so.1 b/src/TrackCenter.Api/kafkalib/librdkafka++.so.1 deleted file mode 100644 index 999f69c9..00000000 Binary files a/src/TrackCenter.Api/kafkalib/librdkafka++.so.1 and /dev/null differ diff --git a/src/TrackCenter.Api/kafkalib/librdkafka.so.1 b/src/TrackCenter.Api/kafkalib/librdkafka.so.1 deleted file mode 100644 index f9fc0367..00000000 Binary files a/src/TrackCenter.Api/kafkalib/librdkafka.so.1 and /dev/null differ diff --git a/src/TrackCenter.Application/CAP/CAPFilter.cs b/src/TrackCenter.Application/CAP/CAPFilter.cs index b725f544..afea2283 100644 --- a/src/TrackCenter.Application/CAP/CAPFilter.cs +++ b/src/TrackCenter.Application/CAP/CAPFilter.cs @@ -20,11 +20,11 @@ public class CapFilter : SubscribeFilter if (!string.IsNullOrEmpty(argumentJsonStr)) { - context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash); + context.DeliverMessage.Headers.TryGetValue("track-arguments-hash", out var hash); if (hash.IsNullOrEmpty()) // 计算参数hash { hash = StringHelper.GetHash(argumentJsonStr); - context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; + context.DeliverMessage.Headers["track-arguments-hash"] = hash; } if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}{DateTime.Now:G}")) @@ -36,7 +36,7 @@ public class CapFilter : SubscribeFilter public override async Task OnSubscribeExecutedAsync(ExecutedContext context) { - context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash); + context.DeliverMessage.Headers.TryGetValue("track-arguments-hash", out var hash); if (!hash.IsNullOrEmpty()) { await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}{DateTime.Now:G}", 1, TimeSpan.FromMinutes(3)); diff --git a/src/TrackCenter.Application/Dtos/ProvideServices/LinkOrderInput.cs b/src/TrackCenter.Application/Dtos/ProvideServices/LinkOrderInput.cs index bb8f5287..1dfa510d 100644 --- a/src/TrackCenter.Application/Dtos/ProvideServices/LinkOrderInput.cs +++ b/src/TrackCenter.Application/Dtos/ProvideServices/LinkOrderInput.cs @@ -8,7 +8,6 @@ public class LinkOrderInput /// 应用 /// [Required] - [Range(1, int.MaxValue)] public string AppKey { get; set; } /// @@ -33,7 +32,6 @@ public class LinkOrderInput /// 关联应用 /// [Required] - [Range(1, int.MaxValue)] public string LinkAppKey { get; set; } /// diff --git a/src/TrackCenter.Application/Dtos/ProvideServices/OrderNoInfo.cs b/src/TrackCenter.Application/Dtos/ProvideServices/OrderNoInfo.cs index 5ea99709..a7ba89b4 100644 --- a/src/TrackCenter.Application/Dtos/ProvideServices/OrderNoInfo.cs +++ b/src/TrackCenter.Application/Dtos/ProvideServices/OrderNoInfo.cs @@ -11,14 +11,12 @@ public class OrderNoInfo /// 应用 /// [Required] - [Range(1, int.MaxValue)] public string AppKey { get; set; } /// /// 订单类型，必填 /// [Required] - [Range(1, int.MaxValue)] public TrackOrderType OrderType { get; set; } /// @@ -35,14 +33,12 @@ public class OrderNosInfo /// 应用 /// [Required] - [Range(1, int.MaxValue)] public string AppKey { get; set; } /// /// 订单类型 /// [Required] - [Range(1, int.MaxValue)] public TrackOrderType OrderType { get; set; } /// diff --git a/src/TrackCenter.Application/Dtos/TrackPush/QueryPushRecordInput.cs b/src/TrackCenter.Application/Dtos/TrackPush/QueryPushRecordInput.cs index b30d98e3..9236edca 100644 --- a/src/TrackCenter.Application/Dtos/TrackPush/QueryPushRecordInput.cs +++ b/src/TrackCenter.Application/Dtos/TrackPush/QueryPushRecordInput.cs @@ -6,12 +6,10 @@ public class QueryPushRecordInput { /// 应用 [Required] - [Range(1, int.MaxValue)] public string AppKey { get; set; } /// 订单类型 [Required] - [Range(1, int.MaxValue)] public TrackOrderType OrderType { get; set; } /// 订单号集合 diff --git a/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs b/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs new file mode 100644 index 00000000..42913291 --- /dev/null +++ b/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs @@ -0,0 +1,91 @@ +using FeishuNetSdk; +using FeishuNetSdk.Im; +using FeishuNetSdk.Im.Dtos; +using TrackCenter.Application.Helpers; + +namespace TrackCenter.Application.Hangfire; + +// 标题：预警通知 +// 内容：尊敬的用户，截止到2024-12-13 10:00:01(UTC +8:00)，轨迹中心监测到与您相关的预警风险信息共计299 条，请及时处理，详情如下： +// 运输时效预警共计100条，其中过去4小时内新增的预警10条，查看详情 +// 回传时效预警共计99条，其中过去4小时内新增的预警10条，查看详情 +// 轨迹完整预警共计100条，其中过去4小时内新增的预警10条，查看详情 +public static class WarningNotifyBuilder +{ + private const string Title = "预警通知"; + + // 构建飞书卡片发消息请求体 + public static PostImV1MessagesBodyDto BuildFeiShuCard(WarningNotifyParam param, string fsUserId, string baseUrl) + { + var totalCount = param.TransportCount + param.ReturnCount + param.CompleteCount; + var body = new List + { + new MarkdownElement(Content: $"尊敬的用户，截止到{DateTime.Now:yyyy-MM-dd HH:mm:ss}({BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下：", TextAlign: "left", TextSize: "normal_v2") + }; + + // 按需构建预警消息体 + if (param.TransportCount > 0) + { + var url1 = WarningUrlBuilder.BuildUrl(baseUrl, WarningRuleType.运输时效预警, param.UserId); + body.Add(new MarkdownElement(Content: $"运输时效预警共计{param.TransportCount}条，其中过去{param.Hours}小时内新增的预警{param.TransportLastNHoursCount}条 [查看详情]({url1})", TextAlign: "left", TextSize: "normal_v2")); + } + + if (param.ReturnCount > 0) + { + var url2 = WarningUrlBuilder.BuildUrl(baseUrl, WarningRuleType.回传时效预警, param.UserId); + body.Add(new MarkdownElement(Content: $"轨迹回传时效预警共计{param.ReturnCount}条，其中过去{param.Hours}小时内新增的预警{param.ReturnLastNHoursCount}条 [查看详情]({url2})", TextAlign: "left", TextSize: "normal_v2")); + } + + if (param.CompleteCount > 0) + { + var url3 = WarningUrlBuilder.BuildUrl(baseUrl, WarningRuleType.完整预警, param.UserId); + body.Add(new MarkdownElement(Content: $"轨迹完整预警共计{param.CompleteCount}条，其中过去{param.Hours}小时内新增的预警{param.CompleteLastNHoursCount}条 [查看详情]({url3})", TextAlign: "left", TextSize: "normal_v2")); + } + + var dto = new PostImV1MessagesBodyDto + { + MsgType = "interactive", + ReceiveId = fsUserId + }; + dto.SetContent(new ElementsCardV2Dto + { + // 配置卡片全局行为 + Config = new ElementsCardV2Dto.ConfigSuffix(), + Header = new ElementsCardV2Dto.HeaderSuffix() + { + Title = new HeaderTitleElement(Title), + Template = "red" + }, + Body = new ElementsCardV2Dto.BodySuffix(body.ToArray()), + }); + + return dto; + } + + // 构建本地时区字符串标识 e.g UTC +8:00 + private static string BuildLocalTimezoneStr() + { + var offset = TimeZoneInfo.Local.BaseUtcOffset; + var flag = offset.Hours > 0 ? "+" : ""; + var offsetStr = offset.ToString().RemovePostFix(":00"); + return $"UTC {flag}{offsetStr}"; + } + + public static class WarningUrlBuilder + { + public const string TransportPath = "/trajectoryCenter/trajectoryWarningOrder/timeWaring"; + public const string ReturnPath = "/trajectoryCenter/trajectoryWarningOrder/ReturnWarning"; + public const string CompletePath = "/trajectoryCenter/trajectoryWarningOrder/CompleteWarning"; + + public static string BuildUrl(string baseUrl, WarningRuleType warningRuleType, int userId) + { + return warningRuleType switch + { + WarningRuleType.运输时效预警 => $"{baseUrl}{TransportPath}?userid={userId}", + WarningRuleType.回传时效预警 => $"{baseUrl}{ReturnPath}?userid={userId}", + WarningRuleType.完整预警 => $"{baseUrl}{CompletePath}?userid={userId}", + _ => throw new ArgumentOutOfRangeException() + }; + } + } +} diff --git a/src/TrackCenter.Application/Hangfire/WarningTaskComponentBackgroundService.cs b/src/TrackCenter.Application/Hangfire/WarningTaskComponentBackgroundService.cs deleted file mode 100644 index 84e1b7f8..00000000 --- a/src/TrackCenter.Application/Hangfire/WarningTaskComponentBackgroundService.cs +++ /dev/null @@ -1,76 +0,0 @@ -using Hangfire; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using TrackCenter.Application.Helpers; -using TrackCenter.Application.Options; - -namespace TrackCenter.Application.Hangfire; - -/// -/// 预警任务组件初始化 -/// tips:支持定时任务执行周期配置 动态更新并应用 √ -/// -public class WarningTaskComponentBackgroundService : BackgroundService -{ - private readonly IOptionsMonitor _optionsMonitor; - private TrackWarningTaskOptions? _oldOptions; - private readonly ILogger _logger; - - public WarningTaskComponentBackgroundService(IOptionsMonitor optionsMonitor, ILogger logger) - { - _optionsMonitor = optionsMonitor; - _logger = logger; - // 防抖配置变更包装 - var onChanged = DebounceHelper.WrapMethod(TimeSpan.FromSeconds(1), () => { ReloadConfig(optionsMonitor); }); - optionsMonitor.OnChange(_ => onChanged.Debounce()); - } - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - // 立即初始化一次 - ReloadConfig(_optionsMonitor); - - return Task.CompletedTask; - } - - private void ReloadConfig(IOptionsMonitor optionsMonitor) - { - var newVal = optionsMonitor.CurrentValue; - _logger.LogInformation("扫描任务配置，预警通知扫描周期{a}h，回传重试推送任务扫描周期{b}min，运输预警明细扫描周期{c}min", - newVal.ScanWarningNotifyTaskHourInterval, - newVal.ScanReturnWarningRetryPushTaskMinuteInterval, - newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval); - if ((_oldOptions?.ScanWarningNotifyTaskHourInterval ?? 0) != newVal.ScanWarningNotifyTaskHourInterval) - { - var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanWarningNotifyTaskLevel]; - // 预警通知任务扫描 - RecurringJob.AddOrUpdate("Scan_WarningNotifyTask", - q, - x => x.ScanNotifyTaskAndSchedulingAsync(), - $"0 */{newVal.ScanWarningNotifyTaskHourInterval} * * *"); - } - - if ((_oldOptions?.ScanReturnWarningRetryPushTaskMinuteInterval ?? 0) != newVal.ScanReturnWarningRetryPushTaskMinuteInterval) - { - var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanReturnWarningRetryPushTaskLevel]; - // 回传预警通知重试推送任务扫描 - RecurringJob.AddOrUpdate("Scan_ReturnWarningRetryPushTask", - q, - x => x.ScanReturnRetryPushTaskAndSchedulingAsync(), - $"*/{newVal.ScanReturnWarningRetryPushTaskMinuteInterval} * * * *"); - } - - if ((_oldOptions?.ScanTransportWarningDetailsGenerateTaskMinuteInterval ?? 0) != newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval) - { - var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanTransportWarningDetailsGenerateTaskLevel]; - // 运输预警明细生成任务扫描 - RecurringJob.AddOrUpdate("Scan_TransportWarningDetailsGenerateTask", - q, - x => x.ScanTransportDetailsGenerateTaskAndSchedulingAsync(), - $"*/{newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval} * * * *"); - } - - _oldOptions = newVal; - } -} diff --git a/src/TrackCenter.Application/Hangfire/WarningTaskScanner.cs b/src/TrackCenter.Application/Hangfire/WarningTaskScanner.cs index 758ae638..a94616e0 100644 --- a/src/TrackCenter.Application/Hangfire/WarningTaskScanner.cs +++ b/src/TrackCenter.Application/Hangfire/WarningTaskScanner.cs @@ -29,7 +29,9 @@ public class WarningTaskScanner : IWarningTaskScanner, ITransientDependency private readonly IWarningTaskService _warningTaskService; private readonly IOptionsMonitor _optionsMonitor; - public WarningTaskScanner(IWarningTaskRepository warningTaskRepository, ITransportWarningHandler transportWarningHandler, WarningNotifyAggregator warningNotifyAggregator, IWarningTaskService warningTaskService, IOptionsMonitor optionsMonitor) + public WarningTaskScanner(IWarningTaskRepository warningTaskRepository, + ITransportWarningHandler transportWarningHandler, WarningNotifyAggregator warningNotifyAggregator, + IWarningTaskService warningTaskService, IOptionsMonitor optionsMonitor) { _warningTaskRepository = warningTaskRepository; _transportWarningHandler = transportWarningHandler; @@ -67,7 +69,8 @@ public class WarningTaskScanner : IWarningTaskScanner, ITransientDependency if (tasks.Count > 0) { var taskIds = tasks.Select(x => x.Id).ToList(); - await _warningTaskRepository.CloseReturnWarningRetryPushTaskAsync(x => !x.IsClosed && taskIds.Contains(x.Id)); + await _warningTaskRepository.CloseReturnWarningRetryPushTaskAsync(x => + !x.IsClosed && taskIds.Contains(x.Id)); } } @@ -83,6 +86,7 @@ public class WarningTaskScanner : IWarningTaskScanner, ITransientDependency // 3.关闭预警生成任务 var taskIds = tasks.Select(x => x.Id).ToList(); - await _warningTaskRepository.CloseTransportWarningDetailsGenerateTaskAsync(x => !x.IsClosed && taskIds.Contains(x.Id)); + await _warningTaskRepository.CloseTransportWarningDetailsGenerateTaskAsync(x => + !x.IsClosed && taskIds.Contains(x.Id)); } } diff --git a/src/TrackCenter.Application/Hangfire/WarningTaskService.cs b/src/TrackCenter.Application/Hangfire/WarningTaskService.cs index ef0c8b56..58b6b781 100644 --- a/src/TrackCenter.Application/Hangfire/WarningTaskService.cs +++ b/src/TrackCenter.Application/Hangfire/WarningTaskService.cs @@ -20,41 +20,33 @@ public interface IWarningTaskService Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity entity, CancellationToken ct); } -public class WarningTaskService : IWarningTaskService, ITransientDependency +public class WarningTaskService( + IOptionsMonitor interfaceRelayOptions, + IOptionsMonitor warningTaskOptions, + IOptionsMonitor trackAppOptions, + IFeishuTenantApi feishuTenantApi, + IWarningDetailsEsRepository warningDetailsEsRepository, + IWarningTaskRepository warningTaskRepository, + InterfaceForwardService interfaceForwardService) + : IWarningTaskService, ITransientDependency { - private readonly IOptionsMonitor _interfaceRelayOptions; - private readonly IOptionsMonitor _warningTaskOptions; - private readonly IFeishuTenantApi _feishuTenantApi; - private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; - private readonly IWarningTaskRepository _warningTaskRepository; - - public WarningTaskService( - IOptionsMonitor interfaceRelayOptions, - IOptionsMonitor warningTaskOptions, - IFeishuTenantApi feishuTenantApi, - IWarningDetailsEsRepository warningDetailsEsRepository, - IWarningTaskRepository warningTaskRepository) - { - _interfaceRelayOptions = interfaceRelayOptions; - _warningTaskOptions = warningTaskOptions; - _feishuTenantApi = feishuTenantApi; - _warningDetailsEsRepository = warningDetailsEsRepository; - _warningTaskRepository = warningTaskRepository; - } - #region 用户中心调用，通知（一次） [AutomaticRetry(Attempts = 1)] public async Task CallNotifyAsync(WarningNotifyParam param, CancellationToken ct) { ct.ThrowIfCancellationRequested(); - var options = _warningTaskOptions.CurrentValue; + var options = warningTaskOptions.CurrentValue; + var baseUrl = trackAppOptions.CurrentValue.WarningNotifyUrl; - // 发送站内信 - // https://centerapi.test.shijizhongyun.com/userApi/index.html?urls.primaryName=%E6%B6%88%E6%81%AF%E8%AE%A2%E9%98%85%E5%AE%9E%E4%BE%8B - // MsgSubscribeInstance/SendMessageByTemplate - - // 发送FeiShu卡片信息 + // todo：发送FeiShu卡片信息 + var fsUserId = ""; + if (string.IsNullOrWhiteSpace(fsUserId)) + throw new BusinessException(message: "查询用户中心失败：SendMessageByTemplateAsync Msg:未找到用户对应的飞书用户Id"); + var dto = WarningNotifyBuilder.BuildFeiShuCard(param, fsUserId, baseUrl); + var res = await feishuTenantApi.PostImV1MessagesAsync("user_id", dto); + if (res.Code != 0) + throw new BusinessException(message: $"飞书调用失败：SendMessageByTemplateAsync Msg:{res.Msg}"); } #endregion @@ -87,27 +79,27 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency using (Track2InterfaceRelayAppContext.Use(new Track2InterfaceRelayAppContext(task.AppCode))) { var upStreamCode = isAdd - ? _interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode - : _interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode; + ? interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode + : interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode; - // var r = await _interfaceForwardServiceV2.Service.BatchForwardThenPushToQueueAsync( - // upStreamCode: upStreamCode, - // serviceProviderCode: task.ServiceProviderCode, - // data: items, - // resSubscribeName: GlobalConst.TrackPushCompletedSubscribe, - // attach: Base64Helper.SerializeObject(task.OptionType)); + var r = await interfaceForwardService.BatchForwardThenPushToQueueAsync( + upStreamCode: upStreamCode, + serviceProviderCode: task.ServiceProviderCode, + data: items, + resSubscribeName: GlobalConst.TrackPushCompletedSubscribe, + attach: Base64Helper.SerializeObject(task.OptionType)); - // if (r.Code != GlobalConst.SuccessCode) - // { - // var msg = $"接口通调用失败：BatchForwardThenPushToQueueAsync" + - // $"RequestId: {r.RequestId};AppCode:{task.AppCode};" + - // $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}"; - // - // // 处理调用接口通失败，直接处于完结状态 - // await HandlerCallBusinessException(task, msg); - // retryCountChanged = true; - // throw new BusinessException(message: msg); - // } + if (r.Code != GlobalConst.SuccessCode) + { + var msg = $"接口通调用失败：BatchForwardThenPushToQueueAsync" + + $"RequestId: {r.RequestId};AppCode:{task.AppCode};" + + $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}"; + + // 处理调用接口通失败，直接处于完结状态 + await HandlerCallBusinessException(task, msg); + retryCountChanged = true; + throw new BusinessException(message: msg); + } } } catch (Exception e) @@ -121,7 +113,7 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency if (!retryCountChanged) { var warningDetailsIds = task.WarningDetailIds.Split(',').ToHashSet(); - await _warningDetailsEsRepository.ReturnWarningDetailsRetryIncrementAsync(warningDetailsIds); + await warningDetailsEsRepository.ReturnWarningDetailsRetryIncrementAsync(warningDetailsIds); } } } @@ -130,7 +122,7 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency { var detailIds = task.WarningDetailIds.Split(',').ToHashSet(); // 1.查询处理（重试中/人工介入）的预警明细 - var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO + var warningDetails = await warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO { DetailsIds = detailIds }); @@ -157,8 +149,8 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency } // 更新预警明细，添加回传报文记录 - await _warningDetailsEsRepository.BulkUpdateReturnWarningDetailsAsync(warningDetails); - await _warningDetailsEsRepository.InsertReturnMessageRecordAsync(new ReturnMessageRecordIndex + await warningDetailsEsRepository.BulkUpdateReturnWarningDetailsAsync(warningDetails); + await warningDetailsEsRepository.InsertReturnMessageRecordAsync(new ReturnMessageRecordIndex { WarningDetailsIds = task.WarningDetailIds, IsSuccess = false, @@ -168,7 +160,7 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency // 添加通知任务 if (addNotifyTasks.Count > 0) - await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks); + await warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks); } private static WarningNotifyTaskEntity BuildWarningNotifyTask(ReturnWarningDetailsIndex detail) diff --git a/src/TrackCenter.Application/Kafka/WarningCalcBackgroundService.cs b/src/TrackCenter.Application/Kafka/WarningCalcBackgroundService.cs index c816d638..85e505aa 100644 --- a/src/TrackCenter.Application/Kafka/WarningCalcBackgroundService.cs +++ b/src/TrackCenter.Application/Kafka/WarningCalcBackgroundService.cs @@ -15,10 +15,14 @@ namespace TrackCenter.Application.Kafka; /// public sealed class WarningCalcBackgroundService : BackgroundService { + private const int MinWarningCalcInterval = 10 * 1000; + private const int MaxWarningCalcBatchSize = 2000; private readonly IServiceScopeFactory _serviceScopeFactory; private readonly IConsumer _consumer; private readonly WarningSignalPublisher _warningSignalPublisher; private TrackWarningTaskOptions _options; + private int _interval; + private int _batchSize; public WarningCalcBackgroundService( IOptionsMonitor optionsMonitor, @@ -31,7 +35,14 @@ public sealed class WarningCalcBackgroundService : BackgroundService _consumer = consumer; _warningSignalPublisher = warningSignalPublisher; _options = optionsMonitor.CurrentValue; - optionsMonitor.OnChange(x => _options = x); + _interval = Math.Max(MinWarningCalcInterval, _options.WarningCalcInterval); + _batchSize = Math.Min(MaxWarningCalcBatchSize, _options.WarningCalcBatchSize); + optionsMonitor.OnChange(newVal => + { + _options = newVal; + _interval = Math.Max(MinWarningCalcInterval, _options.WarningCalcInterval); + _batchSize = Math.Min(MaxWarningCalcBatchSize, _options.WarningCalcBatchSize); + }); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -39,7 +50,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService while (!stoppingToken.IsCancellationRequested) { await DoExecuteAsync(stoppingToken); - await Task.Delay(_options.WarningCalcInterval, stoppingToken); + await Task.Delay(_interval, stoppingToken); } } @@ -75,7 +86,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService private async ValueTask FetchAndHandlerAsync(CancellationToken ct) { // 500ms或者提取到1000条消息任意以条件到达既可返回 - var list = _consumer.ConsumeBatch(_options.WarningCalcBatchSize, TimeSpan.FromMilliseconds(500)).ToList(); + var list = _consumer.ConsumeBatch(_batchSize, TimeSpan.FromMilliseconds(500)).ToList(); // 没有提取到任何消息跳过 if (list.Count == 0) return; @@ -96,6 +107,48 @@ public sealed class WarningCalcBackgroundService : BackgroundService // 提交内存中所有的偏移到服务器，适用于EnableAutoCommit为false的手动提交场景 _consumer.Commit(); + + // 动态调整消费速率 + AdjustSettings(list.Count); + } + + private void AdjustSettings(int actualFetchSize) + { + // MaxBatchSize 2000 MinBatchSize _options + // MinInterval 10s MaxInterval _options + + // 实际拉取为批次大小，设置为最高速率 + if (actualFetchSize >= _batchSize) + { + _interval = MinWarningCalcInterval; + _batchSize = MaxWarningCalcBatchSize; + return; + } + + // 低负载场景：拉取量低于当前批次70%时 + if (actualFetchSize < _batchSize * 0.7) + { + // 逐步缩小批次（每次减少15%） + _batchSize = Math.Max(_options.WarningCalcBatchSize, (int) (_batchSize * 0.85)); + + // 适度增加间隔（当前间隔+20%） + _interval = Math.Min(_options.WarningCalcInterval, (int) (_interval * 1.2)); + return; + } + + // 中等负载场景：拉取量处于70%-90%区间时 + if (actualFetchSize < _batchSize * 0.9) + { + // 温和调整策略 + _batchSize = Math.Min(MaxWarningCalcBatchSize, (int) (_batchSize * 1.05)); // 缓慢扩容 + _interval = Math.Max(MinWarningCalcInterval, (int) (_interval * 0.95)); // 适度缩短间隔 + return; + } + + // 高负载临界：拉取量接近当前批次大小时 + // 直接扩容5%并缩短间隔 + _batchSize = Math.Min(MaxWarningCalcBatchSize, (int) (_batchSize * 1.05)); + _interval = Math.Max(MinWarningCalcInterval, (int) (_interval * 0.9)); } private async ValueTask MessageHandler(IEnumerable> messages, CancellationToken ct) @@ -171,8 +224,10 @@ public sealed class WarningCalcBackgroundService : BackgroundService }).ToArray(); await Task.WhenAll(tasks); - _activity?.SetTag("EndPublicRetryMessage", $"[预警计算] 已重新投递[{needRetryMetadata.Count}]条消息"); - LogHelper.Warn($"[预警计算] 已重新投递[{needRetryMetadata.Count}]条消息", true); + var errorReasons = needRetryMetadata.Select(x => x.LastErrorReason).ToHashSet(); + var errorMessage = "------------\r\n" + string.Join("\r\n------------\r\n", errorReasons); + _activity?.SetTag("EndPublicRetryMessage", $"[预警计算] 已重新投递[{needRetryMetadata.Count}]条消息\r\n{errorMessage}"); + LogHelper.Warn($"[预警计算] 已重新投递[{needRetryMetadata.Count}]条消息\r\n{errorMessage}", true); } // 4.进入死信的消息，投入到死信队列，等待人工处理 @@ -194,8 +249,10 @@ public sealed class WarningCalcBackgroundService : BackgroundService }).ToArray(); await Task.WhenAll(tasks); - _activity?.SetTag("StartPublicDLXMessage", $"[预警计算] 已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息"); - LogHelper.Warn($"[预警计算] 已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息", true); + var errorReasons = needRetryMetadata.Select(x => x.LastErrorReason).ToHashSet(); + var errorMessage = "------------\r\n" + string.Join("\r\n------------\r\n", errorReasons); + _activity?.SetTag("StartPublicDLXMessage", $"[预警计算] 已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息\r\n{errorMessage}"); + LogHelper.Warn($"[预警计算] 已进入死信队列[{needEnqueueDlxMetadata.Count}]条消息\r\n{errorMessage}", true); } } diff --git a/src/TrackCenter.Application/Kafka/WarningCalcServiceExtension.cs b/src/TrackCenter.Application/Kafka/WarningCalcServiceExtension.cs index 98c79b57..985d4753 100644 --- a/src/TrackCenter.Application/Kafka/WarningCalcServiceExtension.cs +++ b/src/TrackCenter.Application/Kafka/WarningCalcServiceExtension.cs @@ -14,7 +14,6 @@ public static class WarningCalcServiceExtension { public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf) { - PrintKafkaLibVersion(); var env = "-" + conf.GetValue("Environment")!.ToLower(); // 预警信号批处理后台任务 services.AddHostedService(); @@ -167,20 +166,6 @@ public static class WarningCalcServiceExtension return list; } - - public static void PrintKafkaLibVersion() - { - var loadSuccess = Library.Load(null); - Console.WriteLine($@" - +-------------------------------------------------------+ - | | - | LibrdKafka | - | 加载：{loadSuccess}，版本号：{Library.VersionString} | - | | - +-------------------------------------------------------+ - "); - - } } internal sealed class JsonCoreSerializer : ISerializer, IDeserializer diff --git a/src/TrackCenter.Application/Rabbit/RabbitOptions.cs b/src/TrackCenter.Application/Rabbit/RabbitOptions.cs index 0a113675..258f5ea7 100644 --- a/src/TrackCenter.Application/Rabbit/RabbitOptions.cs +++ b/src/TrackCenter.Application/Rabbit/RabbitOptions.cs @@ -4,8 +4,6 @@ namespace TrackCenter.Application.Rabbit; public class RabbitOptions { - public string ConnectionString { get; set; } - public List Consumes { get; set; } = new(); public List> Declarations { get; set; } = new(); diff --git a/src/TrackCenter.Application/Track2InterfaceRelay/InterfaceForwardService.cs b/src/TrackCenter.Application/Track2InterfaceRelay/InterfaceForwardService.cs index 72693fc7..87b5742e 100644 --- a/src/TrackCenter.Application/Track2InterfaceRelay/InterfaceForwardService.cs +++ b/src/TrackCenter.Application/Track2InterfaceRelay/InterfaceForwardService.cs @@ -3,7 +3,7 @@ using TrackCenter.Domain.VEntities; namespace TrackCenter.Application.Track2InterfaceRelay; -public class InterfaceForwardService +public class InterfaceForwardService : ITransientDependency { private readonly IHttpClientFactory _httpClientFactory; private readonly IBasicRepository _appRepository; @@ -30,12 +30,14 @@ public class InterfaceForwardService { throw new BusinessException(message: $"请在{nameof(Track2InterfaceRelayAppContext)}中调用接口通"); } + var key = Track2InterfaceRelayAppContext.Current.AppKey; var secret = await _appRepository.GetFirstAsync(x => x.AppKey == key, x => x.AppSecret); if (secret == null) { throw new BusinessException(message: $"未能找到相应app，请联系管理员"); } + content.Headers.Add(nameof(InterfaceForwardOptions.AppKey), key); content.Headers.Add(nameof(InterfaceForwardOptions.AppSecret), secret); diff --git a/src/TrackCenter.Application/TrackCenterApplicationModule.cs b/src/TrackCenter.Application/TrackCenterApplicationModule.cs index 3c3b430d..12bd198a 100644 --- a/src/TrackCenter.Application/TrackCenterApplicationModule.cs +++ b/src/TrackCenter.Application/TrackCenterApplicationModule.cs @@ -10,20 +10,22 @@ using Hangfire; using Hangfire.Redis.StackExchange; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Options; using RabbitMQ.Client; using Savorboard.CAP.InMemoryMessageQueue; using TrackCenter.Application.Kafka; using TrackCenter.Application.CAP; using TrackCenter.Application.Hangfire; +using TrackCenter.Application.HostServices; using TrackCenter.Application.Options; using TrackCenter.Application.Rabbit; namespace TrackCenter.Application; [DependsOn(typeof(FakeRabbitMqModule))] -[DependsOn(typeof(FakeAspNetCoreModule))] [DependsOn(typeof(FakeObjectMappingAutoMapperModule))] [DependsOn(typeof(FakeEventBusRabbitMqModule))] +[DependsOn(typeof(TrackCenterDomainModule))] public class TrackCenterApplicationModule : FakeModule { public override void ConfigureServices(ServiceConfigurationContext context) @@ -35,19 +37,21 @@ public class TrackCenterApplicationModule : FakeModule options.ScanProfiles(); }); + var rabbitOptions = configuration.GetSection("RabbitMQ").Get()!.Default; + //CAP context.Services.AddSingleton(); if (!System.Diagnostics.Debugger.IsAttached) { context.Services.AddCap(options => { - options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!); + options.UseMySql(configuration["ConnectionStrings:Default"]!); options.UseRabbitMQ(configure => { - configure.HostName = configuration["RabbitMQ:HostName"]!; - configure.UserName = configuration["RabbitMQ:UserName"]!; - configure.Password = configuration["RabbitMQ:Password"]!; - configure.Port = configuration.GetValue("RabbitMQ:Port"); + configure.HostName = rabbitOptions.HostName; + configure.UserName = rabbitOptions.UserName; + configure.Password = rabbitOptions.Password; + configure.Port = rabbitOptions.Port; configure.BasicQosOptions = new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount")); }); @@ -86,14 +90,12 @@ public class TrackCenterApplicationModule : FakeModule capOptions.UseInMemoryMessageQueue(); capOptions.UseDashboard(); }).AddSubscribeFilter(); - Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列"); + LogHelper.Info("-- 当前为本地环境，用内存存储和内存队列"); } context.Services.AddHostedService(); context.Services.Configure(options => { - options.ConnectionString = - $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; options.Declarations = new List>() { o => o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true), @@ -149,11 +151,11 @@ public class TrackCenterApplicationModule : FakeModule $",defaultDatabase={configuration["Redis:Database"]}" + $",abortConnect=false"; - // 注册预警计算服务 - context.Services.AddWarningCalcServices(configuration); + // todo：注册预警计算服务 + //context.Services.AddWarningCalcServices(configuration); // 注册预警任务后台服务 - context.Services.AddHostedService(); + context.Services.AddHostedService(); // 注册飞书服务（用于预警通知） context.Services.AddFeishuNetSdk(o => @@ -188,20 +190,7 @@ public class TrackCenterApplicationModule : FakeModule context.Services.AddHttpClient("itfx", client => { client.Timeout = TimeSpan.FromSeconds(20); - client.BaseAddress = new Uri(configuration["itfx:BaseAddress"]!); + client.BaseAddress = new Uri(configuration["TrackApp:ItfxBaseAddress"]!); }); } - - public override void PreConfigureServices(ServiceConfigurationContext context) - { - // PreConfigure(options => - // { - // options.ProxyClientActions.Add((_, _, httpClient) => - // { - // //如果设置Token，前端需要传入Token 否则引发异常 - // httpClient.SetTokenHeader(); - // httpClient.Timeout = TimeSpan.FromMinutes(3); - // }); - // }); - } } diff --git a/src/TrackCenter.Domain/Aggregates/OrderAggregate/OrderTrackPullStatusEntity.cs b/src/TrackCenter.Domain/Aggregates/OrderAggregate/OrderTrackPullStatusEntity.cs index 467c10e8..434f26fc 100644 --- a/src/TrackCenter.Domain/Aggregates/OrderAggregate/OrderTrackPullStatusEntity.cs +++ b/src/TrackCenter.Domain/Aggregates/OrderAggregate/OrderTrackPullStatusEntity.cs @@ -1,3 +1,4 @@ +using Fake.Domain.Entities; using Fake.Domain.Entities.Auditing; using TrackCenter.Domain.SeedWork; @@ -7,7 +8,8 @@ namespace TrackCenter.Domain.Aggregates.OrderAggregate; /// 订单轨迹拉取状态表（1：N） /// [SugarTable("t_order_pull_status")] -public class OrderPullStatusEntity : FullAuditedEntity +public class OrderPullStatusEntity : Entity, IHasCreateUserId, IHasCreateTime, IHasUpdateUserId, + IHasUpdateTime { /// /// 自增主键 @@ -88,7 +90,7 @@ public class OrderPullStatusEntity : FullAuditedEntity public DateTime CreateTime { get; set; } [SugarColumn(ColumnName = "UpdateTime")] - public DateTime? UpdateTime { get; set; } + public DateTime UpdateTime { get; set; } [SugarColumn(ColumnName = "UpdateUserId")] public int UpdateUserId { get; set; } diff --git a/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs b/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs index 57cb6297..0f7cbcc3 100644 --- a/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs +++ b/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs @@ -5,36 +5,36 @@ public partial class GlobalConst /// /// 轨迹明细 es索引名称（模板按月） /// - public const string TrackDetailsIndex = "sjzy_track_details"; + public const string TrackDetailsIndex = "track_details"; /// /// 轨迹推送记录 es索引名称（模板按月） /// - public const string TrackPushRecordIndex = "sjzy_track_push_records"; + public const string TrackPushRecordIndex = "track_push_records"; /// /// 订单操作记录 es索引名称（模板按月） /// - public const string OrderOperationRecordIndex = "sjzy_track_order_operation_records"; + public const string OrderOperationRecordIndex = "track_order_operation_records"; // todo：以下为新增需要配置索引模板(prod) 生产环境还需要建立索引！！ /// /// 运输时效预警明细 es索引名称（模板按月） /// - public const string TransportWarningDetailsIndex = "sjzy_track_transport_warning_details"; + public const string TransportWarningDetailsIndex = "track_transport_warning_details"; /// /// 回传时效预警明细 es索引名称（模板按月） /// - public const string ReturnWarningDetailsIndex = "sjzy_track_return_warning_details"; + public const string ReturnWarningDetailsIndex = "track_return_warning_details"; /// /// 回传报文日志 es索引名称（模板按月） /// - public const string ReturnMessageRecordIndex = "sjzy_track_return_message_records"; + public const string ReturnMessageRecordIndex = "track_return_message_records"; /// /// 完整预警明细 es索引名称（模板按月） /// - public const string CompleteWarningDetailsIndex = "sjzy_track_complete_warning_details"; + public const string CompleteWarningDetailsIndex = "track_complete_warning_details"; } diff --git a/src/TrackCenter.Domain/SJZYTrackDomainModule.cs b/src/TrackCenter.Domain/SJZYTrackDomainModule.cs deleted file mode 100644 index ef8e476a..00000000 --- a/src/TrackCenter.Domain/SJZYTrackDomainModule.cs +++ /dev/null @@ -1,7 +0,0 @@ -using Fake.Modularity; - -namespace TrackCenter.Domain; - -public class SjzyTrackDomainModule : FakeModule -{ -} diff --git a/src/TrackCenter.Domain/TrackCenterDomainModule.cs b/src/TrackCenter.Domain/TrackCenterDomainModule.cs new file mode 100644 index 00000000..cb8ea79f --- /dev/null +++ b/src/TrackCenter.Domain/TrackCenterDomainModule.cs @@ -0,0 +1,17 @@ +using Fake.DomainDrivenDesign; +using Fake.Modularity; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using TrackCenter.Domain.Extensions; + +namespace TrackCenter.Domain; + +[DependsOn(typeof(FakeDomainDrivenDesignModule))] +public class TrackCenterDomainModule : FakeModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + var configuration = context.Services.GetConfiguration(); + LogHelper.Init(configuration.GetSection("FeiShuNotice").Get()!); + } +} diff --git a/src/TrackCenter.Repository/Repositories/TrackDetailsEsRepository.cs b/src/TrackCenter.Repository/Repositories/TrackDetailsEsRepository.cs index 295c65d3..e1582145 100644 --- a/src/TrackCenter.Repository/Repositories/TrackDetailsEsRepository.cs +++ b/src/TrackCenter.Repository/Repositories/TrackDetailsEsRepository.cs @@ -804,7 +804,7 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract GetDetailAsync(string orderNo, string trackCode, TrackOrderType orderType, string appKey, DateTime trackTime) { var res = await Context.SearchAsync(s => s - .Index("sjzy_track_details") + .Index(GlobalConst.TrackDetailsIndex) .Query(q => q .Bool(b => b .Must(mu => mu