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

perf:优化完整预警计算，优化流程减少N+1

Browse Source
This commit is contained in:
hardstylewyl 2025-04-10 14:35:30 +08:00
parent d5b2bfd5ec
commit 8c7a1c5fc1
8 changed files with 143 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,6 +7,7 @@ using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals;
using SJZY.Track.Domain.Dtos.TrackDetails;
using SJZY.Track.Domain.Dtos.Warning;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Extensions;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Services.Warning.Handlers;
@ -35,44 +36,34 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient
// 1.只处理未删除的订单
var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet();
var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id));
if (orders.Count == 0) return; //不存在符合条件的订单，直接提交信号丢弃
if (orders.Count == 0) return;
// 2.查出订单指定状态的轨迹（tips:为了指定状态的轨迹是真的被清空了）
// 2.查出订单轨迹
var orderIds = orders.Select(x => x.Id).ToHashSet();
var orderIdTrackStatusIdsDict = deletedSignals
.Where(x => orderIds.Contains(x.OrderId)) // 过滤掉不必要的信号
.GroupBy(x => x.OrderId)
.ToDictionary(
x => x.Key,
y => y.Select(x => x.DeletedTrack.TrackStatusId)
.ToHashSet()
);
var orderTracksDict = await _trackDetailsEsRepository.GetOrderTracksDictAsync(new GetOrderTracksDictVO
{
OrderIds = orderIdSet
OrderIds = orderIds
});
// 构建k:orderId,v:[k:trackStatusId,v:true]
var orderIdTrackStatusGroupsDict = orderTracksDict
.ToDictionary(x => x.Key,
y => y.Value
.GroupBy(x => x.TrackStatusId)
.ToDictionary(a => a.Key, _ => true)
);
// 构建订单id和轨迹状态ids字典
var orderIdTrackStatusIdsDict = orderTracksDict.ToDictionary(
x => x.Key,
y => y.Value.Select(x => x.TrackStatusId).ToHashSet()
);
// 3.查出预警明细(待处理)
var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO
{
OrderIds = orderIdSet,
OrderIds = orderIds,
});
if (warningDetails.Count == 0) return;
// 查询规则项目
// 4.查询规则项目
var ruleItemIds = warningDetails.Select(x => x.CompleteWarningRuleItemId).ToHashSet();
var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds);
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
// 4.处理明细
// 5.处理明细
var closeDetailIds = new HashSet<string>();
var updateDetails = new List<CompleteWarningDetailsIndex>();
foreach (var detail in warningDetails)
@ -85,16 +76,15 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient
continue;
}
var trackStatusIdDict = orderIdTrackStatusGroupsDict.GetValueOrDefault(detail.OrderId);
var trackStatusIds = orderIdTrackStatusIdsDict.GetValueOrDefault(detail.OrderId, Empty<int>.Set);
// 订单无任何轨迹,或者不存在检查点状态轨迹
if (trackStatusIdDict == null || !trackStatusIdDict.ContainsKey(detail.CheckPointTrackStatusId))
if (!trackStatusIds.Contains(detail.CheckPointTrackStatusId))
{
closeDetailIds.Add(detail.Id);
continue;
}
// 查看缺失轨迹状态表达式是否有变化,如果有则更新
var trackStatusIds = trackStatusIdDict.Keys.ToHashSet();
var expression = ruleItem.BuildMissingTrackStatusIdExpression(trackStatusIds);
if (!string.Equals(detail.MissingTrackStatusIds, expression))
{
@ -102,21 +92,23 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient
updateDetails.Add(detail);
}
// todo:需要处理要求轨迹完整的情况？
// tips:需要处理要求轨迹完整的情况？
}
// 6.批量更新轨迹
if (updateDetails.Count > 0)
await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails);
// 没有需要关闭的预警明细
if (closeDetailIds.Count == 0) return;
// 5.预警信息状态修改为【已关闭】
// 7.预警信息状态修改为【已关闭】
await _warningDetailsEsRepository.BulkCloseCompleteWarningDetailsAsync(new BulkCloseCompleteWarningDetailsVO
{
DetailsIds = closeDetailIds
});
// 6.关闭预警通知任务
// 8.关闭预警通知任务
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
!x.IsClosed
&& closeDetailIds.Contains(x.WarningDetailsId)
@ -133,92 +125,125 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient
// 1.只处理未删除的订单
var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet();
var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id));
if (orders.Count == 0) return; //不存在符合条件的订单，直接提交信号丢弃
if (orders.Count == 0) return;
var orderIds = orders.Select(x => x.Id).ToHashSet();
// 2.查询订单每个轨迹状态的最早轨迹
var orderTracksDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
// 3.以单的维度处理
// todo:优化流程减少N+1
// 2.查询订单每个轨迹状态的最早轨迹
var orderTrackStatusEarliestDictDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
// 3.查规则主体
var conditionDict = orders.GroupBy(x => (x.SubSystemId, x.CustomId)).ToDictionary(x => x.Key, y => y.ToList());
var conditions = conditionDict.Keys.ToHashSet();
var ruleDict = await _warningRuleRepository.GetWarningRuleForEnabledAsync(conditions, WarningRuleType.完整预警);
if (ruleDict.Count == 0) return;
// 4.查规则项目
var ruleIds = ruleDict.Keys.ToHashSet();
var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(x => ruleIds.Contains(x.WarningId));
if (ruleItems.Count == 0) return;
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
// 5.构建订单id 命中规则，完成规则 字典
var orderIdHitRuleItemDict = new Dictionary<int, List<CompleteWarningRuleItemEntity>>();
var orderIdDoneRuleItemIdsDict = new Dictionary<int, HashSet<int>>();
var orderIdRuleItemsDict = new Dictionary<int, HashSet<int>>();
foreach (var group in ruleItems.GroupBy(x => x.WarningId))
{
var rule = ruleDict[group.Key];
var hitRuleOrders = conditionDict[(rule.SubSystemId, rule.CustomId)]; //tips:一定存在
foreach (var order in hitRuleOrders)
{
var trackStatusIdTrackDict = orderTrackStatusEarliestDictDict.GetValueOrDefault(order.Id);
if (trackStatusIdTrackDict == null) continue;
var trackStatusIds = trackStatusIdTrackDict.Keys.ToHashSet();
foreach (var ruleItem in group)
{
if (ruleItem.CanHitRule(order.OrderType, trackStatusIds))
{
orderIdRuleItemsDict.GetOrAdd(order.Id, _ => new HashSet<int>()).Add(ruleItem.Id);
orderIdHitRuleItemDict.GetOrAdd(order.Id, _ => new List<CompleteWarningRuleItemEntity>()).Add(ruleItem);
continue;
}
if (ruleItem.IsDone(order.OrderType, trackStatusIds))
{
orderIdRuleItemsDict.GetOrAdd(order.Id, _ => new HashSet<int>()).Add(ruleItem.Id);
orderIdDoneRuleItemIdsDict.GetOrAdd(order.Id, _ => new HashSet<int>()).Add(ruleItem.Id);
}
}
}
}
// 6.查预警信息
var allWarningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO
{
OrderIdRuleItemIdsDict = orderIdRuleItemsDict,
});
var orderIdWarningDetailDict = allWarningDetails
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.ToList());
// 7.以单的维度处理
var addDetails = new List<CompleteWarningDetailsIndex>();
var updateDetails = new List<CompleteWarningDetailsIndex>();
var closeDetailsIds = new HashSet<string>();
foreach (var order in orders)
{
var trackStatusIdTrackDict = orderTracksDict.GetValueOrDefault(order.Id);
if (trackStatusIdTrackDict == null) continue;
// 订单轨迹状态
var orderTrackEarliestTracksDict = orderTrackStatusEarliestDictDict.GetValueOrDefault(order.Id);
if (orderTrackEarliestTracksDict == null) continue;
var orderTrackStatusIds = orderTrackEarliestTracksDict.Keys.ToHashSet();
// 查询规则主体
var rule = await _warningRuleRepository.GetWarningRuleForEnabledAsync(order.SubSystemId, order.CustomId, WarningRuleType.完整预警);
if (rule == null) continue;
// 订单命中的预警规则项目
var hitRuleItems = orderIdHitRuleItemDict.GetValueOrDefault(order.Id, Empty<CompleteWarningRuleItemEntity>.List);
var hitRuleItemIds = hitRuleItems.Select(x => x.Id).ToHashSet();
// 查询规则项目
var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(rule.Id);
if (ruleItems.Count == 0) continue;
// 匹配规则项目（要生成预警的，已完成的）
var trackStatusIds = trackStatusIdTrackDict.Keys.ToHashSet();
var hitRuleItems = new List<CompleteWarningRuleItemEntity>();
var doneRuleItemIds = new HashSet<int>();
foreach (var ruleItem in ruleItems)
// 处理存在的预警明细
var orderWarningDetails = orderIdWarningDetailDict.GetValueOrDefault(order.Id, Empty<CompleteWarningDetailsIndex>.List);
if (orderWarningDetails.Count > 0)
{
if (ruleItem.CanHitRule(order.OrderType, trackStatusIds))
hitRuleItems.Add(ruleItem);
if (ruleItem.IsDone(order.OrderType, trackStatusIds))
doneRuleItemIds.Add(ruleItem.Id);
var doneRuleItemIds = orderIdDoneRuleItemIdsDict.GetValueOrDefault(order.Id, Empty<int>.Set);
var doneRuleItemDetails = orderWarningDetails.Where(x => doneRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList();
if (doneRuleItemDetails.Count > 0)
{
doneRuleItemDetails.ForEach(detail =>
{
detail.Status = CompleteWarningStatus.已处理.GetHashCode();
detail.MissingTrackStatusIds = null; // 清空缺失轨迹
detail.ProcessedTime = DateTime.Now;
});
updateDetails.AddRange(doneRuleItemDetails);
}
// 处理已存在的预警明细（命中规则的）
var existsWarningDetails = orderWarningDetails.Where(x => hitRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList();
var needUpdateDetails = HandleExistsDetails(orderTrackStatusIds, existsWarningDetails, ruleItemDict);
updateDetails.AddRange(needUpdateDetails);
// 排除已经生成预警信息的规则
hitRuleItemIds.ExceptWith(existsWarningDetails.Select(x => x.CompleteWarningRuleItemId));
}
if (hitRuleItems.Count == 0 && doneRuleItemIds.Count == 0) return;
var hitRuleItemDict = hitRuleItems.ToDictionary(x => x.Id);
// 查出规则已经生成预警信息
var hitRuleItemIds = hitRuleItemDict.Keys.ToHashSet();
var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO
{
OrderIds = new HashSet<int>(1)
{
order.Id
},
RuleItemIds = new HashSet<int>()
.Union(hitRuleItemIds)
.Union(doneRuleItemIds)
.ToHashSet(),
});
// 已完结的预警规则，稍后修改明细状态为【已处理】
closeDetailsIds.UnionWith(warningDetails.Where(x => doneRuleItemIds.Contains(x.CompleteWarningRuleItemId)).Select(x => x.Id));
// 处理已存在的预警明细（命中规则的）
var existsWarningDetails = warningDetails.Where(x => hitRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList();
var needUpdateDetails = HandleExistsDetails(trackStatusIds, existsWarningDetails, hitRuleItemDict);
updateDetails.AddRange(needUpdateDetails);
// 排除已经生成预警信息的规则
hitRuleItemIds.ExceptWith(existsWarningDetails.Select(x => x.CompleteWarningRuleItemId));
if (hitRuleItemIds.Count == 0) continue;
// 生成预警明细
if (hitRuleItemIds.Count == 0) continue;
var rule = ruleDict[ruleItems.First().WarningId]; //tips：一定存在
var needAddDetails = hitRuleItems
.Where(x => hitRuleItemIds.Contains(x.Id))
.Select(x => BuildCompleteWarningDetails(order, trackStatusIds, x, rule.Version));
.Select(x => BuildCompleteWarningDetails(order, orderTrackStatusIds, x, rule.Version));
addDetails.AddRange(needAddDetails);
}
// 4.预警由于不缺失轨迹变为失效，修改明细状态为【已处理】，并关闭相关预警通知任务
if (closeDetailsIds.Count > 0)
await CloseWarningDetailLinksAsync(closeDetailsIds, CompleteWarningStatus.已处理);
// 更新预警明细
// 8.批量更新预警明细
if (updateDetails.Count > 0)
await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails);
if (addDetails.Count == 0) return;
// 5.批量新增预警明细
// 9.批量新增预警明细
await _warningDetailsEsRepository.BulkInsertCompleteWarningDetailsAsync(addDetails);
// 6.批量新增预警通知任务
// 10.批量新增预警通知任务
var notifyTasks = BuildWarningNotifyTask(addDetails);
await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(notifyTasks);
}

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

