From e0e376404eac8483072e2cf7afba776a46123da7 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Sun, 15 Sep 2024 17:13:48 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Dtos/Forewarning/WarningDto.cs | 13 ++ .../TrackChangeAfterUpdateOrderEvent.cs | 20 --- .../Events/TrackWarningCalcEvent.cs | 29 ++++ ...TrackChangeAfterUpdateOrderEventHandler.cs | 47 +---- .../EventHandlers/TrackUpsertEventHandler.cs | 113 ++++++++----- .../TrackWarningCalcEventHandler.cs | 75 ++++++++ .../ForewarningNotifyService.cs | 160 ++++++++++++------ .../ITrackDetailsEsRepository.cs | 2 + .../Consts/GlobalConst.MQ.cs | 2 +- .../Repositories/TrackDetailsEsRepository.cs | 51 ++++++ .../Repositories/WarningRecordEsRepository.cs | 2 +- 11 files changed, 348 insertions(+), 166 deletions(-) create mode 100644 src/SJZY.Track.Application.Contract/Events/TrackWarningCalcEvent.cs create mode 100644 src/SJZY.Track.Application/CAP/EventHandlers/TrackWarningCalcEventHandler.cs diff --git a/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningDto.cs b/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningDto.cs index 9ef9894b..b7249b2a 100644 --- a/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningDto.cs +++ b/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningDto.cs @@ -7,9 +7,22 @@ public class WarningDto /// public int TrackStatusId { get; set; } + /// + /// 子系统id + /// public int SubSystemId { get; set; } + /// + /// 客户id + /// public int CustomerId { get; set; } public List Orders { get; set; } = new(); + + + public override string ToString() + { + return $"子系统:{SubSystemId},客户:{CustomerId},订单:{Orders.Select(x => x.Id) + .JoinAsString(",")},发生轨迹状态{TrackStatusId}"; + } } diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs index 0fa89c4b..037c5a8c 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs @@ -2,16 +2,6 @@ namespace SJZY.Track.Application.Contract.Events; public class TrackChangeAfterUpdateOrderEvent { - /// - /// 子系统id - /// - public int SubSystemId { get; set; } - - /// - /// 客户id - /// - public int CustomId { get; set; } - /// /// 订单Id /// @@ -32,11 +22,6 @@ public class TrackChangeAfterUpdateOrderEvent /// public string TrackEnContent { get; set; } = null!; - /// - /// 轨迹状态Id - /// - public int TrackStatusId { get; set; } - /// /// 轨迹发生地 /// @@ -52,11 +37,6 @@ public class TrackChangeAfterUpdateOrderEvent /// public float TrackTimeZone { get; set; } - /// - /// 轨迹内容代码 - /// - public string? TrackCode { get; set; } - /// /// 时区缺失和发生地缺失标记，1表示时区，2表示发生地，3表示两种 /// diff --git a/src/SJZY.Track.Application.Contract/Events/TrackWarningCalcEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackWarningCalcEvent.cs new file mode 100644 index 00000000..3808ab2e --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Events/TrackWarningCalcEvent.cs @@ -0,0 +1,29 @@ +namespace SJZY.Track.Application.Contract.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; } = null!; + + /// + /// 轨迹发生时间 + /// + public DateTime TrackTime { get; set; } +} diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index 4b5d6cd9..153e9700 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs @@ -1,8 +1,6 @@ using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; -using SJZY.Track.Application.Contract.Dtos.Forewarning; using SJZY.Track.Application.Contract.Events; -using SJZY.Track.Application.Contract.Services; using SJZY.Track.Domain.Aggregates.OrderAggregate; using Volo.Abp.DependencyInjection; @@ -10,14 +8,11 @@ namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe { - private readonly IForewarningNotifyService _forewarningNotifyService; private readonly IOrderRepository _orderRepository; - public TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository, - IForewarningNotifyService forewarningNotifyService) + public TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository) { _orderRepository = orderRepository; - _forewarningNotifyService = forewarningNotifyService; } [CapSubscribe(GlobalConst.TrackChangeAfterUpdateOrderSubscribe)] @@ -27,7 +22,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa if (orders.Count == 0) { - LogHelper.Warn("没有找到订单信息，无法更新轨迹信息"); + LogHelper.Warn($"没有找到订单信息：{@event.OrderIds.JoinAsString(",")}，无法更新轨迹信息"); return; } @@ -74,9 +69,6 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa x.LatestTrackTimeZone, x.DeficiencyTag }); - - // 预警计算 - await _forewarningNotifyService.WarningCalculateAsync(BuiltWarningDto(orders, @event)); } private async Task> GetOrdersAsync(int[] orderIds) @@ -92,41 +84,6 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa LatestTrackPlace = order.LatestTrackPlace, LatestTrackTime = order.LatestTrackTime, LatestTrackTimeZone = order.LatestTrackTimeZone, - - // 因素 - CargoType = order.CargoType, - CreateTime = order.CreateTime, - ArrivalCountry = order.ArrivalCountry, - ArrivalProvince = order.ArrivalProvince, - ArrivalCity = order.ArrivalCity, - ArrivalPostCode = order.ArrivalPostCode, - IsTransit = order.IsTransit, - LeavePort = order.LeavePort, - ProductCode = order.ProductCode, }); } - - private WarningDto BuiltWarningDto(List orders, TrackChangeAfterUpdateOrderEvent @event) - { - return new WarningDto - { - TrackStatusId = @event.TrackStatusId, - SubSystemId = @event.SubSystemId, - CustomerId = @event.CustomId, - Orders = orders.Select(order => new WarningOrderDto - { - Id = order.Id, - TrackTime = @event.TrackTime, - CargoType = order.CargoType, - Month = order.CreateTime.Month, - IsTransit = order.IsTransit, - LeavePort = order.LeavePort, - ProductCode = order.ProductCode, - ArrivalCountry = order.ArrivalCountry, - ArrivalProvince = order.ArrivalProvince, - ArrivalCity = order.ArrivalCity, - ArrivalPostCode = order.ArrivalPostCode, - }).ToList() - }; - } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index b8e6225a..5f666525 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -4,6 +4,7 @@ using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; +using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; using OrderIdentifier = SJZY.Track.Application.Contract.Events.OrderIdentifier; @@ -135,12 +136,74 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return; } + var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent); + + if (canNextOrderIds.Count == 0) return; + + if (insertTracks.Any()) + { + // 批量add to es + await _trackDetailsEsRepository.AddManyAsync(insertTracks); + + // 添加轨迹触发订单拉取状态更新 + await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), + @event.TrackCode); + + // 添加轨迹触发预警计算（异步） + await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent + { + SubSystemId = @event.SubSystemId, + CustomId = @event.CustomId, + OrderIds = canNextOrderIds.ToArray(), + TrackStatusId = trackContent.TrackStatusId, + TrackTime = @event.TrackTime, + }); + } + + // 触发订单主表更新事件（异步） + await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, + new TrackChangeAfterUpdateOrderEvent + { + OrderIds = canNextOrderIds.ToArray(), + IsCompleted = trackContent.IsCompleted, + TrackContent = trackContent.TrackContent, + TrackEnContent = trackContent.TrackEnContent, + TrackPlace = @event.TrackPlace, + TrackTime = @event.TrackTime, + TrackTimeZone = @event.TrackTimeZone, + DeficiencyTagValue = @event.DeficiencyTagValue + }); + + // 触发轨迹推送事件（异步） + var trackPushEvent = new TrackPushEvent + { + OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, + SubSystemId = @event.SubSystemId, + CustomId = @event.CustomId, + OrderIds = canNextOrderIds.ToArray(), + TrackCode = @event.TrackCode, + SystemPushInfo = new SystemPushStruct + { + TrackStatus = trackContent.TrackStatus, + TrackContent = trackContent.TrackContent, + TrackPlace = @event.TrackPlace, + TrackTime = @event.TrackTime, + TrackTimeZone = @event.TrackTimeZone + } + }; + await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + } + + private async Task> UpsertTrackAsync(TrackUpsertEvent @event, List canNextOrderIds, + TrackContentVO trackContent) + { var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); // important：轨迹code相同，时间、发生地不同的，可能存在多条！！ var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); var insertTracks = new List(); + var updateTracks = new List(); foreach (var order in @event.OrderIdentifiers) { @@ -204,59 +267,17 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser; track.UpdateTime = DateTime.Now; - await _trackDetailsEsRepository.UpdateAsync(track); + updateTracks.Add(track); } canNextOrderIds.Add(order.OrderId); } - if (canNextOrderIds.Count == 0) return; - - if (insertTracks.Any()) + if (updateTracks.Any()) { - // 批量add to es - await _trackDetailsEsRepository.AddManyAsync(insertTracks); - - // add track触发订单拉取状态更新 - await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), - @event.TrackCode); + await _trackDetailsEsRepository.UpdateManyAsync(updateTracks); } - // 触发订单主表更新事件（异步） - await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, - new TrackChangeAfterUpdateOrderEvent - { - SubSystemId = @event.SubSystemId, - CustomId = @event.CustomId, - OrderIds = canNextOrderIds.ToArray(), - IsCompleted = trackContent.IsCompleted, - TrackContent = trackContent.TrackContent, - TrackEnContent = trackContent.TrackEnContent, - TrackStatusId = trackContent.TrackStatusId, - TrackPlace = @event.TrackPlace, - TrackTime = @event.TrackTime, - TrackTimeZone = @event.TrackTimeZone, - TrackCode = @event.TrackCode, - DeficiencyTagValue = @event.DeficiencyTagValue - }); - - // 触发轨迹推送事件（异步） - var trackPushEvent = new TrackPushEvent - { - OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, - SubSystemId = @event.SubSystemId, - CustomId = @event.CustomId, - OrderIds = canNextOrderIds.ToArray(), - TrackCode = @event.TrackCode, - SystemPushInfo = new SystemPushStruct - { - TrackStatus = trackContent.TrackStatus, - TrackContent = trackContent.TrackContent, - TrackPlace = @event.TrackPlace, - TrackTime = @event.TrackTime, - TrackTimeZone = @event.TrackTimeZone - } - }; - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + return insertTracks; } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackWarningCalcEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackWarningCalcEventHandler.cs new file mode 100644 index 00000000..acefee10 --- /dev/null +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackWarningCalcEventHandler.cs @@ -0,0 +1,75 @@ +using DotNetCore.CAP; +using SJZY.Expand.ABP.Core.Logger; +using SJZY.Track.Application.Contract.Dtos.Forewarning; +using SJZY.Track.Application.Contract.Events; +using SJZY.Track.Application.Contract.Services; +using SJZY.Track.Domain.Aggregates.OrderAggregate; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.CAP.EventHandlers; + +public class TrackWarningCalcEventHandler : ITransientDependency, ICapSubscribe +{ + private readonly IForewarningNotifyService _forewarningNotifyService; + private readonly IOrderRepository _orderRepository; + + public TrackWarningCalcEventHandler(IOrderRepository orderRepository, + IForewarningNotifyService forewarningNotifyService) + { + _orderRepository = orderRepository; + _forewarningNotifyService = forewarningNotifyService; + } + + [CapSubscribe(GlobalConst.TrackWarningCalcSubscribe)] + public async Task ConsumerAsync(TrackWarningCalcEvent @event) + { + var orders = await _orderRepository.GetListAsync(order => @event.OrderIds.Contains(order.Id), order => + new OrderEntity + { + Id = order.Id, + // 因素 + CargoType = order.CargoType, + CreateTime = order.CreateTime, + ArrivalCountry = order.ArrivalCountry, + ArrivalProvince = order.ArrivalProvince, + ArrivalCity = order.ArrivalCity, + ArrivalPostCode = order.ArrivalPostCode, + IsTransit = order.IsTransit, + LeavePort = order.LeavePort, + ProductCode = order.ProductCode, + }); + + if (orders.Count == 0) + { + LogHelper.Warn($"没有找到订单信息：{@event.OrderIds.JoinAsString(",")}，无法做轨迹预警计算"); + return; + } + + // 预警计算 + await _forewarningNotifyService.WarningCalculateAsync(BuiltWarningDto(orders, @event)); + } + + private WarningDto BuiltWarningDto(List orders, TrackWarningCalcEvent @event) + { + return new WarningDto + { + TrackStatusId = @event.TrackStatusId, + SubSystemId = @event.SubSystemId, + CustomerId = @event.CustomId, + Orders = orders.Select(order => new WarningOrderDto + { + Id = order.Id, + TrackTime = @event.TrackTime, + CargoType = order.CargoType, + Month = order.CreateTime.Month, + IsTransit = order.IsTransit, + LeavePort = order.LeavePort, + ProductCode = order.ProductCode, + ArrivalCountry = order.ArrivalCountry, + ArrivalProvince = order.ArrivalProvince, + ArrivalCity = order.ArrivalCity, + ArrivalPostCode = order.ArrivalPostCode, + }).ToList() + }; + } +} diff --git a/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs b/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs index 2859eda9..20764451 100644 --- a/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs +++ b/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs @@ -89,25 +89,28 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep } var redisKey = GetForewarningNotifySetKeyRedisKey(dto.SubSystemId); + long addCounter, delCounter = 0; var tran = RedisHelper.Client.Multi(); - // 左区间加入预警队列 - await tran.ZAddAsync(redisKey, adds.Select(x => + // 左区间加入预警队列，【只有第一次发生的才会被加入】 + addCounter = await tran.ZAddNxAsync(redisKey, adds.Select(x => { var entry = new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter); return new ZMember(entry.ToString(), x.ts); }).ToArray()); - // 右区间移出预警队列 + // 右区间移出预警队列，【只要发生了就移除】 var right = dels.Select(x => new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter)).ToList(); if (!right.IsNullOrEmpty()) { - await tran.ZRemAsync(redisKey, right.Select(x => x.ToString()).ToArray()); + delCounter = await tran.ZRemAsync(redisKey, right.Select(x => x.ToString()).ToArray()); } tran.Exec(); + LogHelper.Info($"本次共计产生{addCounter}条预警，移除{delCounter}条预警，触发缘由是"); + await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成 } @@ -127,21 +130,21 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep var ruleIds = entries.Select(x => x.RuleId).Distinct().ToList(); var notifyRules = await _warningQueries.GetWarningNotifyRulesAsync(ruleIds); - var userRules = new Dictionary>(); - foreach (var rule in notifyRules) - { - foreach (var userId in rule.NotifyUserIds) - { - if (!userRules.ContainsKey(userId)) - { - userRules[userId] = new List(); - } - - userRules[userId].Add(rule); - } - } - // 写入es预警记录 + var warningRecords = await WriteWarningRecordAsync(entries, notifyRules); + + // 发通知 + await SendCoreAsync(entries, notifyRules); + + // 移除已处理的预警 + await RedisHelper.Client.ZRemRangeByScoreAsync(redisKey, 0, timestamp); + + return warningRecords.Count; + } + + private async Task> WriteWarningRecordAsync(List entries, + List notifyRules) + { var warningRecords = new List(); foreach (var entry in entries) { @@ -163,9 +166,25 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep } await _warningRecordEsRepository.BulkInsertAsync(warningRecords); + return warningRecords; + } + private Task SendCoreAsync(List entries, List notifyRules) + { + var userRules = new Dictionary>(); + foreach (var rule in notifyRules) + { + foreach (var userId in rule.NotifyUserIds) + { + if (!userRules.ContainsKey(userId)) + { + userRules[userId] = new List(); + } + + userRules[userId].Add(rule); + } + } - // 发通知 foreach (var userRule in userRules) { var siteInternalContent = BuildNotifyContent(userRule.Value, entries, NotifyMode.SiteInternal); @@ -173,48 +192,62 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep if (!string.IsNullOrEmpty(siteInternalContent)) { - // todo：发送站内通知 - _ = _requestSendMessageService.Service.SendMessageByEmptyTemplateAsync( - new SendMessageByEmptyTemplateInput + _ = Task.Run(async () => + { + // todo：发送站内通知 + var res = await _requestSendMessageService.Service.SendMessageByEmptyTemplateAsync( + new SendMessageByEmptyTemplateInput + { + IsNeedToHandle = null, + BizId = null, + MsgSource = null, + SystemId = 0, + TemplateCode = null, + Channels = null, + MsgTitle = null, + MsgContent = null, + PcURl = null, + MobileURl = null, + FileUrl = null, + fileName = null, + receivers = null, + SiteLetterComponentStyle = (ComponentStyle) 0 + }); + + if (res.Code != GlobalConst.SuccessCode) { - IsNeedToHandle = null, - BizId = null, - MsgSource = null, - SystemId = 0, - TemplateCode = null, - Channels = null, - MsgTitle = null, - MsgContent = null, - PcURl = null, - MobileURl = null, - FileUrl = null, - fileName = null, - receivers = null, - SiteLetterComponentStyle = (ComponentStyle) 0 - }); + LogHelper.Error($"发送站内通知失败，用户：{userRule.Key}，内容：{siteInternalContent}"); + } + }); } if (!string.IsNullOrEmpty(feiShuContent)) { - // 发送飞书通知 - _ = _messageClient.SendMessageAsync(new SendMessageInput + _ = Task.Run(async () => { - AppId = null, - AppSecret = null, - ReceiveIdType = IdTypeEnums.user_id, - MsgType = MsgTypeEnums.文本, - ReceiveIds = new List + // 发送飞书通知 + var res = await _messageClient.SendMessageAsync(new SendMessageInput { - userRule.Key.ToString() - }, - Content = feiShuContent + AppId = null, + AppSecret = null, + ReceiveIdType = IdTypeEnums.user_id, + MsgType = MsgTypeEnums.文本, + ReceiveIds = new List + { + userRule.Key.ToString() + }, + Content = feiShuContent + }); + + if (res.Code != GlobalConst.SuccessCode) + { + LogHelper.Error($"发送飞书通知失败，用户：{userRule.Key}，内容：{feiShuContent}"); + } }); } } - await RedisHelper.Client.ZRemRangeByScoreAsync(redisKey, 0, timestamp); - - return warningRecords.Count; + return Task.CompletedTask; } private string GetForewarningNotifySetKeyRedisKey(int subSystemId) => @@ -259,6 +292,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep var res = new List(); foreach (var rule in rules) { + // 规则至少有一个因素，如果订单任一因素不符合则不命中 if (rule.IsTransit != null) { if (order.IsTransit != rule.IsTransit) continue; @@ -287,7 +321,8 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep if (!rule.Areas.IsNullOrEmpty()) { var inArea = false; - foreach (var area in rule.Areas!.Where(area => order.ArrivalCountry == area.Country)) + var areas = rule.Areas!.Where(area => order.ArrivalCountry == area.Country).ToList(); + foreach (var area in areas) { if (!area.LeftPostcode.IsNullOrEmpty() && !area.RightPostcode.IsNullOrEmpty()) { @@ -296,12 +331,31 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep string.CompareOrdinal(order.ArrivalPostCode, area.RightPostcode) > 0) continue; } - if (!area.City.IsNullOrEmpty() && order.ArrivalCity != area.City) continue; - if (!area.Province.IsNullOrEmpty() && order.ArrivalProvince != area.Province) continue; inArea = true; break; } + if (!inArea) + { + foreach (var area in areas) + { + if (!area.City.IsNullOrEmpty() && order.ArrivalCity != area.City) continue; + inArea = true; + break; + } + } + + if (!inArea) + { + foreach (var area in areas) + { + if (!area.Province.IsNullOrEmpty() && order.ArrivalProvince != area.Province) continue; + inArea = true; + break; + } + } + + if (!inArea) continue; } diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsEsRepository.cs index e7ba4c60..37e3d1d2 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsEsRepository.cs @@ -14,6 +14,8 @@ public interface ITrackDetailsEsRepository : IElasticsearchClientService indexes, int maxRetries = 3); + Task AddManyAsync(List index); Task AddAsync(TrackDetailsIndex track); diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 29103d0d..d517bfc8 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -50,5 +50,5 @@ public partial class GlobalConst /// /// 预警计算任务 /// - public const string TrackWarningCalc = "SJZY.Track.TrackWarningCalc"; + public const string TrackWarningCalcSubscribe = "SJZY.Track.TrackWarningCalc"; } diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsEsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsEsRepository.cs index 6cf163ff..62070eea 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsEsRepository.cs @@ -154,6 +154,57 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract indexes, int maxRetries = 3) + { + var retryCount = 0; + var updateSuccessful = false; + var track = indexes.First(); + + while (retryCount < maxRetries && !updateSuccessful) + { + var response = await Context.UpdateByQueryAsync(u => u + .Index(GlobalConst.TrackDetailsIndex) + .Query(q => q + .Ids(t => t.Values(indexes.Select(x => x.Id))) + ) + .Script(s => s + .Source(@" + ctx._source.TrackPlace = params.trackPlace; + ctx._source.TrackTime = params.trackTime; + ctx._source.TrackTimeZone = params.trackTimeZone; + ctx._source.UpdateUser = params.updateUser; + ctx._source.UpdateTime = params.updateTime; + ") + .Params(p => p + .Add("trackPlace", track.TrackPlace) + .Add("trackTime", track.TrackTime) + .Add("trackTimeZone", track.TrackTimeZone) + .Add("updateUser", track.UpdateUser) + .Add("updateTime", DateTime.Now) + ) + ) + ); + + if (response.IsValid) + { + updateSuccessful = true; + } + else if (response.ServerError?.Status == 409) // 409 表示版本冲突 + { + retryCount++; + if (retryCount >= maxRetries) + { + throw new InvalidOperationException( + $"轨迹更新冲突达到最大次数，track：{indexes.Count}条，原因：{response.DebugInformation}。"); + } + } + else + { + throw new InvalidOperationException($"轨迹更新失败，track：{indexes.Count}条，原因：{response.DebugInformation}。"); + } + } + } + public async Task AddManyAsync(List tracks) { // 按天写入es diff --git a/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs index 0bc3d029..a3ffd54e 100644 --- a/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs @@ -49,7 +49,7 @@ public class WarningRecordEsRepository : ElasticSearchContextAbstract(u => u - .Index($"{GlobalConst.TrackWarningRecordIndex}-*") + .Index(GlobalConst.TrackWarningRecordIndex) .Query(q => q.Term(f => f.RuleId, rule.ruleId) && q.Term(f => f.OrderId, rule.orderId)