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

Merge remote-tracking branch 'origin/S03-轨迹中台优化项' into test

Browse Source
This commit is contained in:
dingenwen 2024-04-16 15:33:17 +08:00
commit b4a93a005b
8 changed files with 117 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs
View File

@ -7,11 +7,6 @@ public class TrackChangeAfterUpdateOrderEvent
/// </summary>
public int[] OrderIds { get; set; } = null!;
/// <summary>
/// 是否删除引起的更新
/// </summary>
public bool IsDelete { get; set; }
/// <summary>
/// 这条轨迹是否意味着完成（只有刷新会设为true）
/// </summary>

11
src/SJZY.Track.Application/EventHandlers/CAPFilter.cs
View File

@ -21,8 +21,7 @@ public class CapFilter : SubscribeFilter
context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff");
// 轨迹增改要做幂等
if (!context.DeliverMessage.Headers["cap-msg-name"]
.IsIn(GlobalConst.TrackUpsertSubscribe))
if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe))
return Task.CompletedTask;
// 本地调试不做幂等
@ -76,11 +75,13 @@ public class CapFilter : SubscribeFilter
/// <returns></returns>
public override Task OnSubscribeExceptionAsync(ExceptionContext context)
{
context.ExceptionHandled = true;
context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
{
context.ExceptionHandled = true; // 幂等不做重试
LogHelper.Warn(ex.ToString());
}
else
LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}");
@ -101,14 +102,14 @@ public class CapFilter : SubscribeFilter
var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture);
var useTime = DateTime.Now - parsedDate;
if (useTime.Minutes > 20)
if (useTime.Minutes > 10)
{
var log = $@"
|-消息编号：{messageId}
|-消费者名：{receiveName}
|-跳过执行：{(isFromExecptionFliter ? "是" : "否")}
|-执行时间：{receiveTime}
|-消费耗时：{useTime.Milliseconds}毫秒
|-消费耗时：{useTime.Seconds}秒
|-请求参数：{arguments}";
LogHelper.Error(log);
}

36
src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs
View File

@ -36,36 +36,6 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
if (orders.Count == 0) return;
// todo：delete可能发生回退，直接去es查最新轨迹，但是可能es还没有刷盘，导致回退失效，可以考虑加个定时任务弥补
if (@event.IsDelete)
{
foreach (var order in orders)
{
var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds);
var track = tracks.Where(x => x.OrderId == order.Id).MaxBy(x => x.TrackTime);
// 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) return;
// tips：订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
}
else
{
// 删除引发的回退更新
order.IsCompleted = @event.IsCompleted;
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
}
}
else // upsert可以直接计算
{
foreach (var order in orders)
{
if (order.LatestTrackTime != null &&
@ -75,15 +45,17 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
continue;
}
// upsert引发的更新
if (!order.IsCompleted)
{
order.IsCompleted = @event.IsCompleted;
}
order.LatestTrackContent = @event.TrackContent;
order.LatestTrackEnContent = @event.TrackEnContent;
order.LatestTrackPlace = @event.TrackPlace;
order.LatestTrackTime = @event.TrackTime;
order.LatestTrackTimeZone = @event.TrackTimeZone;
}
}
await _orderRepository.UpdateRangeAsync(orders, x => new
{

51
src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -74,19 +74,50 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
private async Task DeleteTrackAsync(TrackDeleteEvent @event)
{
// 删除轨迹（删的是同子系统下的单，所以code是统一的）
var isOk = await _trackDetailsRepository.DeleteTrackAsync(@event.OrderIds, @event.TrackCodes);
var tracks = await _trackDetailsRepository.GetLatestListAsync(@event.OrderIds, @event.TrackCodes);
if (isOk)
await _trackDetailsRepository.DeleteTrackAsync(@event.OrderIds, @event.TrackCodes);
var orders = await _orderRepository.GetListAsync(x => @event.OrderIds.Contains(x.Id),
x => new OrderEntity
{
// 触发订单主表更新事件
await _capPublisher.PublishDelayAsync(TimeSpan.FromSeconds(3),
GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
new TrackChangeAfterUpdateOrderEvent
Id = x.Id,
LatestTrackContent = x.LatestTrackContent,
LatestTrackEnContent = x.LatestTrackEnContent,
LatestTrackPlace = x.LatestTrackPlace,
LatestTrackTime = x.LatestTrackTime,
LatestTrackTimeZone = x.LatestTrackTimeZone,
});
if (orders.Count == 0) return;
foreach (var order in orders)
{
OrderIds = @event.OrderIds,
TrackCode = @event.TrackCodes.JoinAsString(","),
IsDelete = true
var track = tracks.FirstOrDefault(x => x.OrderId == order.Id);
// 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) return;
// tips：订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
}
else
{
// 删除引发的回退更新
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
}
await _orderRepository.UpdateRangeAsync(orders, x => new
{
x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
x.LatestTrackTimeZone
});
}
}
}

4
src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs
View File

@ -20,7 +20,7 @@ public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSub
{
if (await _trackDetailsRepository.AnyAsync(@event.OrderId, @event.TrackTime))
{
return; // 无法映射的轨迹只做新增
return;
}
// 新增服务商轨迹到es
@ -45,6 +45,6 @@ public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSub
track.TrackEnContent = GlobalConst.EmptyString;
track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
await _trackDetailsRepository.UpdateAsync(track);
await _trackDetailsRepository.AddAsync(track);
}
}