@ -57,7 +57,6 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
{
TrackEsIds = records.Select(x => x.TrackEsId).ToHashSet()
});
if (warningDetails.Count == 0) return;
var detailIds = warningDetails.Select(x => x.Id).ToHashSet();
@ -237,32 +236,13 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
if (addNotifyTasks.Count > 0)
await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks);
// 5个订单 每个订单推5个轨迹，失败（整体），5条历史成功推送记录（EsId+版本号），
// 第一次，生成20预警明细，生成一个回传报文并关联20预警明细，生成重试任务，重试任务重试25条推送
// 重试阶段的重试结果维护同维护20条预警明细，如果这一次失败超过了最大重试次数，并且命中了kpi考核，则会生成20预警通知任务
// 可能最后会生成20条人工介入的明细，
// 如果是人工手动推送则还是有可能会失败，回传失败又会走到这里，不要在进行重试，但是要维护回传报文
// todo:人工介入直到推送成功才可关闭吗？
//新增/修改轨迹 -> 系统推送 -> 推送回传 -> 预警计算
//手动推送 -> 推送回传 -> 预警计算
//手动推送 -> 推送回传 -> 预警计算
//重试推送 -> 推送回传 -> 预警计算
// todo: 预警页面的推送，根据明细id找到明细状态，确保是人工介入才能推送（新接口）
// 历史没有1成功的记录
// 1失败，回调，查询1明细包含【已关闭】为0，创建1明细，创建第一次重试任务T1
// 修改为2，导致1明细关闭 因为是同一个EsId，
// T1第一次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，什么也不做
// 。。。假设中途没有变更
// T1第二次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，修改，创建第三次重试任务T1
// T2第一次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，修改，创建第二次重试任务T2
}
// 从推送内容提取轨迹id+版本号
private (Dictionary<string, Dictionary<string, long>> OrderNoTrackIdVersionDict, Dictionary<string, long> AllTrackIdVersionDict)
private static (Dictionary<string, Dictionary<string, long>> OrderNoTrackIdVersionDict, Dictionary<string, long> AllTrackIdVersionDict)
ExtractOrderNoTrackIdVersionDict(List<TrackPushInput> pushInputs, List<TrackUpdateInput> updateInputs)
{
var inputs = (pushInputs.Count > 0
@ -494,7 +474,6 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
{
if (addRuleItems.Count > 0)
await OnlyAddHandleAsync(rule, addRuleItems);
return;
}

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

@ -226,7 +226,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
orders = orders.Where(x => earliestTrackStatusTrackDict.ContainsKey(x.Id)).ToList();
if (orders.Count == 0) return;
// 3.查轨迹主体
// 3.查规则主体
var conditions = orders.Select(x => (x.SubSystemId, x.CustomId)).ToHashSet();
var ruleDict = await _warningRuleRepository.GetWarningRuleForEnabledAsync(conditions, WarningRuleType.运输时效预警);
if (ruleDict.Count == 0) return;

3
src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs
View File

@ -100,6 +100,9 @@ public interface IWarningRuleRepository : IBasicRepository<WarningRuleEntity>
// 按规则项目id获取预警规则
Task<List<CompleteWarningRuleItemEntity>> GetCompleteRuleItemsAsync(HashSet<int> ruleItemIds);
// 获取完整预警规则项目
Task<List<CompleteWarningRuleItemEntity>> GetCompleteRuleItemsAsync(Expression<Func<CompleteWarningRuleItemEntity, bool>> whereExpression);
// 批量更新预警规则规则项目
Task<int> BulkUpdateCompleteRuleItemsAsync(List<CompleteWarningRuleItemEntity> entities);

2
src/SJZY.Track.Domain/Consts/GlobalConst.ES.cs
View File

@ -22,7 +22,7 @@ public partial class GlobalConst
/// </summary>
public const string OrderOperationRecordIndex = "sjzy_track_order_operation_records";
// todo：以下为新增需要配置索引模板(uat/prod)
// todo：以下为新增需要配置索引模板(prod) 生产环境还需要建立索引！！
/// <summary>
/// 运输时效预警明细 es索引名称（模板按天）
/// </summary>

4
src/SJZY.Track.Domain/Dtos/Warning/GetCompleteWarningDetailsVO.cs
View File

@ -20,9 +20,13 @@ public class GetCompleteWarningDetailsVO
// 订单id 与 检查点轨迹状态ids 同时满足
public Dictionary<int, HashSet<int>>? OrderIdCheckTracksStatusIdsDict { get; set; }
// 订单id 与 规则条目ids 同时满足
public Dictionary<int, HashSet<int>>? OrderIdRuleItemIdsDict { get; set; }
public bool HasDetailsIds => DetailsIds?.Count > 0;
public bool HasRuleItemIds => RuleItemIds?.Count > 0;
public bool HasWarningIds => WarningIds?.Count > 0;
public bool HasOrderIds => OrderIds?.Count > 0;
public bool HasOrderIdCheckTracksStatusIdsDict => OrderIdCheckTracksStatusIdsDict?.Count > 0;
public bool HasOrderIdRuleItemIdsDict => OrderIdRuleItemIdsDict?.Count > 0;
}

13
src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs
View File

@ -757,6 +757,19 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien
isShouldQuery = true;
}
else if (input.HasOrderIdRuleItemIdsDict)
{
foreach (var (orderId, ruleItemIds) in input.OrderIdRuleItemIdsDict!)
{
musts.Add(c => c.Bool(b => b.Must
(
mu => mu.Term(f => f.OrderId, orderId),
mu => mu.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(ruleItemIds))
)));
}
isShouldQuery = true;
}
else
{
musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(input.RuleItemIds)));

7
src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs
View File

@ -277,6 +277,13 @@ public class WarningRuleRepository : BasicRepository<WarningRuleEntity>, IWarnin
.ToListAsync();
}
public Task<List<CompleteWarningRuleItemEntity>> GetCompleteRuleItemsAsync(Expression<Func<CompleteWarningRuleItemEntity, bool>> whereExpression)
{
return Context.Queryable<CompleteWarningRuleItemEntity>()
.Where(whereExpression)
.ToListAsync();
}
public async Task<int> BulkUpdateCompleteRuleItemsAsync(List<CompleteWarningRuleItemEntity> entities)
{
return await Context.Updateable(entities)