From c66fe8b7dcb71e6b4eaf1ab0a94eb2cddc62c4ef Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Mon, 14 Jul 2025 18:20:21 +0800 Subject: [PATCH] 1 --- ...TrackChangeAfterUpdateOrderEventHandler.cs | 2 +- .../EventHandlers/TrackDeleteEventHandler.cs | 452 +++++++++--------- .../EventHandlers/TrackPushEventHandler.cs | 62 ++- .../EventHandlers/TrackUpsertEventHandler.cs | 13 +- .../ProvideServices/TrackIdentifierDto.cs | 16 + .../Dtos/TrackPull/TrackPullParamConfigDto.cs | 10 + .../Events/ModifyWarningEvent.cs | 22 - .../Events/PushWarningEvent.cs | 6 - .../Events/TrackWarningCalcEvent.cs | 29 -- .../Hangfire/WarningNotifyBuilder.cs | 120 ----- .../Hangfire/WarningTaskService.cs | 60 +-- .../WarningCalcBackgroundService.cs | 284 +++++++++++ .../WarningTaskComponentHostedService.cs | 81 ++++ .../Rabbit/TrackPushCompletedEventHandler.cs | 81 +--- .../Rabbit/TrackRefreshEventHandler.cs | 133 +++++- .../Services/ProvideServices/OrderService.cs | 42 +- .../Services/TrackOrders/TrackOrderService.cs | 2 +- .../Consts/GlobalConst.ES.cs | 5 - .../Consts/GlobalConst.MQ.cs | 5 - .../Consts/GlobalConst.Redis.cs | 5 - src/TrackCenter.Domain/Consts/GlobalConst.cs | 2 + 21 files changed, 808 insertions(+), 624 deletions(-) delete mode 100644 src/TrackCenter.Application/Events/ModifyWarningEvent.cs delete mode 100644 src/TrackCenter.Application/Events/PushWarningEvent.cs delete mode 100644 src/TrackCenter.Application/Events/TrackWarningCalcEvent.cs delete mode 100644 src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs create mode 100644 src/TrackCenter.Application/HostServices/WarningCalcBackgroundService.cs create mode 100644 src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index 8af3f497..32aafd8b 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs @@ -96,7 +96,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa } ); - // todo：预警计算 + // 引发预警计算 if (warningCalcOrders.Count > 0) WarningCalc_OrderDeletedOrCompleted(warningCalcOrders); } diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 57573cf2..75efca83 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -47,282 +47,177 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] - public async Task ConsumerAsync(TrackDeleteEvent @event) - { - if (!CanDelete(@event)) return; - - var trackList = await DeleteTrackAsync(@event); - - //当轨迹删除，根据轨迹标记/状态上IsToMq选择进行广播 - if (trackList.Count > 0) + public async Task ConsumerAsync(TrackDeleteEvent @event) { - _ = PublicTrackTopicAsync(@event, trackList); - //引发预警计算 - WarningCalc_DeletedTrack(trackList); + if (!CanDelete(@event)) return; + + // 处理删除轨迹 + await HandleDeleteTrackAsync(@event); + + // 根据订单关联关系异步向下刷（异步） + await DownByOrderLink(@event); } - // 根据订单关联关系异步向下刷（异步） - await DownByOrderLink(@event); - } - - private async Task DownByOrderLink(TrackDeleteEvent @event) - { - var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); - foreach (var orderId in @event.OrderIds) + private async Task DownByOrderLink(TrackDeleteEvent @event) { - foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.SubSystemId)) + var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); + foreach (var orderId in @event.OrderIds) { - var nextEvent = @event.DeepCopy(); - nextEvent.Visited.AddRange(@event.OrderIds); - nextEvent.TrackIdentifiers = @event.TrackIdentifiers; - nextEvent.SubSystemId = group.Key; - - // 不同子系统需要做内容映射 - if (@event.SubSystemId != nextEvent.SubSystemId) + foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.SubSystemId)) { - var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( - @event.SubSystemId, - nextEvent.SubSystemId, - @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() - ); + var nextEvent = @event.DeepCopy(); + nextEvent.Visited.AddRange(@event.OrderIds); + nextEvent.TrackIdentifiers = @event.TrackIdentifiers; + nextEvent.SubSystemId = group.Key; - if (codeMaps.Count == 0) + // 不同子系统需要做内容映射 + if (@event.SubSystemId != nextEvent.SubSystemId) { - LogHelper.Warn($"订单跨子系统，未配置内容映射，请联系管理员：{@event}", true); - continue; + var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( + @event.SubSystemId, + nextEvent.SubSystemId, + @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() + ); + + if (codeMaps.Count == 0) + { + LogHelper.Warn($"订单跨子系统，未配置内容映射，请联系管理员：{@event}", true); + continue; + } + + //映射轨迹code + foreach (var trackIdentity in nextEvent.TrackIdentifiers) + { + var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); + if (match == null) continue; + trackIdentity.TrackCode = match.TrackContentCode; + } } - //映射轨迹code - foreach (var trackIdentity in nextEvent.TrackIdentifiers) + foreach (var subOrderIds in group.Select(x => x.OrderId).Chunk(GlobalConst.ChunkSize)) { - var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); - if (match == null) continue; - trackIdentity.TrackCode = match.TrackContentCode; + var subEvent = nextEvent.DeepCopy(); + subEvent.OrderIds = subOrderIds; + await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); } } - - var identifiers = group.Select(x => x.OrderId).ToList(); - var idx = 0; - while (idx < identifiers.Count) - { - var subEvent = nextEvent.DeepCopy(); - var cnt = idx + GlobalConst.ChunkSize > identifiers.Count - ? identifiers.Count - idx - : GlobalConst.ChunkSize; - subEvent.OrderIds = identifiers.GetRange(idx, cnt).ToArray(); - idx += GlobalConst.ChunkSize; - - await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, subEvent); - } } } - } - private static bool CanDelete(TrackDeleteEvent @event) - { - if (@event.OrderIds.IsEmpty()) + private static bool CanDelete(TrackDeleteEvent @event) + { + if (@event.OrderIds.Length == 0) + return true; + + // 订单循环关联检测 + var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); + if (dup.Any()) + { + LogHelper.Warn($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true); + return false; + } + return true; - - // 订单循环关联检测 - var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); - if (dup.Any()) - { - LogHelper.Warn($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true); - return false; } - return true; - } - - private async Task> DeleteTrackAsync(TrackDeleteEvent @event) - { - List orderIds; - List tracks; - List deleteTracks; - // 优先根据EsId查询轨迹 - if (@event.EsIds.Length > 0) //根据明确的esid进行删除 + private async Task HandleDeleteTrackAsync(TrackDeleteEvent @event) { - deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); - orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); - @event.OrderIds = orderIds.ToArray(); - @event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto + HashSet orderIds; + List oldTracks; + List deleteTracks; + // 优先根据EsId查询轨迹 + if (@event.EsIds.Length > 0) { - TrackCode = x.TrackCode, - TrackTime = x.TrackTime - }).ToList(); + deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); - // 外部api获取的轨迹不允许删除 - if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) - return deleteTracks; + // 外部api获取的轨迹不允许删除 + if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return; - //查出订单下的所有轨迹，以供后面回退使用 - tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + // 从被删除轨迹寻找订单id，稍后会更新订单主表 + orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); - // 删除轨迹批量 - await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); + // 清理已经删除过的EsId,并设置用于下刷删除的其他信息 + @event.EsIds = Array.Empty(); + @event.OrderIds = orderIds.ToArray(); + @event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto + { + TrackCode = x.TrackCode, + TrackTime = x.TrackTime + }).ToList(); - //清理已经删除过的esid - @event.EsIds = Array.Empty(); - } - else //原有逻辑，根据轨迹code和订单进行删除 - { - //根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新 - tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); - - //比较字段为空则使用时间来作为额外的标识key - if (string.IsNullOrWhiteSpace(@event.CompareFields)) - { - //构建标识符集合 - var ids = @event.TrackIdentifiers.Select(x => $"{x.TrackCode}{x.TrackTime:G}").ToHashSet(); - //根据标识符选出要删除的轨迹明细 - deleteTracks = tracks.Where(x => ids.Contains($"{x.TrackCode}{x.TrackTime:G}")).ToList(); + // 查出订单下的所有轨迹，以供后面回退使用 + oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); } - //否则则使用比较字段,使用扩展字段中的内容进行比较 else { - var codes = @event.TrackIdentifiers.Select(x => x.TrackCode).ToImmutableHashSet(); - deleteTracks = tracks - .Where(x => + // 根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新 + oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); + + // 构建轨迹标识符 + var trackIds = @event.TrackIdentifiers.Select(y => $"{y.TrackCode}{y.GetUtc0TrackTime():G}").ToHashSet(); + + // 选出要删除的轨迹 + deleteTracks = oldTracks + .Where(x => trackIds.Contains($"{x.TrackCode}{x.GetUtc0TrackTime():G}")) + .WhereIf(!string.IsNullOrWhiteSpace(@event.CompareFields), y => { - if (!codes.Contains(x.TrackCode)) - return false; + // 扩展信息不存在，跳过 + if (string.IsNullOrWhiteSpace(y.Extra)) return false; - if (string.IsNullOrWhiteSpace(x.Extra)) - { - throw new BusinessException(message: $"当前轨迹不存在扩展信息，无法利用扩展字段：{@event.CompareFields}"); - } + // 从扩展信息提取比较值和传入值比较匹配 + var extraDict = TextJsonHelper.DeserializeToDictionary(y.Extra); + if (extraDict!.TryGetValue(@event.CompareFields!, out var compareValueJsonElement)) + return TextJsonHelper.EqJsonElement(compareValueJsonElement, @event.CompareValue); - var extraDict = TextJsonHelper.DeserializeToDictionary(x.Extra); - if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) - { - throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}"); - } - - return TextJsonHelper.EqJsonElement(compareValue, @event.CompareValue); + return false; }) .ToList(); + + //从被删除轨迹寻找订单id，稍后会更新订单主表 + orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet(); } - - if (deleteTracks.Count == 0) return deleteTracks; + if (deleteTracks.Count == 0) return; // 删除轨迹批量 await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); - //从被删除轨迹寻找订单id，稍后会更新订单主表 - orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); + // 删除轨迹后，更新订单上的轨迹信息 + await TrackDeleteAfterUpdateOrderAsync(@event, oldTracks, deleteTracks, orderIds); + + // 删除轨迹广播 + await PublicTrackTopicAsync(@event, deleteTracks); + + // 发布预警信号，稍后将进行预警计算 + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, deleteTracks); } - //排除掉已经删除的轨迹 - var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); - tracks = tracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); - - var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), - x => new OrderEntity - { - Id = x.Id, - LatestTrackStatus = x.LatestTrackStatus, - LatestTrackContent = x.LatestTrackContent, - LatestTrackEnContent = x.LatestTrackEnContent, - LatestTrackPlace = x.LatestTrackPlace, - LatestTrackTime = x.LatestTrackTime, - LatestTrackTimeZone = x.LatestTrackTimeZone, - EstimatedDeliveryDate = x.EstimatedDeliveryDate, - DestinationTimeZone = x.DestinationTimeZone - }); - if (orders.Count == 0) return tracks; - - foreach (var order in orders) + private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List delTracks) { - //找到最新的轨迹数据 - var track = tracks?.OrderByDescending(x => x.ZeroTrackTime) - .FirstOrDefault(x => x.OrderId == order.Id); + var trackContentIds = delTracks.Select(x => x.TrackContentId).ToHashSet(); + var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds); + var contentDict = trackContents.ToDictionary(x => x.Id); - // 删除全部轨迹导致的重置行为 - if (track == null) + var needPublicTracks = delTracks + // 轨迹内容已经不存在了，不需要广播 + .Where(x => contentDict.ContainsKey(x.TrackContentId)) + // 当删除操作没有标记广播时，根据轨迹状态/轨迹标记 + .WhereIf(@event.IsToMq == false, x => contentDict[x.TrackContentId].IsToMq) + .ToList(); + + if (needPublicTracks.Count == 0) return; + + //增加延时确保写盘成功 + await Task.Delay(1000); + + foreach (var track in needPublicTracks) { - if (order.LatestTrackTime == null) continue; - - // tips：订单是支持从已完成态到未完成态的 - order.ResetTrackInfo(); - } - else - { - if (order.LatestTrackContent == track.TrackContent) continue; - // 删除引发的回退更新 - order.LatestTrackStatus = track.TrackStatus; - order.LatestTrackContent = track.TrackContent; - order.LatestTrackEnContent = track.TrackEnContent; - order.LatestTrackPlace = track.TrackPlace; - order.LatestTrackTime = track.TrackTime; - order.LatestTrackTimeZone = track.TrackTimeZone; - order.EstimatedDeliveryDate = track.EstimatedDeliveryDate; - order.DestinationTimeZone = track.DestinationTimeZone; - } - } - - //更新订单上冗余轨迹明细的字段 - await _orderRepository.UpdateRangeAsync(orders, - x => new - { - x.LatestTrackStatus, - x.LatestTrackContent, - x.LatestTrackEnContent, - x.LatestTrackPlace, - x.LatestTrackTime, - x.LatestTrackTimeZone, - x.EstimatedDeliveryDate, - x.DestinationTimeZone - }); - - //批量写入订单操作日志 - _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex - { - OrderId = order.Id, - OrderNo = order.OrderNo, - OptionType = OrderOperationType.删除轨迹.ToString(), - Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(), - CreateUserId = @event.OperationManId, - CreateUserName = @event.OperationMan ?? "", - CreateTime = DateTime.Now - }).ToList()); - - return deleteTracks; - } - - private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List trackList) - { - var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray(); - var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray(); - - //从轨迹内容中查询出标记是否需要mq广播 - var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes); - var toMqDict = contentList - .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq); - - //选出需要广播的轨迹,并按code和订单号分组 - var trackGroup = trackList - .Where(x => toMqDict[x.SubSystemId + x.TrackCode]) - .GroupBy(m => new - { - m.TrackCode, - m.OrderNo - }) - .ToList(); - if (trackGroup.Count == 0) return; - - //增加延时确保写盘成功 - await Task.Delay(1000); - foreach (var dataItem in trackGroup) - { - var item = dataItem.First(); - var content = contentList.First(m => m.TrackContentCode == item.TrackCode); - - //将每个订单的每个轨迹的删除变更发布 - foreach (var track in dataItem) - { - _broadcastingService.TrackChangePublicTopic(new TrackChangeEto + var eto = new TrackChangeEto { + TrackEsId = track.Id, + Option = 2, + IsSuccess = true, + FailedReason = null, EstimatedDeliveryDate = track.EstimatedDeliveryDate, DestinationTimeZone = track.DestinationTimeZone, TrackPlacePostcode = track.TrackPlacePostcode, @@ -332,7 +227,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe TrackPlaceProvinceId = track.TrackPlaceProvinceId, TrackPlaceCity = track.TrackPlaceCity, TrackPlaceCityId = track.TrackPlaceCityId, - Files = track.Files?.Select(x => new FileItemDto() + Files = track.Files?.Select(x => new FileItemDto { FileName = x.FileName, OssKey = x.OssKey @@ -358,15 +253,94 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe OrderType = track.OrderType } }, - SubSystemId = @event.SubSystemId, + SubSystemId = track.SubSystemId, Extra = track.Extra - }, content.SubSystemCode); + }; + + var content = contentDict[track.TrackContentId]; + _broadcastingService.TrackChangePublicTopic(eto, content.SubSystemCode); } } - } - private void WarningCalc_DeletedTrack(List delTracks) - { - _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); - } + private async Task TrackDeleteAfterUpdateOrderAsync( + TrackDeleteEvent @event, + List oldTracks, + List deleteTracks, + HashSet orderIds + ) + { + // 排除掉已经删除的轨迹后的新的轨迹 + var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); + var newTracks = oldTracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); + + var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), + x => new OrderEntity + { + Id = x.Id, + LatestTrackStatus = x.LatestTrackStatus, + LatestTrackContent = x.LatestTrackContent, + LatestTrackEnContent = x.LatestTrackEnContent, + LatestTrackPlace = x.LatestTrackPlace, + LatestTrackTime = x.LatestTrackTime, + LatestTrackTimeZone = x.LatestTrackTimeZone, + EstimatedDeliveryDate = x.EstimatedDeliveryDate, + DestinationTimeZone = x.DestinationTimeZone + }); + if (orders.Count == 0) return; + + foreach (var order in orders) + { + //找到最新的轨迹数据 + var track = newTracks?.OrderByDescending(x => x.ZeroTrackTime) + .FirstOrDefault(x => x.OrderId == order.Id); + + // 删除全部轨迹导致的重置行为 + if (track == null) + { + if (order.LatestTrackTime == null) continue; + + // tips：订单是支持从已完成态到未完成态的 + order.ResetTrackInfo(); + } + else + { + if (order.LatestTrackContent == track.TrackContent) continue; + // 删除引发的回退更新 + order.LatestTrackStatus = track.TrackStatus; + order.LatestTrackContent = track.TrackContent; + order.LatestTrackEnContent = track.TrackEnContent; + order.LatestTrackPlace = track.TrackPlace; + order.LatestTrackTime = track.TrackTime; + order.LatestTrackTimeZone = track.TrackTimeZone; + order.EstimatedDeliveryDate = track.EstimatedDeliveryDate; + order.DestinationTimeZone = track.DestinationTimeZone; + } + } + + //更新订单上冗余轨迹明细的字段 + await _orderRepository.UpdateRangeAsync(orders, + x => new + { + x.LatestTrackStatus, + x.LatestTrackContent, + x.LatestTrackEnContent, + x.LatestTrackPlace, + x.LatestTrackTime, + x.LatestTrackTimeZone, + x.EstimatedDeliveryDate, + x.DestinationTimeZone + }); + + //批量写入订单操作日志 + _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex + { + OrderId = order.Id, + OrderNo = order.OrderNo, + OptionType = OrderOperationType.删除轨迹.ToString(), + Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(), + CreateUserId = @event.OperationManId, + CreateUserName = @event.OperationMan ?? "", + CreateTime = DateTime.Now + }).ToList()); + } } diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 898a113b..e47958d5 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -24,7 +24,6 @@ namespace TrackCenter.Application.CAP.EventHandlers; public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { - private readonly IHttpClientProxy _httpClientProxy; private readonly IOptionsMonitor _options; private readonly IOrderRepository _orderRepository; private readonly TrackPushCacheService _trackPushCacheService; @@ -35,7 +34,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private List _pushConfigs = null!; private Dictionary> _orderTypeSortedConfigsDict = null!; //订单类型,根据sort从小到大排序后的配置字典 - public TrackPushEventHandler(IHttpClientProxy httpClientProxy, + public TrackPushEventHandler( IOrderRepository orderRepository, TrackPushCacheService trackPushCacheService, ITrackPushRecordEsRepository trackPushRecordEsRepository, @@ -43,7 +42,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe BroadcastingService broadcastingService, ITrackDetailsEsRepository trackDetailsEsRepository) { - _httpClientProxy = httpClientProxy; _orderRepository = orderRepository; _trackPushCacheService = trackPushCacheService; _trackPushRecordEsRepository = trackPushRecordEsRepository; @@ -61,7 +59,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (pushConfigs == null || pushConfigs.Count == 0) return; if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace()) { - throw new BusinessException(GlobalConst.CapNoRetryErrorCode, $"该客户{@event.CustomId}未关联服务商，无法推送"); + throw new CapNoRetryException($"该客户{@event.CustomId}未关联服务商，无法推送"); } _pushConfigs = pushConfigs; @@ -132,20 +130,20 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; - var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( - _options.CurrentValue.PushTrackInterfaceCode, - serviceProviderCode, - items.Cast().ToList(), - GlobalConst.TrackPushCompletedSubscribe, - attach: asciiString); - - if (r.Code != GlobalConst.SuccessCode) - { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, - items.Cast().ToList()); - throw new BusinessException( - message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); - } + // var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + // _options.CurrentValue.PushTrackInterfaceCode, + // serviceProviderCode, + // items.Cast().ToList(), + // GlobalConst.TrackPushCompletedSubscribe, + // attach: asciiString); + // + // if (r.Code != GlobalConst.SuccessCode) + // { + // TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + // items.Cast().ToList()); + // throw new BusinessException( + // message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + // } } } @@ -354,20 +352,20 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { var asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; - var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( - _options.CurrentValue.TrackUpdateInterfaceCode, - serviceProviderCode, - items.Cast().ToList(), - GlobalConst.TrackPushCompletedSubscribe, - attach: asciiString); - - if (r.Code != GlobalConst.SuccessCode) - { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, - items.Cast().ToList()); - throw new BusinessException( - message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); - } + // var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + // _options.CurrentValue.TrackUpdateInterfaceCode, + // serviceProviderCode, + // items.Cast().ToList(), + // GlobalConst.TrackPushCompletedSubscribe, + // attach: asciiString); + // + // if (r.Code != GlobalConst.SuccessCode) + // { + // TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + // items.Cast().ToList()); + // throw new BusinessException( + // message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + // } } } diff --git a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index ba791528..6a5bc0e1 100644 --- a/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -114,17 +114,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe OrderId = x.OrderId, OrderNo = x.OrderNo, OrderType = x.OrderType, - }).ToList(); - var idx = 0; - while (idx < identifiers.Count) + }); + + foreach (var subOrderIds in identifiers.Chunk(GlobalConst.ChunkSize)) { var subEvent = nextEvent.DeepCopy(); - var cnt = idx + GlobalConst.ChunkSize > identifiers.Count - ? identifiers.Count - idx - : GlobalConst.ChunkSize; - subEvent.OrderIdentifiers = identifiers.GetRange(idx, cnt).ToArray(); - idx += GlobalConst.ChunkSize; - + subEvent.OrderIdentifiers = subOrderIds.ToArray(); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); } } diff --git a/src/TrackCenter.Application/Dtos/ProvideServices/TrackIdentifierDto.cs b/src/TrackCenter.Application/Dtos/ProvideServices/TrackIdentifierDto.cs index 772e82ff..f0b94e95 100644 --- a/src/TrackCenter.Application/Dtos/ProvideServices/TrackIdentifierDto.cs +++ b/src/TrackCenter.Application/Dtos/ProvideServices/TrackIdentifierDto.cs @@ -18,4 +18,20 @@ public sealed class TrackIdentifierDto /// [Required] public DateTime TrackTime { get; set; } + + /// + /// 轨迹时区 + /// + public float? TrackTimeZone { get; set; } + + /// + /// 获取UTC0时区时间 (传null/99999视为+8时区) + /// + public DateTime GetUtc0TrackTime() + { + var timezone = TrackTimeZone ?? 99999f; + return Math.Abs(timezone - 99999f) < 0.01f + ? TrackTime.AddHours(-8.0f) + : TrackTime.AddHours(-timezone); + } } diff --git a/src/TrackCenter.Application/Dtos/TrackPull/TrackPullParamConfigDto.cs b/src/TrackCenter.Application/Dtos/TrackPull/TrackPullParamConfigDto.cs index c9d45d68..b79437f9 100644 --- a/src/TrackCenter.Application/Dtos/TrackPull/TrackPullParamConfigDto.cs +++ b/src/TrackCenter.Application/Dtos/TrackPull/TrackPullParamConfigDto.cs @@ -27,6 +27,16 @@ public class TrackPullParamConfigDto /// public string? TrackStatusName { get; set; } + /// + /// 轨迹扩展信息参数名集合（逗号分割） + /// + public string? TrackExtraNames { get; set; } + + /// + /// 轨迹UTC时间参数名 + /// + public string? TrackUtcTimeName { get; set; } + /// /// 轨迹发生ISO时间参数名 /// diff --git a/src/TrackCenter.Application/Events/ModifyWarningEvent.cs b/src/TrackCenter.Application/Events/ModifyWarningEvent.cs deleted file mode 100644 index 007b8b72..00000000 --- a/src/TrackCenter.Application/Events/ModifyWarningEvent.cs +++ /dev/null @@ -1,22 +0,0 @@ -namespace TrackCenter.Application.Events; - -/// -/// redis 的发布订阅事件 -/// -public class ModifyWarningEvent : RedisEvent -{ - public string Key { get; set; } -} - -public class RedisEvent : IRedisEvent -{ - public string EventType { get; set; } -} - -/// -/// 简单的事件接口，后续可以根据需要扩展 -/// -public interface IRedisEvent -{ - string EventType { get; set; } -} diff --git a/src/TrackCenter.Application/Events/PushWarningEvent.cs b/src/TrackCenter.Application/Events/PushWarningEvent.cs deleted file mode 100644 index 658a3d13..00000000 --- a/src/TrackCenter.Application/Events/PushWarningEvent.cs +++ /dev/null @@ -1,6 +0,0 @@ -namespace TrackCenter.Application.Events; - -public class PushWarningEvent -{ - public int PageSize { get; set; } -} diff --git a/src/TrackCenter.Application/Events/TrackWarningCalcEvent.cs b/src/TrackCenter.Application/Events/TrackWarningCalcEvent.cs deleted file mode 100644 index 8f6d6203..00000000 --- a/src/TrackCenter.Application/Events/TrackWarningCalcEvent.cs +++ /dev/null @@ -1,29 +0,0 @@ -namespace TrackCenter.Application.Events; - -public class TrackWarningCalcEvent -{ - /// - /// 子系统id - /// - public int SubSystemId { get; set; } - - /// - /// 客户id - /// - public int CustomId { get; set; } - - /// - /// 轨迹状态Id - /// - public int TrackStatusId { get; set; } - - /// - /// 订单id - /// - public int[] OrderIds { get; set; } - - /// - /// 轨迹发生时间 - /// - public DateTime TrackTime { get; set; } -} diff --git a/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs b/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs deleted file mode 100644 index 709ea345..00000000 --- a/src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs +++ /dev/null @@ -1,120 +0,0 @@ -using FeishuNetSdk; -using FeishuNetSdk.Im; -using FeishuNetSdk.Im.Dtos; - -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 SendMessageByTemplateInput BuildZhanNeiXinMessage(WarningNotifyParam param, int systemId) - { - const string templateCode = "TrackWarningNotify"; // 模板code - const string contentKey = "content"; //内容参数 - var dateTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"); - var totalCount = param.TransportCount + param.ReturnCount + param.CompleteCount; - - // todo:无法使用a标签填充（可能是为了安全考虑），需要具体给定一个样式 - var url1 = ""; - var url2 = ""; - var url3 = ""; - - var content = $"{Title}\n尊敬的用户，截止到{dateTime}({BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下："; - - if (param.TransportCount > 0) - content += $"\n运输时效预警共计{param.TransportCount}条，其中过去{param.Hours}小时内新增的预警{param.TransportLastNHoursCount}条"; - if (param.ReturnCount > 0) - content += $"\n回传时效预警共计{param.ReturnCount}条，其中过去{param.Hours}小时内新增的预警{param.ReturnLastNHoursCount}条"; - if (param.CompleteCount > 0) - content += $"\n轨迹完整预警共计{param.CompleteCount}条，其中过去{param.Hours}小时内新增的预警{param.CompleteLastNHoursCount}条"; - - // 构建用户中心发消息请求体对象 - var dto = new SendMessageByTemplateInput - { - IsNeedToHandle = true, - MsgSource = MsgSourceEnum.消息, - SystemId = systemId, - templateCode = templateCode, - channels = new List(1) - { - ChannelEnum.站内信 - }, - variables = new List(1) - { - new VariableDto - { - Key = contentKey, - Value = content - } - }, - receivers = new List(1) - { - new MsgInstanceReceiverInput - { - RecipientType = 1, //接收人类型 (1:内部用户 2.外部用户) - ReceiverId = param.UserId, //接收人id - ReceiverName = null, //接收人名称 - Mail = null, // 邮箱 - MobilePhone = null, // 手机号 - } - }, - }; - - return dto; - } - - // 构建飞书卡片发消息请求体 - public static PostImV1MessagesBodyDto BuildFeiShuCard(WarningNotifyParam param, string fsUserId) - { - 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") - }; - // todo:补充点击链接 - var url1 = ""; - var url2 = ""; - var url3 = ""; - if (param.TransportCount > 0) - body.Add(new MarkdownElement(Content: $"运输时效预警共计{param.TransportCount}条，其中过去{param.Hours}小时内新增的预警{param.TransportLastNHoursCount}条 [查看详情]({url1})", TextAlign: "left", TextSize: "normal_v2")); - if (param.ReturnCount > 0) - body.Add(new MarkdownElement(Content: $"轨迹回传时效预警共计{param.ReturnCount}条，其中过去{param.Hours}小时内新增的预警{param.ReturnLastNHoursCount}条 [查看详情]({url2})", TextAlign: "left", TextSize: "normal_v2")); - if (param.CompleteCount > 0) - 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}"; - } -} diff --git a/src/TrackCenter.Application/Hangfire/WarningTaskService.cs b/src/TrackCenter.Application/Hangfire/WarningTaskService.cs index 80c01639..800cef27 100644 --- a/src/TrackCenter.Application/Hangfire/WarningTaskService.cs +++ b/src/TrackCenter.Application/Hangfire/WarningTaskService.cs @@ -3,21 +3,15 @@ using Fake.DependencyInjection; using FeishuNetSdk; using Hangfire; using Microsoft.Extensions.Options; -using SJZY.Expand.ABP.Core.Logger; -using SJZY.InterfaceRelay.Application.Contracts; using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Domain.Aggregates.WarningAggregate; using TrackCenter.Domain.Aggregates.WarningAggregate.Return; using TrackCenter.Domain.Dtos.Warning; using TrackCenter.Domain.Enums; using TrackCenter.Domain.Extensions; -using TrackCenter.Domain.Queries; -using SJZY.UserCenter.HttpApi.Services.Message; using TrackCenter.Application.Helpers; using TrackCenter.Application.Options; using TrackCenter.Application.Track2InterfaceRelay; -using Volo.Abp.DependencyInjection; -using Volo.Abp.Http.Client.DynamicProxying; namespace TrackCenter.Application.Hangfire; @@ -34,31 +28,22 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency { private readonly IOptionsMonitor _interfaceRelayOptions; private readonly IOptionsMonitor _warningTaskOptions; - private readonly IHttpClientProxy _requestSendMessageService; - private readonly IHttpClientProxy _interfaceForwardServiceV2; private readonly IFeishuTenantApi _feishuTenantApi; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly IWarningTaskRepository _warningTaskRepository; - private readonly IVFsUserQueries _vfsUserQueries; public WarningTaskService( IOptionsMonitor interfaceRelayOptions, IOptionsMonitor warningTaskOptions, - IHttpClientProxy interfaceForwardServiceV2, - IHttpClientProxy requestSendMessageService, IFeishuTenantApi feishuTenantApi, IWarningDetailsEsRepository warningDetailsEsRepository, - IWarningTaskRepository warningTaskRepository, - IVFsUserQueries vfsUserQueries) + IWarningTaskRepository warningTaskRepository) { _interfaceRelayOptions = interfaceRelayOptions; _warningTaskOptions = warningTaskOptions; - _interfaceForwardServiceV2 = interfaceForwardServiceV2; - _requestSendMessageService = requestSendMessageService; _feishuTenantApi = feishuTenantApi; _warningDetailsEsRepository = warningDetailsEsRepository; _warningTaskRepository = warningTaskRepository; - _vfsUserQueries = vfsUserQueries; } #region 用户中心调用，通知（一次） @@ -72,17 +57,8 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency // 发送站内信 // 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 - var r = await _requestSendMessageService.Service.SendMessageByTemplateAsync(WarningNotifyBuilder.BuildZhanNeiXinMessage(param, options.MiddleSystemId)); - if (r.Code != GlobalConst.SuccessCode) - throw new BusinessException(message: $"用户中心调用失败：SendMessageByTemplateAsync Msg:{r.Msg}"); // 发送FeiShu卡片信息 - var fsUserId = await _vfsUserQueries.GetFirstAsync(x => x.UserCenterId == param.UserId, y => y.FsEmployeeId); - if (string.IsNullOrWhiteSpace(fsUserId)) - throw new BusinessException(message: "查询用户中心失败：SendMessageByTemplateAsync Msg:未找到用户对应的飞书用户Id"); - var res = await _feishuTenantApi.PostImV1MessagesAsync("user_id", WarningNotifyBuilder.BuildFeiShuCard(param, fsUserId)); - if (res.Code != 0) - throw new BusinessException(message: $"飞书调用失败：SendMessageByTemplateAsync Msg:{res.Msg}"); } #endregion @@ -118,24 +94,24 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency ? _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 _interfaceForwardServiceV2.Service.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};SubSystemCode:{task.SubSystemCode};" + - $" 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};SubSystemCode:{task.SubSystemCode};" + + // $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}"; + // + // // 处理调用接口通失败，直接处于完结状态 + // await HandlerCallBusinessException(task, msg); + // retryCountChanged = true; + // throw new BusinessException(message: msg); + // } } } catch (Exception e) diff --git a/src/TrackCenter.Application/HostServices/WarningCalcBackgroundService.cs b/src/TrackCenter.Application/HostServices/WarningCalcBackgroundService.cs new file mode 100644 index 00000000..02473c2e --- /dev/null +++ b/src/TrackCenter.Application/HostServices/WarningCalcBackgroundService.cs @@ -0,0 +1,284 @@ +using System.Diagnostics; +using Confluent.Kafka; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; +using TrackCenter.Application.Kafka; +using TrackCenter.Application.Options; +using TrackCenter.Application.Services.Warning.Handlers; +using TrackCenter.Application.Services.Warning.Signals; +using TrackCenter.Domain.Extensions; +using TrackCenter.Repository.Otel; + +namespace TrackCenter.Application.HostServices; + +/// +/// 预警计算后台任务 +/// 预警信号从kafka摄取来进行批处理 +/// +public sealed class WarningCalcBackgroundService : BackgroundService +{ + private readonly IServiceScopeFactory _serviceScopeFactory; + private readonly IConsumer _consumer; + private readonly WarningSignalPublisher _warningSignalPublisher; + private TrackWarningTaskOptions _options; + + public WarningCalcBackgroundService( + IOptionsMonitor optionsMonitor, + IServiceScopeFactory serviceScopeFactory, + IConsumer consumer, + WarningSignalPublisher warningSignalPublisher + ) + { + _serviceScopeFactory = serviceScopeFactory; + _consumer = consumer; + _warningSignalPublisher = warningSignalPublisher; + _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(); + _activity = null; + } + } + + 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(); + } + + private async ValueTask MessageHandler(IEnumerable> messages, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + + // 1.根据类型分组信号 + var typeSignalDict = new Dictionary>(); + var needRetryMetadata = new List(); // 需要重试的消息 + var needEnqueueDlxMetadata = new List(); // 需要投递到死信队列的 + var needHandleMessage = new List>(); + foreach (var m in messages) + { + // tips:不支持的消息需要及时上报 + var mType = m.Key.Type; + var signal = m.Value; + if (!CheckSupportMessage(mType)) + { + needEnqueueDlxMetadata.Add(WarningSignalMetadata.FormMessageOrCreate(m, false, true, errorReason: $"不支持的消息类型：{mType}")); + continue; + } + + var list = typeSignalDict.GetValueOrDefault(mType, new List()); + list.Add(signal); + typeSignalDict[mType] = list; + needHandleMessage.Add(m); + } + + // tips:上报异常(类型异常) + Report_AggregateNoSupport(needEnqueueDlxMetadata); + + // 2.处理消息 + if (typeSignalDict.Count > 0) + { + var (success, error) = await TrackSignalHandler(typeSignalDict, ct); + + if (!success) + { + // 取消异常，这里只认为服务器下线时引发的，直接中断即可，等待重分配 + if (error is TaskCanceledException) + { + // 被取消不需要增加重试次数 + //needRetryMetadata.AddRange(needHandleMessage.Select(x => TrackSignalMetadata.FormMessageOrCreate(x, false, false, error: error))); + LogHelper.Warn($"处理被取消，消息会重新分配到其他Host,影响消息条数[{needHandleMessage.Count}]"); + throw error; + } + + // tips：处理消息出错，批量上报异常指明出错的详细原因 + foreach (var metadata in needHandleMessage.Select(x => WarningSignalMetadata.FormMessageOrCreate(x, true, false, error: error!))) + { + // 超过了最大的重试次数，此时应该投递到死信队列 + if (metadata.RetryCount >= _options.WarningCalcMaxRetryCount) + { + needEnqueueDlxMetadata.Add(metadata); + continue; + } + + needRetryMetadata.Add(metadata); + } + } + } + + // 3.需要重试的消息,重新投递到原队列 + if (needRetryMetadata.Count > 0) + { + _activity?.SetTag("StartPublicRetryMessage", $"[预警计算] 开始重新投递[{needRetryMetadata.Count}]条消息"); + LogHelper.Warn($"[预警计算] 开始重新投递[{needRetryMetadata.Count}]条消息"); + + var tasks = needRetryMetadata.Select(async metadata => + { + metadata.SetTrackSignalMessageHeader(); + await _warningSignalPublisher.PublishSignalAsync(metadata.OwenMessage, CancellationToken.None); + }).ToArray(); + await Task.WhenAll(tasks); + + 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.进入死信的消息，投入到死信队列，等待人工处理 + 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 _warningSignalPublisher.PublishDlxSignalAsync(new Message + { + Key = msg.Value.OrderId, + Value = msg.Value, + Headers = msg.Headers + }, CancellationToken.None); + }).ToArray(); + await Task.WhenAll(tasks); + + 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); + } + } + + private static bool CheckSupportMessage(WarningSignalType type) + { + return type switch + { + WarningSignalType.轨迹新增 => true, + WarningSignalType.轨迹更新 => true, + WarningSignalType.轨迹删除 => true, + _ => false + }; + } + + private async Task<(bool Success, Exception? error)> TrackSignalHandler(Dictionary> typeSignalDict, CancellationToken ct) + { + var scope = _serviceScopeFactory.CreateAsyncScope(); + try + { + ct.ThrowIfCancellationRequested(); + var handler = scope.ServiceProvider.GetRequiredService(); + var handler2 = scope.ServiceProvider.GetRequiredService(); + var handler3 = scope.ServiceProvider.GetRequiredService(); + + // tips:处理失败视为整体失败，消息会被重新投递，考虑减小处理批次大小？ + var addSignalr = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹新增, Empty.List); + var updateSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹更新, Empty.List); + var deleteSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹删除, Empty.List); + + // 增 + if (addSignalr.Count > 0) + await handler.TrackAddedHandleAsync(addSignalr); + // 删 + if (deleteSignals.Count > 0) + { + await handler.TrackDeletedHandleAsync(deleteSignals); + await handler2.TrackDeletedHandleAsync(deleteSignals); + } + + // 增+改 + var addOrUpdateSignal = new List(addSignalr.Count + updateSignals.Count); + addOrUpdateSignal.AddRange(addSignalr); + addOrUpdateSignal.AddRange(updateSignals); + if (addOrUpdateSignal.Count > 0) + await handler2.TrackAddedOrUpdatedHandleAsync(addOrUpdateSignal); + + // 改+删 + var updateOrDeleteSignal = new List(updateSignals.Count + deleteSignals.Count); + updateOrDeleteSignal.AddRange(updateSignals); + updateOrDeleteSignal.AddRange(deleteSignals); + if (updateOrDeleteSignal.Count > 0) + await handler3.TrackUpdatedOrDeletedHandleAsync(updateOrDeleteSignal); + + return (true, null); + } + catch (Exception e) + { + return (false, e); + } + finally + { + await scope.DisposeAsync(); + } + } + + private void Report_AggregateNoSupport(List ms) + { + if (ms.Count == 0) return; + var reasons = string.Join(",", ms.GroupBy(x => x.LastErrorReason).Select(x => x.Key)); + _activity?.SetTag("AggregateNoSupport", reasons); + LogHelper.Error($"预警信号类型不支持：{reasons}"); + } + + public override void Dispose() + { + base.Dispose(); + _consumer.Close(); + _consumer.Dispose(); + } +} diff --git a/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs b/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs new file mode 100644 index 00000000..621f1f5e --- /dev/null +++ b/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs @@ -0,0 +1,81 @@ +using Hangfire; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using TrackCenter.Application.Hangfire; +using TrackCenter.Application.Helpers; +using TrackCenter.Application.Options; + +namespace TrackCenter.Application.HostServices; + +/// +/// 预警任务组件初始化 +/// tips:支持定时任务执行周期配置 动态更新并应用 √ +/// +public class WarningTaskComponentHostedService : IHostedService +{ + private readonly IOptionsMonitor _optionsMonitor; + private TrackWarningTaskOptions? _oldOptions; + private readonly ILogger _logger; + + public WarningTaskComponentHostedService(IOptionsMonitor optionsMonitor, ILogger logger) + { + _optionsMonitor = optionsMonitor; + _logger = logger; + // 防抖配置变更包装 + var onChanged = DebounceHelper.WrapMethod(TimeSpan.FromSeconds(1), () => { ReloadConfig(optionsMonitor); }); + optionsMonitor.OnChange(_ => onChanged.Debounce()); + } + + public Task StartAsync(CancellationToken cancellationToken) + { + // 立即初始化一次 + ReloadConfig(_optionsMonitor); + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken) + { + 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/Rabbit/TrackPushCompletedEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs index 413e5685..fbd02b20 100644 --- a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -4,7 +4,6 @@ using Fake.DependencyInjection; using Newtonsoft.Json; using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPush; -using TrackCenter.Application.Services; using TrackCenter.Application.Helpers; using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.Warning.Handlers; @@ -42,15 +41,28 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl 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) + var msg = string.Empty; + try { - LogHelper.Warn( - $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); - return; - } + msg = Encoding.UTF8.GetString(args.Body.ToArray()); + var @event = System.Text.Json.JsonSerializer.Deserialize(msg); + if (@event == null) + { + LogHelper.Warn( + $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); + return; + } + await ConsumerAsync(@event); + } + catch (Exception ex) + { + LogHelper.Error($"TrackPushCompletedEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}"); + } + } + + public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) + { var optionType = Base64Helper.DeSerialize(@event.Attach); var pushInputs = new List(); var updateInputs = new List(); @@ -91,18 +103,8 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl // 同步更新订单主表 await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs); - // todo:引发预警计算 改MQ批量+分流 + // 引发预警计算 await _returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs); - - // // 推送广播 - // if (log.IsToMq) - // _broadcastingService.TrackPushCompletedPublicTopic(BuildDto(log)); - // - // // 往es 写推送日志 - // await _trackPushRecordEsRepository.AddAsync(log); - // - // // 同步去更新订单主表 - // await TrackPushAfterUpdateOrderAsync(log); } private async Task TrackPushAfterUpdateOrderAsync(bool isAdd, bool isSuccess, List logs) @@ -148,47 +150,6 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl } } - private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log) - { - // tips：只有新增会影响推送进度？ - if (log.OptionType != OptionType.新增.ToString()) return; - - var order = new OrderEntity - { - Id = log.OrderId - }; - - /* - * 异常：最近一条推送是失败 - */ - if (!log.IsSuccess) - { - order.QuestionType = (int) TrackProblemType.Error; - order.LatestPushError = log.Message; - await _orderRepository.UpdateColumnAsync(order, x => new - { - x.QuestionType, - x.LatestPushError - }); - } - else - { - // 还原成正常态 - order.QuestionType = (int) TrackProblemType.None; - order.LatestPushError = null; - order.LatestPushTrackCode = log.TrackCode; - order.LatestPushTime = log.CreateTime; - await _orderRepository.UpdateColumnAsync(order, - x => new - { - x.LatestPushTrackCode, - x.LatestPushTime, - x.QuestionType, - x.LatestPushError - }); - } - } - private static TrackPushRecordDto BuildDto(TrackPushRecordIndex log, string interfaceRelayData) { return new TrackPushRecordDto diff --git a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs index 378382d9..31508b43 100644 --- a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs @@ -14,6 +14,7 @@ using TrackCenter.Application.Services.TrackPull; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Enums; +using TrackCenter.Domain.Extensions; namespace TrackCenter.Application.Rabbit; @@ -39,7 +40,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler public ConsumeOptions Options() { - return new ConsumeOptions() + return new ConsumeOptions { Queue = GlobalConst.TrackRefreshSubscribe, }; @@ -47,16 +48,24 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler 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) + var msg = string.Empty; + try { - _logger.LogWarning( - $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); - return; - } + 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); + await ConsumerAsync(@event); + } + catch (Exception ex) + { + LogHelper.Error($"TrackRefreshEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}"); + } } public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) @@ -102,7 +111,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler TrackRefreshAttach attach) { var paramConfig = paramsWithContent.ParamConfig; - var orderNo = orderToken[paramConfig.OrderNoName]?.Value(); + var orderNo = orderToken.SelectToken(paramConfig.OrderNoName)?.Value(); if (orderNo.IsNullOrWhiteSpace()) return; var trackToken = orderToken.SelectToken(paramConfig.TrackDetailsPath); @@ -158,14 +167,14 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler // 轨迹内容 var trackContent = paramConfig.TrackContentName == null ? null - : trackToken[paramConfig.TrackContentName]?.Value(); + : trackToken.SelectToken(paramConfig.TrackContentName)?.Value(); // 1、匹配轨迹代码 // 2、匹配轨迹状态 if (matchedContentConfig == null && !paramConfig.TrackStatusName.IsNullOrWhiteSpace()) { - var trackStatus = trackToken[paramConfig.TrackStatusName!]?.Value(); + var trackStatus = trackToken.SelectToken(paramConfig.TrackStatusName!)?.Value(); if (!trackStatus.IsNullOrWhiteSpace()) { matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValues.Contains(trackStatus!)); @@ -211,26 +220,36 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler // 轨迹发生ISO时间 var trackIsoTime = paramConfig.TrackIsoTimeName == null ? null - : trackToken[paramConfig.TrackIsoTimeName]?.ToObject(); + : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.ToObject(); + // 轨迹发生UTC时间 + var trackUtcTime = paramConfig.TrackUtcTimeName == null + ? null + : trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.ToObject(); // 轨迹发生时间 var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null ? null - : trackToken[paramConfig.TrackTimeName]?.ToObject()); + : trackToken.SelectToken(paramConfig.TrackTimeName)?.ToObject()); // 轨迹发生时区 var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null ? null - : trackToken[paramConfig.TrackTimeZoneName]?.ToObject()); + : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.ToObject()); // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName == null ? null - : trackToken[paramConfig.TrackPlaceName]?.ToObject(); + : trackToken.SelectToken(paramConfig.TrackPlaceName)?.ToObject(); + // 轨迹扩展信息，根据输入的多个字段提取报文内容，构建一个新的对象的json形式 + var trackExtra = paramConfig.TrackExtraNames == null + ? null + : ExtraBuild(trackToken, paramConfig.TrackExtraNames); + // 计算时间和时区信息 + (trackTime, trackTimeZone) = CalcTimeAndTimezone(trackIsoTime, trackUtcTime, trackTime, trackTimeZone); if (matchedContentConfig is null) { _logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}", context.OrderNo, trackToken.ToString()); - await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, trackContent); + await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime.Value, trackTimeZone.Value, trackContent, trackExtra); return; } @@ -257,9 +276,10 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler CustomId = context.CustomId, TrackCode = matchedContentConfig.TrackContentCode, TrackPlace = trackPlace ?? GlobalConst.EmptyString, - TrackTime = trackTime ?? DateTime.Now, - TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, + TrackTime = trackTime.Value, + TrackTimeZone = trackTimeZone.Value, IsAdd = true, + //S06新增 //服务商轨迹默认不对外 IsExternal = false, Source = (int) TrackSourceEnum.ExternalServiceProvide, @@ -276,13 +296,14 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler TrackPlacePostcode = null, DestinationTimeZone = null, EstimatedDeliveryDate = null, + Extra = trackExtra }; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, - DateTime? trackTime, float? trackTimeZone, string? trackContent) + DateTime trackTime, float trackTimeZone, string? trackContent, string? extra) { // 新增服务商轨迹到es var track = new TrackDetailsIndex @@ -293,8 +314,8 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler OrderType = context.OrderType, OrderNo = context.OrderNo, TrackPlace = trackPlace ?? GlobalConst.EmptyString, - TrackTime = trackTime ?? DateTime.Now, - TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, + TrackTime = trackTime, + TrackTimeZone = trackTimeZone, ServiceProviderCode = context.ServiceProviderCode, OriginTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)), OriginTrackContent = trackContent ?? GlobalConst.EmptyString, @@ -307,6 +328,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler TrackContentId = GlobalConst.NotExistId, TrackContent = GlobalConst.EmptyString, TrackEnContent = GlobalConst.EmptyString, + //S06新增轨迹明细字段 //服务商轨迹默认不对外 IsExternal = false, Source = (int) TrackSourceEnum.ExternalServiceProvide, @@ -321,7 +343,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler TrackPlacePostcode = null, DestinationTimeZone = null, EstimatedDeliveryDate = null, - Extra = null + Extra = extra }; track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 @@ -330,4 +352,69 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler await _trackDetailsEsRepository.AddAsync(track); } + + private static string? ExtraBuild(JToken jToken, string trackExtraNames) + { + const string separator = ","; + var trackExtraNamesArray = trackExtraNames.Split(separator); + if (trackExtraNamesArray.Length == 0) return null; + + var jObject = new JObject(); + foreach (var fieldName in trackExtraNamesArray) + { + var token = jToken.SelectToken(fieldName); + if (token == null) continue; + jObject.Add(fieldName, token); + } + + return jObject.ToString(); + } + + // 原始时间（utc0+时区），时区 + private static (DateTime Time, float TimeZone) CalcTimeAndTimezone(DateTimeOffset? isoTime, DateTime? utcTime, DateTime? unknownTime, float? timezone) + { + // 存在iso时间 + if (isoTime.HasValue) + { + return (isoTime.Value.DateTime, (float) isoTime.Value.Offset.TotalHours); + } + + // 存在时区 + if (timezone.HasValue) + { + if (utcTime.HasValue) + return (utcTime.Value.AddHours(timezone.Value), timezone.Value); + + if (unknownTime.HasValue) + return (unknownTime.Value, timezone.Value); + + // tips:不存在时间信息，使用一个非常小的时间来代表不存在！ + return (GlobalConst.MinDateTime, timezone.Value); + } + + if (utcTime.HasValue && unknownTime.HasValue) + { + // 可计算出时区信息 + if (utcTime != unknownTime) + { + var timeZone = (float) (unknownTime - utcTime).Value.TotalHours; + return (unknownTime.Value, timeZone); + } + + // 只有utc时间，存为0时区 + return (utcTime.Value, 0); + } + + if (utcTime.HasValue) + { + return (utcTime.Value, 0); + } + + if (unknownTime.HasValue) + { + return (unknownTime.Value, GlobalConst.NoneTrackTimeZone); + } + + return (GlobalConst.MinDateTime, GlobalConst.NoneTrackTimeZone); + } } diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs index 88496128..03036289 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs @@ -1,14 +1,10 @@ using DotNetCore.CAP; using Fake.Application; using InterfaceForward.Domain.Shared.Helpers; -using SJZY.Expand.ABP.Core.Logger; -using SJZY.Expand.ABP.Core.PublicMethords; using TrackCenter.Application.Dtos.ProvideServices; -using TrackCenter.Application.Services.ProvideServices; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Extensions; -using SJZY.UserCenter.HttpApi.Services; using SqlSugar; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Repository.Aggregates.OrderAggregate; @@ -20,7 +16,7 @@ namespace TrackCenter.Application.Services.ProvideServices; /// [ApiExplorerSettings(GroupName = "提供接口")] [Route("/ProviderService")] -public partial class OrderService : ApplicationServiceBase, IOrderService +public partial class OrderService : ApplicationServiceBase { private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; private readonly IOrderRepository _orderRepository; @@ -95,19 +91,19 @@ public partial class OrderService : ApplicationServiceBase, IOrderService { OrderId = orderId, OrderNo = input.OrderNo, - OptionType = OrderOperationType.创建订单.ToString(), + OptionType = nameof(OrderOperationType.创建订单), Content = input.ToJson(), - CreateUserId = currentUser.UserId, - CreateUserName = currentUser.Name, + CreateUserId = CurrentUser.Id ?? Guid.Empty, + CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }; _ = _orderOperationRecordEsRepository.AddAsync(indexForAdd); - return new UnifyResultDto(true, new IdBaseDto + return new IdBaseDto { Id = orderId - }); + }; } /// @@ -117,7 +113,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService /// /// [HttpPost("Order/BatchUpsert")] - public async Task BatchUpsertAsync(BatchUpsertOrderInput input) + public async Task BatchUpsertAsync(BatchUpsertOrderInput input) { if (input.OrderInputs.Any(x => (x.EstimatedDeliveryDate.HasValue && x.DestinationTimeZone.HasValue == false) || (x.EstimatedDeliveryDate.HasValue == false && x.DestinationTimeZone.HasValue))) @@ -125,8 +121,6 @@ public partial class OrderService : ApplicationServiceBase, IOrderService throw new BusinessException(message: "【预计签收时间】和【签收地时区】必须同时输入"); } - var currentUser = CurrentUserRequest.CurrentUser; - var exp = Expressionable.Create(); foreach (var order in input.OrderInputs) exp.Or(x => x.OrderType == order.OrderType && x.SubSystemId == order.SubSystemId && x.OrderNo == order.OrderNo); @@ -150,10 +144,10 @@ public partial class OrderService : ApplicationServiceBase, IOrderService { OrderId = order.Id, OrderNo = order.OrderNo, - OptionType = OrderOperationType.修改订单.ToString(), + OptionType = nameof(OrderOperationType.修改订单), Content = input2.ToJson(), - CreateUserId = currentUser.UserId, - CreateUserName = currentUser.Name, + CreateUserId = CurrentUser.Id ?? Guid.Empty, + CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }); } @@ -163,8 +157,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService } // 如果全部操作为修改 - if (existsOrders.Count == input.OrderInputs.Count) - return new UnifyResultDto(true); + if (existsOrders.Count == input.OrderInputs.Count) return; // 新增订单 var existsOrderNos = existsOrders.Select(x => x.OrderNo).ToHashSet(); @@ -188,8 +181,8 @@ public partial class OrderService : ApplicationServiceBase, IOrderService OrderNo = order.OrderNo, OptionType = OrderOperationType.创建订单.ToString(), Content = insertInput.ToJson(), - CreateUserId = currentUser.UserId, - CreateUserName = currentUser.Name, + CreateUserId = CurrentUser.Id ?? Guid.Empty, + CreateUserName = CurrentUser.UserName ?? string.Empty, CreateTime = DateTime.Now }); } @@ -203,7 +196,6 @@ public partial class OrderService : ApplicationServiceBase, IOrderService } _ = _orderOperationRecordEsRepository.AddManyAsync(createRecords); - return new UnifyResultDto(true); } /// @@ -213,7 +205,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService /// /// [HttpPost("Order/Delete")] - public async Task> DeleteOrderAsync(DeleteOrderInput input) + public async Task DeleteOrderAsync(DeleteOrderInput input) { var order = await _orderRepository.GetFirstAsync(x => x.SubSystemId == input.SubSystemId && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo); @@ -224,13 +216,13 @@ public partial class OrderService : ApplicationServiceBase, IOrderService await _orderRepository.DeleteOrderAsync(order.Id); - // todo:预警计算 + // 预警计算 WarningCalc_OrderDeletedOrCompleted(order); - return new UnifyResultDto(true, new IdBaseDto + return new IdBaseDto { Id = order.Id - }); + }; } private async Task UpdateOrderAsync(UpdateOrderInput input, OrderEntity order) diff --git a/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs b/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs index db19e697..afe292ed 100644 --- a/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs +++ b/src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs @@ -100,7 +100,7 @@ public class TrackOrderService : ApplicationServiceBase x.IsCompleted }); - // todo:预警计算 + // 预警计算 if (order.IsCompleted) WarningCalc_OrderDeletedOrCompleted(order); diff --git a/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs b/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs index c98d7060..57cb6297 100644 --- a/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs +++ b/src/TrackCenter.Domain/Consts/GlobalConst.ES.cs @@ -12,11 +12,6 @@ public partial class GlobalConst /// public const string TrackPushRecordIndex = "sjzy_track_push_records"; - /// - /// 运输时效预警记录 es索引名称（模板按月） todo:可以删除 - /// - public const string TrackWarningRecordIndex = "sjzy_track_warning_records"; - /// /// 订单操作记录 es索引名称（模板按月） /// diff --git a/src/TrackCenter.Domain/Consts/GlobalConst.MQ.cs b/src/TrackCenter.Domain/Consts/GlobalConst.MQ.cs index fa9805e5..70c2f6d2 100644 --- a/src/TrackCenter.Domain/Consts/GlobalConst.MQ.cs +++ b/src/TrackCenter.Domain/Consts/GlobalConst.MQ.cs @@ -47,11 +47,6 @@ public partial class GlobalConst /// public const string TrackChangeAfterUpdateOrderSubscribe = "TrackCenter.TrackChangeAfterUpdateOrder"; - /// - /// 预警计算任务 - /// - public const string TrackWarningCalcSubscribe = "TrackCenter.TrackWarningCalc"; - /// /// 轨迹变更通知交换机 /// diff --git a/src/TrackCenter.Domain/Consts/GlobalConst.Redis.cs b/src/TrackCenter.Domain/Consts/GlobalConst.Redis.cs index 15d8a55e..c1f688e6 100644 --- a/src/TrackCenter.Domain/Consts/GlobalConst.Redis.cs +++ b/src/TrackCenter.Domain/Consts/GlobalConst.Redis.cs @@ -2,11 +2,6 @@ namespace TrackCenter.Domain.Consts; public partial class GlobalConst { - /// - /// 轨迹要推送的客户redis key前缀 - /// - public const string TrackPushKeyPrefix = "TrackPush"; - /// /// 轨迹推送内容配置redis key前缀 /// diff --git a/src/TrackCenter.Domain/Consts/GlobalConst.cs b/src/TrackCenter.Domain/Consts/GlobalConst.cs index 3c249cc4..c8e463d4 100644 --- a/src/TrackCenter.Domain/Consts/GlobalConst.cs +++ b/src/TrackCenter.Domain/Consts/GlobalConst.cs @@ -17,5 +17,7 @@ public partial class GlobalConst public const string CapNoRetryErrorCode = "599"; + public static readonly DateTime MinDateTime = new DateTime(1900, 1, 1, 0, 0, 0); + public static readonly DateTime MaxDateTime = new DateTime(2999, 1, 1, 0, 0, 0); }