From 2a76994edaeeb7204b4c0a69037fa4601bcf21f2 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Mon, 12 Aug 2024 15:16:47 +0800 Subject: [PATCH] =?UTF-8?q?important=EF=BC=9A=E5=AF=B9=E6=8E=A5=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E9=80=9A=E8=8E=B7=E5=8F=96=E8=BD=A8=E8=BF=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Dtos/TrackPull/TrackPullPayload.cs | 9 ++ .../SJZY.Track.Application.Contract.csproj | 3 - .../EventHandlers/TrackPushEventHandler.cs | 4 +- .../TrackRefreshEventHandler.cs | 117 ++++++++++++++---- .../SJZY.Track.Application.csproj | 3 +- .../ProvideServices/TrackPullService.cs | 31 +++-- .../SjzyTrackApplicationModule.cs | 28 ++++- .../Consts/GlobalConst.MQ.cs | 20 +-- .../Consts/InterfaceRelayConst.cs | 2 +- .../Queries/ProvideQueries.cs | 6 +- 10 files changed, 157 insertions(+), 66 deletions(-) create mode 100644 src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs rename src/SJZY.Track.Application/{CAP/EventHandlers => Rabbit}/TrackRefreshEventHandler.cs (80%) diff --git a/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs b/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs new file mode 100644 index 00000000..c20366d1 --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs @@ -0,0 +1,9 @@ +namespace SJZY.Track.Application.Contract.Dtos.TrackPull; + +public class TrackPullPayload +{ + /// + /// 订单号 + /// + public string OrderNo { get; set; } = null!; +} \ No newline at end of file diff --git a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj index 749e25a6..82b6c428 100644 --- a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj +++ b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj @@ -15,8 +15,5 @@ - - - diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 402668e3..84b86d8e 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -74,7 +74,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private async Task DoPushAsync(TrackPushEvent @event) { - var items = new List(); + var items = new List(); TrackPushContentConfigDto? config; if (@event.OperationMan == null) @@ -165,7 +165,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null || !config.IsCanUpdate) return; - var items = new List(); + var items = new List(); foreach (var order in _orders) { var log = new TrackPushRecordIndex diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs b/src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs similarity index 80% rename from src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs rename to src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs index 6f878362..9f9dcfbe 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs +++ b/src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs @@ -1,8 +1,14 @@ using System.Diagnostics; using System.Linq.Expressions; +using System.Text; +using BSI.Stack.Rabbit; +using BSI.Stack.Rabbit.Handler; using DotNetCore.CAP; +using Microsoft.Extensions.Logging; using Newtonsoft.Json; using Newtonsoft.Json.Linq; +using RabbitMQ.Client.Events; +using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.TrackPull; using SJZY.Track.Application.Contract.Events; @@ -15,37 +21,72 @@ using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.CAP.EventHandlers; +namespace SJZY.Track.Application.Rabbit; -public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe +public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler { private readonly ICapPublisher _capPublisher; private readonly IOrderRepository _orderRepository; private readonly ITimezoneQueries _timezoneQueries; + private readonly ILogger _logger; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ITrackPullCacheService _trackPullCacheService; public TrackRefreshEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService, - ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries) + ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries, + ILogger logger) { _orderRepository = orderRepository; _capPublisher = capPublisher; _trackPullCacheService = trackPullCacheService; _trackDetailsRepository = trackDetailsRepository; _timezoneQueries = timezoneQueries; + _logger = logger; + } + + + public bool Enable(ConsumeOptions options) + { + return options.Queue == GlobalConst.TrackRefreshSubscribe; + } + + public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) + { + var msg = Encoding.UTF8.GetString(args.Body.ToArray()); + var @event = System.Text.Json.JsonSerializer.Deserialize(msg); + if (@event == null) + { + _logger.LogWarning( + $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); + return; + } + + await ConsumerAsync(@event); } - [CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)] public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { - if (!@event.IsSuccess) return; + if (!@event.IsSuccess) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹刷新失败", @event.RequestId); + return; + } + var attach = Base64Helper.DeSerialize(@event.Attach); - if (attach is null) return; + if (attach is null || @event.Data == null) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，attach或data解析为空", @event.RequestId); + return; + } var config = await _trackPullCacheService.GetPullParamWithContentConfigAsync(attach.SubSystemId, attach.ServiceProviderId); - if (@event.Data == null || config == null) return; + if (config == null) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，找不到拉取配置：{@Attach}", attach.ToJson(), @event.RequestId); + return; + } var data = JObject.Parse(@event.Data); @@ -183,28 +224,22 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe // 轨迹时区 var (trackTimeZone, hasTimezoneBottomLine) = - await GetTimezone(paramConfig, matchedContentConfig, context.OrderId, trackToken); + await GetTimezone(paramConfig, matchedContentConfig, context, trackToken); // 轨迹发生地 var (trackPlace, hasPlaceBottomLine) = - await GetTrackPlace(paramConfig, matchedContentConfig, context.OrderId, trackToken); + await GetTrackPlace(paramConfig, matchedContentConfig, context, trackToken); if (matchedContentConfig is null) { + _logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}", + context.OrderNo, trackToken.ToString()); await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, paramConfig); return; } - int deficiencyTagValue = 0; - // 触发了兜底，但是没有兜底住，对订单进行标记 - if (hasPlaceBottomLine || hasTimezoneBottomLine) - { - //1表示时区，2表示发生地，3表示两种 - var placeTagValue = hasPlaceBottomLine ? 2 : 0; - var timezoneTagValue = hasTimezoneBottomLine ? 1 : 0; - deficiencyTagValue = placeTagValue + timezoneTagValue; - } - + var deficiencyTagValue = CalcDeficiencyTagValue(hasPlaceBottomLine, hasTimezoneBottomLine, trackToken, context); + // 新增/更新轨迹 var trackUpsertEvent = new TrackUpsertEvent { @@ -239,6 +274,35 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } + /// + /// 触发了兜底，但是没有兜底住，对订单进行标记 + /// 1表示时区，2表示发生地，3表示两种 + /// + /// + /// + /// + /// + /// + private int CalcDeficiencyTagValue(bool hasPlaceBottomLine, bool hasTimezoneBottomLine, JToken trackToken, + TrackRefreshContext context) + { + var deficiencyTagValue = 0; + + if (hasTimezoneBottomLine) + { + _logger.LogWarning("订单号：{@OrderNo}, 轨迹发生时区无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString()); + deficiencyTagValue += 1; + } + + if (hasPlaceBottomLine) + { + _logger.LogWarning("订单号：{@OrderNo}, 轨迹发生地无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString()); + deficiencyTagValue += 2; + } + + return deficiencyTagValue; + } + private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, DateTime? trackTime, float? trackTimeZone, TrackPullParamConfigDto paramConfig) { @@ -275,11 +339,11 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe /// /// 配置信息 /// 配置信息 - /// 订单id + /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(float? trackTimeZone, bool hasTimezoneBottomLine)> GetTimezone( - TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, int orderId, + TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生时区 @@ -291,7 +355,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe if (!trackTimeZone.HasValue && matchedContentConfig?.TrackZonePlaceValue is not null) { //兜底没配置的就不需要兜底，拉取的发生地没就没 - var timezone = await GetTimezone(matchedContentConfig, orderId); + var timezone = await GetTimezone(matchedContentConfig, context.OrderId); trackTimeZone = timezone; hasTimezoneBottomLine = !timezone.HasValue; @@ -305,22 +369,21 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe /// /// 配置信息 /// 配置信息 - /// 订单id + /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(string? trackPlace, bool hasPlaceBottomLine)> GetTrackPlace(TrackPullParamConfigDto paramConfig, - TrackPullContentConfigDto? matchedContentConfig, int orderId, JToken trackToken) + TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName == null ? null : trackToken[paramConfig.TrackPlaceName]?.Value(); var hasPlaceBottomLine = false; //执行了兜底但是没有兜底住，需要记录 - if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue != null && - matchedContentConfig.TrackPlaceValue.HasValue) + if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue is not null) { //兜底没配置的就不需要兜底，拉取的发生地没就没 - trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, orderId); + trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, context.OrderId); hasPlaceBottomLine = string.IsNullOrWhiteSpace(trackPlace); } diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index 011dd0d7..5f44d891 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -12,7 +12,8 @@ - + + diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs index b781d848..494352a4 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs @@ -1,14 +1,13 @@ using Microsoft.AspNetCore.Authorization; -using Microsoft.Extensions.Logging; using SJZY.Expand.ABP.Core; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; -using SJZY.Track.Application.CAP.EventHandlers; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Dtos.TrackPull; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Helpers; +using SJZY.Track.Application.Rabbit; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.Http.Client.DynamicProxying; @@ -59,11 +58,10 @@ public class TrackPullService : ApplicationService foreach (var group in orders.GroupBy(x => new { x.ServiceProviderId, x.ServiceProviderName, x.SubSystemId, x.OrderType })) { - var data = group.Select(x => new + var data = new List(group.Select(x => new TrackPullPayload { - x.OrderNo - }).ToList(); - + OrderNo = x.OrderNo + })); var serviceProviderCode = group.FirstOrDefault()!.ServiceProviderCode; var attach = Base64Helper.SerializeObject(group.Key); @@ -71,7 +69,7 @@ public class TrackPullService : ApplicationService var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); - + if (res.Code != InterfaceRelayConst.SuccessCode) { LogHelper.Error($"刷新轨迹失败：{res.ToJson()}"); @@ -94,7 +92,7 @@ public class TrackPullService : ApplicationService await Task.WhenAll(tasks); return true; } - + /// /// 轨迹刷新测试 /// @@ -105,7 +103,8 @@ public class TrackPullService : ApplicationService #if DEBUG [AllowAnonymous] #endif - public async Task TrackRefreshEventTest([FromServices]TrackRefreshEventHandler trackRefreshEventHandler, InterfaceRelayForwardReplayEvent input) + public async Task TrackRefreshEventTest([FromServices] TrackRefreshEventHandler trackRefreshEventHandler, + InterfaceRelayForwardReplayEvent input) { await trackRefreshEventHandler.ConsumerAsync(input); @@ -121,7 +120,7 @@ public class TrackPullService : ApplicationService } var page = 1; - var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page, 2000); + var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page); while (orders.Any()) { var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; @@ -129,9 +128,15 @@ public class TrackPullService : ApplicationService foreach (var group in orders.GroupBy(x => new { x.SubSystemId, x.OrderType })) { - var data = group.Select(x => new { x.OrderNo }).ToList(); - var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, serviceProviderName!, group.Key.SubSystemId, group.Key.OrderType)); - var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); + var data = new List(group.Select(x => new TrackPullPayload + { + OrderNo = x.OrderNo + })); + var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, + serviceProviderName!, group.Key.SubSystemId, group.Key.OrderType)); + var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, + GlobalConst.TrackRefreshSubscribe, attach); if (res.Code != InterfaceRelayConst.SuccessCode) { diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 0dfe7b1e..50dc5959 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -1,6 +1,8 @@ -using DotNetCore.CAP; +using BSI.Stack.Rabbit; +using DotNetCore.CAP; using DotNetCore.CAP.Internal; using Microsoft.Extensions.Configuration; +using RabbitMQ.Client; using Savorboard.CAP.InMemoryMessageQueue; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; @@ -9,6 +11,7 @@ using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.CAP; using SJZY.Track.Application.Contract; using SJZY.Track.Application.Localization; +using SJZY.Track.Domain.Consts; using SJZY.UserCenter.Contracts; using SJZY.UserCenter.SharedModel; using SJZY.UserCenter.SharedModel.Pub; @@ -121,6 +124,29 @@ namespace SJZY.Track.Application } }; }).AddSubscribeFilter(); + + + context.Services.AddRabbitBoot(options => + { + options.ConnectionString = + $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; + options.PoolSize = configuration.GetValue("RabbitMQ:PoolSize"); + options.Consumes = new List + { + new() + { + Queue = GlobalConst.TrackRefreshSubscribe, + FetchCount = configuration.GetValue("RabbitMQ:.TrackRefreshFetchCount"), + FailedRequeue = false, + Declaration = o => + { + o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true); + o.QueueDeclare(GlobalConst.TrackRefreshSubscribe, true); + o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange, GlobalConst.TrackRefreshSubscribe); + } + } + }; + }, typeof(SjzyTrackApplicationModule).Assembly); } public override void PreConfigureServices(ServiceConfigurationContext context) diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 515afe91..a6c4af1f 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -12,11 +12,6 @@ public partial class GlobalConst /// public const string TrackUpsertForRefreshSubscribe = "SJZY.Track.TrackUpsertForRefresh"; - /// - /// 无法映射的轨迹新增 - /// - public const string TrackInsertForNotMappedEventSubscribe = "SJZY.Track.TrackInsertForNotMapped"; - /// /// 轨迹删除 /// @@ -36,6 +31,11 @@ public partial class GlobalConst /// 轨迹人工手动推送 /// public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush"; + + /// + /// 接口通交换机 + /// + public const string InterfaceRelayExchange = "sjzy.interface.relay.direct"; /// /// 轨迹推送完成（接口通回调） @@ -47,16 +47,6 @@ public partial class GlobalConst /// public const string TrackChangeAfterUpdateOrderSubscribe = "SJZY.Track.TrackChangeAfterUpdateOrder"; - /// - /// 轨迹推送后触发更新订单主表 - /// - public const string TrackPushAfterUpdateOrderSubscribe = "SJZY.Track.TrackPushAfterUpdateOrder"; - - /// - /// 关联订单后补轨迹 - /// - public const string OrderLinkAfterFillTrack = "SJZY.Track.OrderLinkAfterFillTrack"; - /// /// 推送预警任务 /// diff --git a/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs b/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs index 24743a5b..1d66b92f 100644 --- a/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs +++ b/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs @@ -13,7 +13,7 @@ public class InterfaceRelayConst /// /// 获取轨迹接口代码 /// - public const string GetTrackInterfaceCode = "A012"; + public const string GetTrackInterfaceCode = "TEST"; /// /// 推送轨迹接口代码 diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 60043689..8ca8c9a6 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -122,7 +122,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, return await Context.Queryable() .RightJoin((a, b) => a.Id == b.OrderId) .LeftJoin((a, b, c) => b.ServiceProviderId == c.Id) - .Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull==1) + .Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull == 1) .Select((a, b, c) => new OrderWithServiceProviderVO { OrderNo = a.OrderNo, @@ -136,12 +136,12 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, .Take(pageSize) .ToListAsync(); } - + public async Task> FindOrderWithServiceProviderAsync(int[] orderIds) { return await Context.Queryable() .InnerJoin((a, b) => a.Id == b.OrderId) - .Where((a, b) => orderIds.Contains( a.Id) && !a.IsCompleted) + .Where((a, b) => orderIds.Contains(a.Id) && !a.IsCompleted) .Select((a, b) => new OrderTransportDto { OrderId = a.Id,