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

1

Browse Source
This commit is contained in:
xiaolipro 2025-07-14 18:20:21 +08:00
parent e1b2eef5f2
commit c66fe8b7dc
21 changed files with 808 additions and 624 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/TrackCenter.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs
View File

@ -96,7 +96,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
} }
); );
// todo：预警计算 // 引发预警计算
if (warningCalcOrders.Count > 0) if (warningCalcOrders.Count > 0)
WarningCalc_OrderDeletedOrCompleted(warningCalcOrders); WarningCalc_OrderDeletedOrCompleted(warningCalcOrders);
} }

452
src/TrackCenter.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -47,282 +47,177 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
} }
[CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)]
public async Task ConsumerAsync(TrackDeleteEvent @event) public async Task ConsumerAsync(TrackDeleteEvent @event)
{
if (!CanDelete(@event)) return;
var trackList = await DeleteTrackAsync(@event);
//当轨迹删除，根据轨迹标记/状态上IsToMq选择进行广播
if (trackList.Count > 0)
{ {
_ = PublicTrackTopicAsync(@event, trackList); if (!CanDelete(@event)) return;
//引发预警计算
WarningCalc_DeletedTrack(trackList); // 处理删除轨迹
await HandleDeleteTrackAsync(@event);
// 根据订单关联关系异步向下刷（异步）
await DownByOrderLink(@event);
} }
// 根据订单关联关系异步向下刷（异步） private async Task DownByOrderLink(TrackDeleteEvent @event)
await DownByOrderLink(@event);
}
private async Task DownByOrderLink(TrackDeleteEvent @event)
{
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds);
foreach (var orderId in @event.OrderIds)
{ {
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(); foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.SubSystemId))
nextEvent.Visited.AddRange(@event.OrderIds);
nextEvent.TrackIdentifiers = @event.TrackIdentifiers;
nextEvent.SubSystemId = group.Key;
// 不同子系统需要做内容映射
if (@event.SubSystemId != nextEvent.SubSystemId)
{ {
var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( var nextEvent = @event.DeepCopy();
@event.SubSystemId, nextEvent.Visited.AddRange(@event.OrderIds);
nextEvent.SubSystemId, nextEvent.TrackIdentifiers = @event.TrackIdentifiers;
@event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() nextEvent.SubSystemId = group.Key;
);
if (codeMaps.Count == 0) // 不同子系统需要做内容映射
if (@event.SubSystemId != nextEvent.SubSystemId)
{ {
LogHelper.Warn($"订单跨子系统，未配置内容映射，请联系管理员：{@event}", true); var codeMaps = await _provideQueries.GetTrackCodeMapsAsync(
continue; @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 subOrderIds in group.Select(x => x.OrderId).Chunk(GlobalConst.ChunkSize))
foreach (var trackIdentity in nextEvent.TrackIdentifiers)
{ {
var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); var subEvent = nextEvent.DeepCopy();
if (match == null) continue; subEvent.OrderIds = subOrderIds;
trackIdentity.TrackCode = match.TrackContentCode; 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) private static bool CanDelete(TrackDeleteEvent @event)
{ {
if (@event.OrderIds.IsEmpty()) 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; 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 HandleDeleteTrackAsync(TrackDeleteEvent @event)
}
private async Task<List<TrackDetailsIndex>> DeleteTrackAsync(TrackDeleteEvent @event)
{
List<int> orderIds;
List<TrackDetailsIndex> tracks;
List<TrackDetailsIndex> deleteTracks;
// 优先根据EsId查询轨迹
if (@event.EsIds.Length > 0) //根据明确的esid进行删除
{ {
deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); HashSet<int> orderIds;
orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); List<TrackDetailsIndex> oldTracks;
@event.OrderIds = orderIds.ToArray(); List<TrackDetailsIndex> deleteTracks;
@event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto // 优先根据EsId查询轨迹
if (@event.EsIds.Length > 0)
{ {
TrackCode = x.TrackCode, deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds);
TrackTime = x.TrackTime
}).ToList();
// 外部api获取的轨迹不允许删除 // 外部api获取的轨迹不允许删除
if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return;
return deleteTracks;
//查出订单下的所有轨迹，以供后面回退使用 // 从被删除轨迹寻找订单id，稍后会更新订单主表
tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet();
// 删除轨迹批量 // 清理已经删除过的EsId,并设置用于下刷删除的其他信息
await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); @event.EsIds = Array.Empty<string>();
@event.OrderIds = orderIds.ToArray();
@event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto
{
TrackCode = x.TrackCode,
TrackTime = x.TrackTime
}).ToList();
//清理已经删除过的esid // 查出订单下的所有轨迹，以供后面回退使用
@event.EsIds = Array.Empty<string>(); oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds);
}
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();
} }
//否则则使用比较字段,使用扩展字段中的内容进行比较
else else
{ {
var codes = @event.TrackIdentifiers.Select(x => x.TrackCode).ToImmutableHashSet(); // 根据订单查询出所有的轨迹，查所有是因为主单上需要进行回退更新
deleteTracks = tracks oldTracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds);
.Where(x =>
// 构建轨迹标识符
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)) // 从扩展信息提取比较值和传入值比较匹配
{ var extraDict = TextJsonHelper.DeserializeToDictionary(y.Extra);
throw new BusinessException(message: $"当前轨迹不存在扩展信息，无法利用扩展字段：{@event.CompareFields}"); if (extraDict!.TryGetValue(@event.CompareFields!, out var compareValueJsonElement))
} return TextJsonHelper.EqJsonElement(compareValueJsonElement, @event.CompareValue);
var extraDict = TextJsonHelper.DeserializeToDictionary(x.Extra); return false;
if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
{
throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}");
}
return TextJsonHelper.EqJsonElement(compareValue, @event.CompareValue);
}) })
.ToList(); .ToList();
//从被删除轨迹寻找订单id，稍后会更新订单主表
orderIds = deleteTracks.Select(m => m.OrderId).ToHashSet();
} }
if (deleteTracks.Count == 0) return;
if (deleteTracks.Count == 0) return deleteTracks;
// 删除轨迹批量 // 删除轨迹批量
await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); 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);
} }
//排除掉已经删除的轨迹 private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List<TrackDetailsIndex> delTracks)
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)
{ {
//找到最新的轨迹数据 var trackContentIds = delTracks.Select(x => x.TrackContentId).ToHashSet();
var track = tracks?.OrderByDescending(x => x.ZeroTrackTime) var trackContents = await _provideQueries.GetTrackContentAsync(trackContentIds);
.FirstOrDefault(x => x.OrderId == order.Id); var contentDict = trackContents.ToDictionary(x => x.Id);
// 删除全部轨迹导致的重置行为 var needPublicTracks = delTracks
if (track == null) // 轨迹内容已经不存在了，不需要广播
.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; var eto = new TrackChangeEto
// 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<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;
//增加延时确保写盘成功
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
{ {
TrackEsId = track.Id,
Option = 2,
IsSuccess = true,
FailedReason = null,
EstimatedDeliveryDate = track.EstimatedDeliveryDate, EstimatedDeliveryDate = track.EstimatedDeliveryDate,
DestinationTimeZone = track.DestinationTimeZone, DestinationTimeZone = track.DestinationTimeZone,
TrackPlacePostcode = track.TrackPlacePostcode, TrackPlacePostcode = track.TrackPlacePostcode,
@ -332,7 +227,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
TrackPlaceProvinceId = track.TrackPlaceProvinceId, TrackPlaceProvinceId = track.TrackPlaceProvinceId,
TrackPlaceCity = track.TrackPlaceCity, TrackPlaceCity = track.TrackPlaceCity,
TrackPlaceCityId = track.TrackPlaceCityId, TrackPlaceCityId = track.TrackPlaceCityId,
Files = track.Files?.Select(x => new FileItemDto() Files = track.Files?.Select(x => new FileItemDto
{ {
FileName = x.FileName, FileName = x.FileName,
OssKey = x.OssKey OssKey = x.OssKey
@ -358,15 +253,94 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
OrderType = track.OrderType OrderType = track.OrderType
} }
}, },
SubSystemId = @event.SubSystemId, SubSystemId = track.SubSystemId,
Extra = track.Extra Extra = track.Extra
}, content.SubSystemCode); };
var content = contentDict[track.TrackContentId];
_broadcastingService.TrackChangePublicTopic(eto, content.SubSystemCode);
} }
} }
}
private void WarningCalc_DeletedTrack(List<TrackDetailsIndex> delTracks) private async Task TrackDeleteAfterUpdateOrderAsync(
{ TrackDeleteEvent @event,
_warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); List<TrackDetailsIndex> oldTracks,
} List<TrackDetailsIndex> deleteTracks,
HashSet<int> 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());
}
} }

