perf:优化轨迹删除增加广播延时发送,并且修正为不阻塞主流程

This commit is contained in:
hardstylewyl 2025-01-16 11:55:46 +08:00
parent e0be935489
commit 40c663634e

View File

@ -47,7 +47,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
//当轨迹删除,根据轨迹标记/状态上IsToMq选择进行广播 //当轨迹删除,根据轨迹标记/状态上IsToMq选择进行广播
if (trackList.Count != 0) if (trackList.Count != 0)
await PublicTrackTopicAsync(@event, trackList); _ = PublicTrackTopicAsync(@event, trackList);
// 根据订单关联关系异步向下刷(异步) // 根据订单关联关系异步向下刷(异步)
await DownByOrderLink(@event); await DownByOrderLink(@event);
@ -121,89 +121,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
return true; return true;
} }
private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List<TrackDetailsIndex> 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;
foreach (var dataItem in trackGroup)
{
var item = dataItem.First();
var content = contentList.First(m => m.TrackContentCode == item.TrackCode);
// _broadcastingService.TrackChangePublicTopic(track,
// dataItem.Select(m => new OrderIdentifier
// {
// OrderId = m.OrderId,
// OrderNo = m.OrderNo,
// OrderType = m.OrderType
// }).ToArray(),
// item.TrackCode,
// item.TrackStatusCode,
// content.SubSystemCode
// );
//将每个订单的每个轨迹的删除变更发布
foreach (var track in dataItem)
{
_broadcastingService.TrackChangePublicTopic(new TrackChangeEto
{
EstimatedDeliveryDate = track.EstimatedDeliveryDate,
TrackPlacePostcode = track.TrackPlacePostcode,
TrackPlaceNation = track.TrackPlaceNation,
TrackPlaceNationId = track.TrackPlaceNationId,
TrackPlaceProvince = track.TrackPlaceProvince,
TrackPlaceProvinceId = track.TrackPlaceProvinceId,
TrackPlaceCity = track.TrackPlaceCity,
TrackPlaceCityId = track.TrackPlaceCityId,
Files = track.Files?.Select(x => new FileItemDto()
{
FileName = x.FileName,
OssKey = x.OssKey
}).ToList(),
Remarks = track.Remarks,
Source = track.Source,
IsVisibility = track.IsVisibility,
TrackTimeZone = track.TrackTimeZone,
TrackTime = track.TrackTime,
TrackPlace = track.TrackPlace,
TrackStatusCode = track.TrackStatusCode,
EnStatus = track.TrackEnContent,
Content = track.TrackContent,
TrackCode = track.TrackCode,
OperationMan = @event.OperationMan,
OrderIdentifiers = new[]
{
new OrderIdentifier()
{
OrderId = track.OrderId,
OrderNo = track.OrderNo,
OrderType = track.OrderType
}
},
SubSystemId = @event.SubSystemId,
Extra = track.Extra
}, content.SubSystemCode);
}
}
}
private async Task<List<TrackDetailsIndex>> DeleteTrackAsync(TrackDeleteEvent @event) private async Task<List<TrackDetailsIndex>> DeleteTrackAsync(TrackDeleteEvent @event)
{ {
List<int> orderIds; List<int> orderIds;
@ -264,7 +181,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
throw new BusinessException(message: $"当前轨迹不存在扩展信息,无法利用扩展字段:{@event.CompareFields}"); throw new BusinessException(message: $"当前轨迹不存在扩展信息,无法利用扩展字段:{@event.CompareFields}");
} }
var extraDict = GetExtraDictForTextJson(x.Extra); var extraDict = TextJsonHelper.DeserializeDictionary(x.Extra);
if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
{ {
throw new BusinessException(message: $"请检查拓展字段,为什么找不到比较字段:{@event.CompareFields}"); throw new BusinessException(message: $"请检查拓展字段,为什么找不到比较字段:{@event.CompareFields}");
@ -361,68 +278,88 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
return deleteTracks; return deleteTracks;
} }
private Dictionary<string, object?>? GetExtraDictForTextJson(string? extraStr) private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List<TrackDetailsIndex> trackList)
{ {
return string.IsNullOrWhiteSpace(extraStr) var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray();
? null var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray();
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(extraStr);
}
//json Element和object的比较 //从轨迹内容中查询出标记是否需要mq广播
private bool EqJsonElementObject2(object? a, object? b) var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes);
{ var toMqDict = contentList
if (a == null && b == null) .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq);
{
return true;
}
if (a is JsonElement aJsonElement) //选出需要广播的轨迹,并按code和订单号分组
{ var trackGroup = trackList
switch (aJsonElement.ValueKind) .Where(x => toMqDict[x.SubSystemId + x.TrackCode])
.GroupBy(m => new
{ {
case JsonValueKind.Undefined: m.TrackCode,
return false; m.OrderNo
case JsonValueKind.Object: })
return false; .ToList();
case JsonValueKind.Array: if (trackGroup.Count == 0) return;
return false;
case JsonValueKind.String: //增加延时确保写盘成功
return string.Equals(aJsonElement.GetString(), b as string); await Task.Delay(1000);
case JsonValueKind.Number: foreach (var dataItem in trackGroup)
{
var item = dataItem.First();
var content = contentList.First(m => m.TrackContentCode == item.TrackCode);
// _broadcastingService.TrackChangePublicTopic(track,
// dataItem.Select(m => new OrderIdentifier
// {
// OrderId = m.OrderId,
// OrderNo = m.OrderNo,
// OrderType = m.OrderType
// }).ToArray(),
// item.TrackCode,
// item.TrackStatusCode,
// content.SubSystemCode
// );
//将每个订单的每个轨迹的删除变更发布
foreach (var track in dataItem)
{
_broadcastingService.TrackChangePublicTopic(new TrackChangeEto
{
EstimatedDeliveryDate = track.EstimatedDeliveryDate,
TrackPlacePostcode = track.TrackPlacePostcode,
TrackPlaceNation = track.TrackPlaceNation,
TrackPlaceNationId = track.TrackPlaceNationId,
TrackPlaceProvince = track.TrackPlaceProvince,
TrackPlaceProvinceId = track.TrackPlaceProvinceId,
TrackPlaceCity = track.TrackPlaceCity,
TrackPlaceCityId = track.TrackPlaceCityId,
Files = track.Files?.Select(x => new FileItemDto()
{ {
if (b == null) return false; FileName = x.FileName,
OssKey = x.OssKey
if (b is int bInt) }).ToList(),
{ Remarks = track.Remarks,
var aInt = aJsonElement.GetInt32(); Source = track.Source,
return aInt == bInt; IsVisibility = track.IsVisibility,
} TrackTimeZone = track.TrackTimeZone,
TrackTime = track.TrackTime,
if (b is long bLong) TrackPlace = track.TrackPlace,
{ TrackStatusCode = track.TrackStatusCode,
var aLong = aJsonElement.GetInt64(); EnStatus = track.TrackEnContent,
return aLong == bLong; Content = track.TrackContent,
} TrackCode = track.TrackCode,
OperationMan = @event.OperationMan,
return false; OrderIdentifiers = new[]
}
case JsonValueKind.True:
case JsonValueKind.False:
{ {
if (b is bool bl) new OrderIdentifier()
{ {
return aJsonElement.GetBoolean() == bl; OrderId = track.OrderId,
OrderNo = track.OrderNo,
OrderType = track.OrderType
} }
},
return false; SubSystemId = @event.SubSystemId,
} Extra = track.Extra
case JsonValueKind.Null: }, content.SubSystemCode);
{
return false;
}
} }
} }
return false;
} }
} }