14
src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -39,10 +39,11 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return;
}
await HandleTrackAsync(@event);
var canNextOrderIds = new List<int>();
await HandleTrackAsync(@event, canNextOrderIds);
// 根据订单关联关系异步向下刷（异步）
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(orderIds);
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray());
foreach (var order in @event.OrderIdentifiers)
{
// 理论上小包在这不会分组
@ -88,7 +89,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
}
}
private async Task HandleTrackAsync(TrackUpsertEvent @event)
private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
{
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode);
@ -101,15 +102,14 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
if (@event.IsAdd != null)
{
if (track == null && !@event.IsAdd!.Value) return; // 不存在不让改
if (track != null && @event.IsAdd!.Value) return; // 存在不让创建
if (track == null && !@event.IsAdd!.Value) continue; // 不存在不让改
if (track != null && @event.IsAdd!.Value) continue; // 存在不让创建
}
if (track == default)
{
track = new TrackDetailsIndex
{
IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}",
OrderId = order.OrderId,
OrderNo = order.OrderNo,
OrderType = order.OrderType,
@ -168,6 +168,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
};
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
canNextOrderIds.Add(order.OrderId);
}
var firstTrack = insertTracks.FirstOrDefault();

12
src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs
View File

@ -6,7 +6,8 @@ public interface ITrackDetailsRepository
{
Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode);
Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds);
Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] eventOrderIds, List<string> trackCodes);
Task<bool> AnyAsync(int orderId, DateTime trackTime);
@ -16,17 +17,12 @@ public interface ITrackDetailsRepository
Task AddManyAsync(List<TrackDetailsIndex> index);
Task AddAsync(TrackDetailsIndex track);
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos);
Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input);
Task<List<TrackDetailsIndex>> GetListAsync(int orderId);
/// <summary>
/// 删除轨迹
/// </summary>
/// <param name="orderIds"></param>
/// <param name="trackCodes"></param>
/// <returns></returns>
Task<bool> DeleteTrackAsync(int[] orderIds, List<string> trackCodes);
Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes);
}

53
src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs
View File

@ -1,7 +1,7 @@
using Nest;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ElasticSearch.Services;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Dtos.TrackDetails;
namespace SJZY.Track.Repository.Repositories;
@ -30,8 +30,10 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
return r.ToIdList();
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds)
public async Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] orderIds, List<string> trackCodes)
{
var excludeTrackCodes = trackCodes.ToList();
excludeTrackCodes.Add(GlobalConst.EmptyString);
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.TrackEnContent)
@ -40,6 +42,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
.Field(f => f.TrackTime)
.Field(f => f.TrackTimeZone)
.Field(f => f.OrderId)
.Field(f => f.TrackCode)
))
.Query(q => q
.Bool(b => b
@ -47,15 +50,27 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
)
.MustNot(mn => mn
.Term(t => t
.Terms(t => t
.Field(f => f.TrackCode.Suffix("keyword"))
.Value("--")
.Terms(excludeTrackCodes)
)
)
)
)
.Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending))
.Size(orderIds.Length)
.Aggregations(a => a
.Terms("orders", t => t
.Field(f => f.OrderId)
.Aggregations(aa => aa
.TopHits("latest_track", th => th
.Size(1)
.Sort(so => so
.Field(f => f.ZeroTrackTime, SortOrder.Descending)
)
)
)
)
)
);
return data.Documents.ToList();
@ -98,18 +113,34 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
{
LogHelper.Error($"轨迹更新失败，原因：{r.DebugInformation}。");
throw new Exception($"轨迹更新失败，原因：{r.DebugInformation}。");
}
}
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
{
// 按天写入es
var r = await Context.IndexManyAsync(tracks, tracks.First().IndexName);
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
tracks.ForEach(x => x.IndexName = indexName);
var r = await Context.IndexManyAsync(tracks, indexName);
if (!r.IsValid)
{
LogHelper.Error($"{tracks.Count}条轨迹新增失败，原因：{r.DebugInformation}。");
throw new Exception($"{tracks.Count}条轨迹新增失败，原因：{r.DebugInformation}。");
}
}
public async Task AddAsync(TrackDetailsIndex track)
{
// 按天写入es
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
track.IndexName = indexName;
var r = await Context.IndexAsync(track, s => s.Index(indexName));
if (!r.IsValid)
{
throw new Exception($"轨迹新增失败，原因：{r.DebugInformation}。");
}
}
@ -146,12 +177,12 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
q.Term(t => t.OrderId, orderId)
)
.Size(1000)
.Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending)));
.Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending)));
return data.Documents.ToList();
}
public async Task<bool> DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
{
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
.Query(
@ -164,8 +195,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid)
throw new Exception($"轨迹数据删除失败，原因：{r.DebugInformation}");
return r.Deleted > 0;
}
public async Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input)