62
src/TrackCenter.Application/CAP/EventHandlers/TrackPushEventHandler.cs
View File

@ -24,7 +24,6 @@ namespace TrackCenter.Application.CAP.EventHandlers;
public class TrackPushEventHandler : ITransientDependency, ICapSubscribe public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{ {
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _options; private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _options;
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
private readonly TrackPushCacheService _trackPushCacheService; private readonly TrackPushCacheService _trackPushCacheService;
@ -35,7 +34,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
private List<TrackPushContentConfigDto> _pushConfigs = null!; private List<TrackPushContentConfigDto> _pushConfigs = null!;
private Dictionary<int, List<TrackPushContentConfigDto>> _orderTypeSortedConfigsDict = null!; //订单类型,根据sort从小到大排序后的配置字典 private Dictionary<int, List<TrackPushContentConfigDto>> _orderTypeSortedConfigsDict = null!; //订单类型,根据sort从小到大排序后的配置字典
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy, public TrackPushEventHandler(
IOrderRepository orderRepository, IOrderRepository orderRepository,
TrackPushCacheService trackPushCacheService, TrackPushCacheService trackPushCacheService,
ITrackPushRecordEsRepository trackPushRecordEsRepository, ITrackPushRecordEsRepository trackPushRecordEsRepository,
@ -43,7 +42,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
BroadcastingService broadcastingService, BroadcastingService broadcastingService,
ITrackDetailsEsRepository trackDetailsEsRepository) ITrackDetailsEsRepository trackDetailsEsRepository)
{ {
_httpClientProxy = httpClientProxy;
_orderRepository = orderRepository; _orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService; _trackPushCacheService = trackPushCacheService;
_trackPushRecordEsRepository = trackPushRecordEsRepository; _trackPushRecordEsRepository = trackPushRecordEsRepository;
@ -61,7 +59,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
if (pushConfigs == null || pushConfigs.Count == 0) return; if (pushConfigs == null || pushConfigs.Count == 0) return;
if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace()) if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace())
{ {
throw new BusinessException(GlobalConst.CapNoRetryErrorCode, $"该客户{@event.CustomId}未关联服务商，无法推送"); throw new CapNoRetryException($"该客户{@event.CustomId}未关联服务商，无法推送");
} }
_pushConfigs = pushConfigs; _pushConfigs = pushConfigs;
@ -132,20 +130,20 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{ {
var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString());
var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; var serviceProviderCode = _pushConfigs.First().ServiceProviderCode;
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( // var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
_options.CurrentValue.PushTrackInterfaceCode, // _options.CurrentValue.PushTrackInterfaceCode,
serviceProviderCode, // serviceProviderCode,
items.Cast<object>().ToList(), // items.Cast<object>().ToList(),
GlobalConst.TrackPushCompletedSubscribe, // GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString); // attach: asciiString);
//
if (r.Code != GlobalConst.SuccessCode) // if (r.Code != GlobalConst.SuccessCode)
{ // {
TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, // TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
items.Cast<dynamic>().ToList()); // items.Cast<dynamic>().ToList());
throw new BusinessException( // throw new BusinessException(
message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); // 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 asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString());
var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; var serviceProviderCode = _pushConfigs.First().ServiceProviderCode;
var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( // var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
_options.CurrentValue.TrackUpdateInterfaceCode, // _options.CurrentValue.TrackUpdateInterfaceCode,
serviceProviderCode, // serviceProviderCode,
items.Cast<object>().ToList(), // items.Cast<object>().ToList(),
GlobalConst.TrackPushCompletedSubscribe, // GlobalConst.TrackPushCompletedSubscribe,
attach: asciiString); // attach: asciiString);
//
if (r.Code != GlobalConst.SuccessCode) // if (r.Code != GlobalConst.SuccessCode)
{ // {
TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, // TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
items.Cast<dynamic>().ToList()); // items.Cast<dynamic>().ToList());
throw new BusinessException( // throw new BusinessException(
message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); // message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
} // }
} }
} }

13
src/TrackCenter.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -114,17 +114,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
OrderId = x.OrderId, OrderId = x.OrderId,
OrderNo = x.OrderNo, OrderNo = x.OrderNo,
OrderType = x.OrderType, OrderType = x.OrderType,
}).ToList(); });
var idx = 0;
while (idx < identifiers.Count) foreach (var subOrderIds in identifiers.Chunk(GlobalConst.ChunkSize))
{ {
var subEvent = nextEvent.DeepCopy(); var subEvent = nextEvent.DeepCopy();
var cnt = idx + GlobalConst.ChunkSize > identifiers.Count subEvent.OrderIdentifiers = subOrderIds.ToArray();
? identifiers.Count - idx
: GlobalConst.ChunkSize;
subEvent.OrderIdentifiers = identifiers.GetRange(idx, cnt).ToArray();
idx += GlobalConst.ChunkSize;
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent);
} }
} }

16
src/TrackCenter.Application/Dtos/ProvideServices/TrackIdentifierDto.cs
View File

@ -18,4 +18,20 @@ public sealed class TrackIdentifierDto
/// </summary> /// </summary>
[Required] [Required]
public DateTime TrackTime { get; set; } public DateTime TrackTime { get; set; }
/// <summary>
/// 轨迹时区
/// </summary>
public float? TrackTimeZone { get; set; }
/// <summary>
/// 获取UTC0时区时间 (传null/99999视为+8时区)
/// </summary>
public DateTime GetUtc0TrackTime()
{
var timezone = TrackTimeZone ?? 99999f;
return Math.Abs(timezone - 99999f) < 0.01f
? TrackTime.AddHours(-8.0f)
: TrackTime.AddHours(-timezone);
}
} }

10
src/TrackCenter.Application/Dtos/TrackPull/TrackPullParamConfigDto.cs
View File

@ -27,6 +27,16 @@ public class TrackPullParamConfigDto
/// </summary> /// </summary>
public string? TrackStatusName { get; set; } public string? TrackStatusName { get; set; }
/// <summary>
/// 轨迹扩展信息参数名集合（逗号分割）
/// </summary>
public string? TrackExtraNames { get; set; }
/// <summary>
/// 轨迹UTC时间参数名
/// </summary>
public string? TrackUtcTimeName { get; set; }
/// <summary> /// <summary>
/// 轨迹发生ISO时间参数名 /// 轨迹发生ISO时间参数名
/// </summary> /// </summary>

22
src/TrackCenter.Application/Events/ModifyWarningEvent.cs
View File

@ -1,22 +0,0 @@
namespace TrackCenter.Application.Events;
/// <summary>
/// redis 的发布订阅事件
/// </summary>
public class ModifyWarningEvent : RedisEvent
{
public string Key { get; set; }
}
public class RedisEvent : IRedisEvent
{
public string EventType { get; set; }
}
/// <summary>
/// 简单的事件接口，后续可以根据需要扩展
/// </summary>
public interface IRedisEvent
{
string EventType { get; set; }
}

6
src/TrackCenter.Application/Events/PushWarningEvent.cs
View File

@ -1,6 +0,0 @@
namespace TrackCenter.Application.Events;
public class PushWarningEvent
{
public int PageSize { get; set; }
}

29
src/TrackCenter.Application/Events/TrackWarningCalcEvent.cs
View File

@ -1,29 +0,0 @@
namespace TrackCenter.Application.Events;
public class TrackWarningCalcEvent
{
/// <summary>
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 客户id
/// </summary>
public int CustomId { get; set; }
/// <summary>
/// 轨迹状态Id
/// </summary>
public int TrackStatusId { get; set; }
/// <summary>
/// 订单id
/// </summary>
public int[] OrderIds { get; set; }
/// <summary>
/// 轨迹发生时间
/// </summary>
public DateTime TrackTime { get; set; }
}

120
src/TrackCenter.Application/Hangfire/WarningNotifyBuilder.cs
View File

@ -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<ChannelEnum>(1)
{
ChannelEnum.站内信
},
variables = new List<VariableDto>(1)
{
new VariableDto
{
Key = contentKey,
Value = content
}
},
receivers = new List<MsgInstanceReceiverInput>(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<object>
{
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}";
}
}

60
src/TrackCenter.Application/Hangfire/WarningTaskService.cs
View File

@ -3,21 +3,15 @@ using Fake.DependencyInjection;
using FeishuNetSdk; using FeishuNetSdk;
using Hangfire; using Hangfire;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.InterfaceRelay.Application.Contracts;
using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Dtos.TrackPush;
using TrackCenter.Domain.Aggregates.WarningAggregate; using TrackCenter.Domain.Aggregates.WarningAggregate;
using TrackCenter.Domain.Aggregates.WarningAggregate.Return; using TrackCenter.Domain.Aggregates.WarningAggregate.Return;
using TrackCenter.Domain.Dtos.Warning; using TrackCenter.Domain.Dtos.Warning;
using TrackCenter.Domain.Enums; using TrackCenter.Domain.Enums;
using TrackCenter.Domain.Extensions; using TrackCenter.Domain.Extensions;
using TrackCenter.Domain.Queries;
using SJZY.UserCenter.HttpApi.Services.Message;
using TrackCenter.Application.Helpers; using TrackCenter.Application.Helpers;
using TrackCenter.Application.Options; using TrackCenter.Application.Options;
using TrackCenter.Application.Track2InterfaceRelay; using TrackCenter.Application.Track2InterfaceRelay;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;
namespace TrackCenter.Application.Hangfire; namespace TrackCenter.Application.Hangfire;
@ -34,31 +28,22 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
{ {
private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _interfaceRelayOptions; private readonly IOptionsMonitor<InterfaceRelayParamsConfigOptions> _interfaceRelayOptions;
private readonly IOptionsMonitor<TrackWarningTaskOptions> _warningTaskOptions; private readonly IOptionsMonitor<TrackWarningTaskOptions> _warningTaskOptions;
private readonly IHttpClientProxy<IRequestSendMessageService> _requestSendMessageService;
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _interfaceForwardServiceV2;
private readonly IFeishuTenantApi _feishuTenantApi; private readonly IFeishuTenantApi _feishuTenantApi;
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
private readonly IWarningTaskRepository _warningTaskRepository; private readonly IWarningTaskRepository _warningTaskRepository;
private readonly IVFsUserQueries _vfsUserQueries;
public WarningTaskService( public WarningTaskService(
IOptionsMonitor<InterfaceRelayParamsConfigOptions> interfaceRelayOptions, IOptionsMonitor<InterfaceRelayParamsConfigOptions> interfaceRelayOptions,
IOptionsMonitor<TrackWarningTaskOptions> warningTaskOptions, IOptionsMonitor<TrackWarningTaskOptions> warningTaskOptions,
IHttpClientProxy<IInterfaceForwardServiceV2> interfaceForwardServiceV2,
IHttpClientProxy<IRequestSendMessageService> requestSendMessageService,
IFeishuTenantApi feishuTenantApi, IFeishuTenantApi feishuTenantApi,
IWarningDetailsEsRepository warningDetailsEsRepository, IWarningDetailsEsRepository warningDetailsEsRepository,
IWarningTaskRepository warningTaskRepository, IWarningTaskRepository warningTaskRepository)
IVFsUserQueries vfsUserQueries)
{ {
_interfaceRelayOptions = interfaceRelayOptions; _interfaceRelayOptions = interfaceRelayOptions;
_warningTaskOptions = warningTaskOptions; _warningTaskOptions = warningTaskOptions;
_interfaceForwardServiceV2 = interfaceForwardServiceV2;
_requestSendMessageService = requestSendMessageService;
_feishuTenantApi = feishuTenantApi; _feishuTenantApi = feishuTenantApi;
_warningDetailsEsRepository = warningDetailsEsRepository; _warningDetailsEsRepository = warningDetailsEsRepository;
_warningTaskRepository = warningTaskRepository; _warningTaskRepository = warningTaskRepository;
_vfsUserQueries = vfsUserQueries;
} }
#region 用户中心调用，通知（一次） #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 // 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 // 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卡片信息 // 发送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 #endregion
@ -118,24 +94,24 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency
? _interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode ? _interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode
: _interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode; : _interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode;
var r = await _interfaceForwardServiceV2.Service.BatchForwardThenPushToQueueAsync( // var r = await _interfaceForwardServiceV2.Service.BatchForwardThenPushToQueueAsync(
upStreamCode: upStreamCode, // upStreamCode: upStreamCode,
serviceProviderCode: task.ServiceProviderCode, // serviceProviderCode: task.ServiceProviderCode,
data: items, // data: items,
resSubscribeName: GlobalConst.TrackPushCompletedSubscribe, // resSubscribeName: GlobalConst.TrackPushCompletedSubscribe,
attach: Base64Helper.SerializeObject(task.OptionType)); // attach: Base64Helper.SerializeObject(task.OptionType));
if (r.Code != GlobalConst.SuccessCode) // if (r.Code != GlobalConst.SuccessCode)
{ // {
var msg = $"接口通调用失败：BatchForwardThenPushToQueueAsync" + // var msg = $"接口通调用失败：BatchForwardThenPushToQueueAsync" +
$"RequestId: {r.RequestId};SubSystemCode:{task.SubSystemCode};" + // $"RequestId: {r.RequestId};SubSystemCode:{task.SubSystemCode};" +
$" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}"; // $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}";
//
// 处理调用接口通失败，直接处于完结状态 // // 处理调用接口通失败，直接处于完结状态
await HandlerCallBusinessException(task, msg); // await HandlerCallBusinessException(task, msg);
retryCountChanged = true; // retryCountChanged = true;
throw new BusinessException(message: msg); // throw new BusinessException(message: msg);
} // }
} }
} }
catch (Exception e) catch (Exception e)

284
src/TrackCenter.Application/HostServices/WarningCalcBackgroundService.cs Normal file
View File

@ -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;
/// <summary>
/// 预警计算后台任务
/// 预警信号从kafka摄取来进行批处理
/// </summary>
public sealed class WarningCalcBackgroundService : BackgroundService
{
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IConsumer<WarningSignalKey, WarningSignal> _consumer;
private readonly WarningSignalPublisher _warningSignalPublisher;
private TrackWarningTaskOptions _options;
public WarningCalcBackgroundService(
IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor,
IServiceScopeFactory serviceScopeFactory,
IConsumer<WarningSignalKey, WarningSignal> 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<Message<WarningSignalKey, WarningSignal>> messages, CancellationToken ct)
{
ct.ThrowIfCancellationRequested();
// 1.根据类型分组信号
var typeSignalDict = new Dictionary<WarningSignalType, List<WarningSignal>>();
var needRetryMetadata = new List<WarningSignalMetadata>(); // 需要重试的消息
var needEnqueueDlxMetadata = new List<WarningSignalMetadata>(); // 需要投递到死信队列的
var needHandleMessage = new List<Message<WarningSignalKey, WarningSignal>>();
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<WarningSignal>());
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<int, WarningSignal>
{
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<WarningSignalType, List<WarningSignal>> typeSignalDict, CancellationToken ct)
{
var scope = _serviceScopeFactory.CreateAsyncScope();
try
{
ct.ThrowIfCancellationRequested();
var handler = scope.ServiceProvider.GetRequiredService<ICompleteWarningHandler>();
var handler2 = scope.ServiceProvider.GetRequiredService<ITransportWarningHandler>();
var handler3 = scope.ServiceProvider.GetRequiredService<IReturnWarningHandler>();
// tips:处理失败视为整体失败，消息会被重新投递，考虑减小处理批次大小？
var addSignalr = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹新增, Empty<WarningSignal>.List);
var updateSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹更新, Empty<WarningSignal>.List);
var deleteSignals = typeSignalDict.GetValueOrDefault(WarningSignalType.轨迹删除, Empty<WarningSignal>.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<WarningSignal>(addSignalr.Count + updateSignals.Count);
addOrUpdateSignal.AddRange(addSignalr);
addOrUpdateSignal.AddRange(updateSignals);
if (addOrUpdateSignal.Count > 0)
await handler2.TrackAddedOrUpdatedHandleAsync(addOrUpdateSignal);
// 改+删
var updateOrDeleteSignal = new List<WarningSignal>(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<WarningSignalMetadata> 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();
}
}

81
src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs Normal file
View File

@ -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;
/// <summary>
/// 预警任务组件初始化
/// tips:支持定时任务执行周期配置 动态更新并应用 √
/// </summary>
public class WarningTaskComponentHostedService : IHostedService
{
private readonly IOptionsMonitor<TrackWarningTaskOptions> _optionsMonitor;
private TrackWarningTaskOptions? _oldOptions;
private readonly ILogger<WarningTaskComponentHostedService> _logger;
public WarningTaskComponentHostedService(IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor, ILogger<WarningTaskComponentHostedService> 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<TrackWarningTaskOptions> 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<IWarningTaskScanner>("Scan_WarningNotifyTask",
q,
x => x.ScanNotifyTaskAndSchedulingAsync(),
$"0 */{newVal.ScanWarningNotifyTaskHourInterval} * * *");
}
if ((_oldOptions?.ScanReturnWarningRetryPushTaskMinuteInterval ?? 0) != newVal.ScanReturnWarningRetryPushTaskMinuteInterval)
{
var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanReturnWarningRetryPushTaskLevel];
// 回传预警通知重试推送任务扫描
RecurringJob.AddOrUpdate<IWarningTaskScanner>("Scan_ReturnWarningRetryPushTask",
q,
x => x.ScanReturnRetryPushTaskAndSchedulingAsync(),
$"*/{newVal.ScanReturnWarningRetryPushTaskMinuteInterval} * * * *");
}
if ((_oldOptions?.ScanTransportWarningDetailsGenerateTaskMinuteInterval ?? 0) != newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval)
{
var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanTransportWarningDetailsGenerateTaskLevel];
// 运输预警明细生成任务扫描
RecurringJob.AddOrUpdate<IWarningTaskScanner>("Scan_TransportWarningDetailsGenerateTask",
q,
x => x.ScanTransportDetailsGenerateTaskAndSchedulingAsync(),
$"*/{newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval} * * * *");
}
_oldOptions = newVal;
}
}

81
src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs
View File

@ -4,7 +4,6 @@ using Fake.DependencyInjection;
using Newtonsoft.Json; using Newtonsoft.Json;
using RabbitMQ.Client.Events; using RabbitMQ.Client.Events;
using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Dtos.TrackPush;
using TrackCenter.Application.Services;
using TrackCenter.Application.Helpers; using TrackCenter.Application.Helpers;
using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.Broadcasting;
using TrackCenter.Application.Services.Warning.Handlers; 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) public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options)
{ {
var msg = Encoding.UTF8.GetString(args.Body.ToArray()); var msg = string.Empty;
var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg); try
if (@event == null)
{ {
LogHelper.Warn( msg = Encoding.UTF8.GetString(args.Body.ToArray());
$"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg);
return; 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<string>(@event.Attach); var optionType = Base64Helper.DeSerialize<string>(@event.Attach);
var pushInputs = new List<TrackPushInput>(); var pushInputs = new List<TrackPushInput>();
var updateInputs = new List<TrackUpdateInput>(); var updateInputs = new List<TrackUpdateInput>();
@ -91,18 +103,8 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
// 同步更新订单主表 // 同步更新订单主表
await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs); await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs);
// todo:引发预警计算 改MQ批量+分流 // 引发预警计算
await _returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs); 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<TrackPushRecordIndex> logs) private async Task TrackPushAfterUpdateOrderAsync(bool isAdd, bool isSuccess, List<TrackPushRecordIndex> 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) private static TrackPushRecordDto BuildDto(TrackPushRecordIndex log, string interfaceRelayData)
{ {
return new TrackPushRecordDto return new TrackPushRecordDto

133
src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs
View File

@ -14,6 +14,7 @@ using TrackCenter.Application.Services.TrackPull;
using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
using TrackCenter.Domain.Enums; using TrackCenter.Domain.Enums;
using TrackCenter.Domain.Extensions;
namespace TrackCenter.Application.Rabbit; namespace TrackCenter.Application.Rabbit;
@ -39,7 +40,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
public ConsumeOptions Options() public ConsumeOptions Options()
{ {
return new ConsumeOptions() return new ConsumeOptions
{ {
Queue = GlobalConst.TrackRefreshSubscribe, Queue = GlobalConst.TrackRefreshSubscribe,
}; };
@ -47,16 +48,24 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options)
{ {
var msg = Encoding.UTF8.GetString(args.Body.ToArray()); var msg = string.Empty;
var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg); try
if (@event == null)
{ {
_logger.LogWarning( msg = Encoding.UTF8.GetString(args.Body.ToArray());
$"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); var @event = System.Text.Json.JsonSerializer.Deserialize<InterfaceRelayForwardReplayEvent>(msg);
return; 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) public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
@ -102,7 +111,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
TrackRefreshAttach attach) TrackRefreshAttach attach)
{ {
var paramConfig = paramsWithContent.ParamConfig; var paramConfig = paramsWithContent.ParamConfig;
var orderNo = orderToken[paramConfig.OrderNoName]?.Value<string>(); var orderNo = orderToken.SelectToken(paramConfig.OrderNoName)?.Value<string>();
if (orderNo.IsNullOrWhiteSpace()) return; if (orderNo.IsNullOrWhiteSpace()) return;
var trackToken = orderToken.SelectToken(paramConfig.TrackDetailsPath); var trackToken = orderToken.SelectToken(paramConfig.TrackDetailsPath);
@ -158,14 +167,14 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
// 轨迹内容 // 轨迹内容
var trackContent = paramConfig.TrackContentName == null var trackContent = paramConfig.TrackContentName == null
? null ? null
: trackToken[paramConfig.TrackContentName]?.Value<string>(); : trackToken.SelectToken(paramConfig.TrackContentName)?.Value<string>();
// 1、匹配轨迹代码 // 1、匹配轨迹代码
// 2、匹配轨迹状态 // 2、匹配轨迹状态
if (matchedContentConfig == null && !paramConfig.TrackStatusName.IsNullOrWhiteSpace()) if (matchedContentConfig == null && !paramConfig.TrackStatusName.IsNullOrWhiteSpace())
{ {
var trackStatus = trackToken[paramConfig.TrackStatusName!]?.Value<string>(); var trackStatus = trackToken.SelectToken(paramConfig.TrackStatusName!)?.Value<string>();
if (!trackStatus.IsNullOrWhiteSpace()) if (!trackStatus.IsNullOrWhiteSpace())
{ {
matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValues.Contains(trackStatus!)); matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValues.Contains(trackStatus!));
@ -211,26 +220,36 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
// 轨迹发生ISO时间 // 轨迹发生ISO时间
var trackIsoTime = paramConfig.TrackIsoTimeName == null var trackIsoTime = paramConfig.TrackIsoTimeName == null
? null ? null
: trackToken[paramConfig.TrackIsoTimeName]?.ToObject<DateTimeOffset>(); : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.ToObject<DateTimeOffset>();
// 轨迹发生UTC时间
var trackUtcTime = paramConfig.TrackUtcTimeName == null
? null
: trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.ToObject<DateTime>();
// 轨迹发生时间 // 轨迹发生时间
var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null
? null ? null
: trackToken[paramConfig.TrackTimeName]?.ToObject<DateTime>()); : trackToken.SelectToken(paramConfig.TrackTimeName)?.ToObject<DateTime>());
// 轨迹发生时区 // 轨迹发生时区
var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null
? null ? null
: trackToken[paramConfig.TrackTimeZoneName]?.ToObject<float>()); : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.ToObject<float>());
// 轨迹发生地 // 轨迹发生地
var trackPlace = paramConfig.TrackPlaceName == null var trackPlace = paramConfig.TrackPlaceName == null
? null ? null
: trackToken[paramConfig.TrackPlaceName]?.ToObject<string>(); : trackToken.SelectToken(paramConfig.TrackPlaceName)?.ToObject<string>();
// 轨迹扩展信息，根据输入的多个字段提取报文内容，构建一个新的对象的json形式
var trackExtra = paramConfig.TrackExtraNames == null
? null
: ExtraBuild(trackToken, paramConfig.TrackExtraNames);
// 计算时间和时区信息
(trackTime, trackTimeZone) = CalcTimeAndTimezone(trackIsoTime, trackUtcTime, trackTime, trackTimeZone);
if (matchedContentConfig is null) if (matchedContentConfig is null)
{ {
_logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}", _logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}",
context.OrderNo, trackToken.ToString()); context.OrderNo, trackToken.ToString());
await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, trackContent); await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime.Value, trackTimeZone.Value, trackContent, trackExtra);
return; return;
} }
@ -257,9 +276,10 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
CustomId = context.CustomId, CustomId = context.CustomId,
TrackCode = matchedContentConfig.TrackContentCode, TrackCode = matchedContentConfig.TrackContentCode,
TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackPlace = trackPlace ?? GlobalConst.EmptyString,
TrackTime = trackTime ?? DateTime.Now, TrackTime = trackTime.Value,
TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, TrackTimeZone = trackTimeZone.Value,
IsAdd = true, IsAdd = true,
//S06新增
//服务商轨迹默认不对外 //服务商轨迹默认不对外
IsExternal = false, IsExternal = false,
Source = (int) TrackSourceEnum.ExternalServiceProvide, Source = (int) TrackSourceEnum.ExternalServiceProvide,
@ -276,13 +296,14 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
TrackPlacePostcode = null, TrackPlacePostcode = null,
DestinationTimeZone = null, DestinationTimeZone = null,
EstimatedDeliveryDate = null, EstimatedDeliveryDate = null,
Extra = trackExtra
}; };
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent);
} }
private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, 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 // 新增服务商轨迹到es
var track = new TrackDetailsIndex var track = new TrackDetailsIndex
@ -293,8 +314,8 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
OrderType = context.OrderType, OrderType = context.OrderType,
OrderNo = context.OrderNo, OrderNo = context.OrderNo,
TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackPlace = trackPlace ?? GlobalConst.EmptyString,
TrackTime = trackTime ?? DateTime.Now, TrackTime = trackTime,
TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, TrackTimeZone = trackTimeZone,
ServiceProviderCode = context.ServiceProviderCode, ServiceProviderCode = context.ServiceProviderCode,
OriginTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)), OriginTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)),
OriginTrackContent = trackContent ?? GlobalConst.EmptyString, OriginTrackContent = trackContent ?? GlobalConst.EmptyString,
@ -307,6 +328,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
TrackContentId = GlobalConst.NotExistId, TrackContentId = GlobalConst.NotExistId,
TrackContent = GlobalConst.EmptyString, TrackContent = GlobalConst.EmptyString,
TrackEnContent = GlobalConst.EmptyString, TrackEnContent = GlobalConst.EmptyString,
//S06新增轨迹明细字段
//服务商轨迹默认不对外 //服务商轨迹默认不对外
IsExternal = false, IsExternal = false,
Source = (int) TrackSourceEnum.ExternalServiceProvide, Source = (int) TrackSourceEnum.ExternalServiceProvide,
@ -321,7 +343,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
TrackPlacePostcode = null, TrackPlacePostcode = null,
DestinationTimeZone = null, DestinationTimeZone = null,
EstimatedDeliveryDate = null, EstimatedDeliveryDate = null,
Extra = null Extra = extra
}; };
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
@ -330,4 +352,69 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
await _trackDetailsEsRepository.AddAsync(track); 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);
}
} }

42
src/TrackCenter.Application/Services/ProvideServices/OrderService.cs
View File

@ -1,14 +1,10 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.Application; using Fake.Application;
using InterfaceForward.Domain.Shared.Helpers; 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.Dtos.ProvideServices;
using TrackCenter.Application.Services.ProvideServices;
using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
using TrackCenter.Domain.Extensions; using TrackCenter.Domain.Extensions;
using SJZY.UserCenter.HttpApi.Services;
using SqlSugar; using SqlSugar;
using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Application.Services.Warning.Handlers;
using TrackCenter.Repository.Aggregates.OrderAggregate; using TrackCenter.Repository.Aggregates.OrderAggregate;
@ -20,7 +16,7 @@ namespace TrackCenter.Application.Services.ProvideServices;
/// </summary> /// </summary>
[ApiExplorerSettings(GroupName = "提供接口")] [ApiExplorerSettings(GroupName = "提供接口")]
[Route("/ProviderService")] [Route("/ProviderService")]
public partial class OrderService : ApplicationServiceBase, IOrderService public partial class OrderService : ApplicationServiceBase
{ {
private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
@ -95,19 +91,19 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
{ {
OrderId = orderId, OrderId = orderId,
OrderNo = input.OrderNo, OrderNo = input.OrderNo,
OptionType = OrderOperationType.创建订单.ToString(), OptionType = nameof(OrderOperationType.创建订单),
Content = input.ToJson(), Content = input.ToJson(),
CreateUserId = currentUser.UserId, CreateUserId = CurrentUser.Id ?? Guid.Empty,
CreateUserName = currentUser.Name, CreateUserName = CurrentUser.UserName ?? string.Empty,
CreateTime = DateTime.Now CreateTime = DateTime.Now
}; };
_ = _orderOperationRecordEsRepository.AddAsync(indexForAdd); _ = _orderOperationRecordEsRepository.AddAsync(indexForAdd);
return new UnifyResultDto<IdBaseDto>(true, new IdBaseDto return new IdBaseDto
{ {
Id = orderId Id = orderId
}); };
} }
/// <summary> /// <summary>
@ -117,7 +113,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
/// <returns></returns> /// <returns></returns>
/// <exception cref="BusinessException"></exception> /// <exception cref="BusinessException"></exception>
[HttpPost("Order/BatchUpsert")] [HttpPost("Order/BatchUpsert")]
public async Task<UnifyResultDto> BatchUpsertAsync(BatchUpsertOrderInput input) public async Task BatchUpsertAsync(BatchUpsertOrderInput input)
{ {
if (input.OrderInputs.Any(x => (x.EstimatedDeliveryDate.HasValue && x.DestinationTimeZone.HasValue == false) || if (input.OrderInputs.Any(x => (x.EstimatedDeliveryDate.HasValue && x.DestinationTimeZone.HasValue == false) ||
(x.EstimatedDeliveryDate.HasValue == false && x.DestinationTimeZone.HasValue))) (x.EstimatedDeliveryDate.HasValue == false && x.DestinationTimeZone.HasValue)))
@ -125,8 +121,6 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
throw new BusinessException(message: "【预计签收时间】和【签收地时区】必须同时输入"); throw new BusinessException(message: "【预计签收时间】和【签收地时区】必须同时输入");
} }
var currentUser = CurrentUserRequest.CurrentUser;
var exp = Expressionable.Create<OrderEntity>(); var exp = Expressionable.Create<OrderEntity>();
foreach (var order in input.OrderInputs) foreach (var order in input.OrderInputs)
exp.Or(x => x.OrderType == order.OrderType && x.SubSystemId == order.SubSystemId && x.OrderNo == order.OrderNo); 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, OrderId = order.Id,
OrderNo = order.OrderNo, OrderNo = order.OrderNo,
OptionType = OrderOperationType.修改订单.ToString(), OptionType = nameof(OrderOperationType.修改订单),
Content = input2.ToJson(), Content = input2.ToJson(),
CreateUserId = currentUser.UserId, CreateUserId = CurrentUser.Id ?? Guid.Empty,
CreateUserName = currentUser.Name, CreateUserName = CurrentUser.UserName ?? string.Empty,
CreateTime = DateTime.Now CreateTime = DateTime.Now
}); });
} }
@ -163,8 +157,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
} }
// 如果全部操作为修改 // 如果全部操作为修改
if (existsOrders.Count == input.OrderInputs.Count) if (existsOrders.Count == input.OrderInputs.Count) return;
return new UnifyResultDto(true);
// 新增订单 // 新增订单
var existsOrderNos = existsOrders.Select(x => x.OrderNo).ToHashSet(); var existsOrderNos = existsOrders.Select(x => x.OrderNo).ToHashSet();
@ -188,8 +181,8 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
OrderNo = order.OrderNo, OrderNo = order.OrderNo,
OptionType = OrderOperationType.创建订单.ToString(), OptionType = OrderOperationType.创建订单.ToString(),
Content = insertInput.ToJson(), Content = insertInput.ToJson(),
CreateUserId = currentUser.UserId, CreateUserId = CurrentUser.Id ?? Guid.Empty,
CreateUserName = currentUser.Name, CreateUserName = CurrentUser.UserName ?? string.Empty,
CreateTime = DateTime.Now CreateTime = DateTime.Now
}); });
} }
@ -203,7 +196,6 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
} }
_ = _orderOperationRecordEsRepository.AddManyAsync(createRecords); _ = _orderOperationRecordEsRepository.AddManyAsync(createRecords);
return new UnifyResultDto(true);
} }
/// <summary> /// <summary>
@ -213,7 +205,7 @@ public partial class OrderService : ApplicationServiceBase, IOrderService
/// <returns></returns> /// <returns></returns>
/// <exception cref="BusinessException"></exception> /// <exception cref="BusinessException"></exception>
[HttpPost("Order/Delete")] [HttpPost("Order/Delete")]
public async Task<UnifyResultDto<IdBaseDto>> DeleteOrderAsync(DeleteOrderInput input) public async Task<IdBaseDto> DeleteOrderAsync(DeleteOrderInput input)
{ {
var order = await _orderRepository.GetFirstAsync(x => var order = await _orderRepository.GetFirstAsync(x =>
x.SubSystemId == input.SubSystemId && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo); 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); await _orderRepository.DeleteOrderAsync(order.Id);
// todo:预警计算 // 预警计算
WarningCalc_OrderDeletedOrCompleted(order); WarningCalc_OrderDeletedOrCompleted(order);
return new UnifyResultDto<IdBaseDto>(true, new IdBaseDto return new IdBaseDto
{ {
Id = order.Id Id = order.Id
}); };
} }
private async Task UpdateOrderAsync(UpdateOrderInput input, OrderEntity order) private async Task UpdateOrderAsync(UpdateOrderInput input, OrderEntity order)

2
src/TrackCenter.Application/Services/TrackOrders/TrackOrderService.cs
View File

@ -100,7 +100,7 @@ public class TrackOrderService : ApplicationServiceBase
x.IsCompleted x.IsCompleted
}); });
// todo:预警计算 // 预警计算
if (order.IsCompleted) if (order.IsCompleted)
WarningCalc_OrderDeletedOrCompleted(order); WarningCalc_OrderDeletedOrCompleted(order);

5
src/TrackCenter.Domain/Consts/GlobalConst.ES.cs
View File

@ -12,11 +12,6 @@ public partial class GlobalConst
/// </summary> /// </summary>
public const string TrackPushRecordIndex = "sjzy_track_push_records"; public const string TrackPushRecordIndex = "sjzy_track_push_records";
/// <summary>
/// 运输时效预警记录 es索引名称（模板按月） todo:可以删除
/// </summary>
public const string TrackWarningRecordIndex = "sjzy_track_warning_records";
/// <summary> /// <summary>
/// 订单操作记录 es索引名称（模板按月） /// 订单操作记录 es索引名称（模板按月）
/// </summary> /// </summary>

5
src/TrackCenter.Domain/Consts/GlobalConst.MQ.cs
View File

@ -47,11 +47,6 @@ public partial class GlobalConst
/// </summary> /// </summary>
public const string TrackChangeAfterUpdateOrderSubscribe = "TrackCenter.TrackChangeAfterUpdateOrder"; public const string TrackChangeAfterUpdateOrderSubscribe = "TrackCenter.TrackChangeAfterUpdateOrder";
/// <summary>
/// 预警计算任务
/// </summary>
public const string TrackWarningCalcSubscribe = "TrackCenter.TrackWarningCalc";
/// <summary> /// <summary>
/// 轨迹变更通知交换机 /// 轨迹变更通知交换机
/// </summary> /// </summary>

5
src/TrackCenter.Domain/Consts/GlobalConst.Redis.cs
View File

@ -2,11 +2,6 @@ namespace TrackCenter.Domain.Consts;
public partial class GlobalConst public partial class GlobalConst
{ {
/// <summary>
/// 轨迹要推送的客户redis key前缀
/// </summary>
public const string TrackPushKeyPrefix = "TrackPush";
/// <summary> /// <summary>
/// 轨迹推送内容配置redis key前缀 /// 轨迹推送内容配置redis key前缀
/// </summary> /// </summary>

2
src/TrackCenter.Domain/Consts/GlobalConst.cs
View File

@ -17,5 +17,7 @@ public partial class GlobalConst
public const string CapNoRetryErrorCode = "599"; 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); public static readonly DateTime MaxDateTime = new DateTime(2999, 1, 1, 0, 0, 0);
} }