diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs index 47a507d2..3fd35deb 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs @@ -7,11 +7,6 @@ public class TrackChangeAfterUpdateOrderEvent /// public int[] OrderIds { get; set; } = null!; - /// - /// 是否删除引起的更新 - /// - public bool IsDelete { get; set; } - /// /// 这条轨迹是否意味着完成（只有刷新会设为true） /// diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs index 20d168f2..94cbc37d 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs @@ -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 /// 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); } diff --git a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index 25e74d44..3f586b15 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs @@ -36,53 +36,25 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa if (orders.Count == 0) return; - // todo：delete可能发生回退，直接去es查最新轨迹，但是可能es还没有刷盘，导致回退失效，可以考虑加个定时任务弥补 - if (@event.IsDelete) + foreach (var order in orders) { - foreach (var order in orders) + if (order.LatestTrackTime != null && + @event.TrackTime.Subtract(TimeSpan.FromHours(@event.TrackTimeZone)) < + order.LatestTrackTime.Value.Subtract(TimeSpan.FromHours(order.LatestTrackTimeZone ?? 0))) { - 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; - } + continue; } - } - else // upsert可以直接计算 - { - foreach (var order in orders) - { - if (order.LatestTrackTime != null && - @event.TrackTime.Subtract(TimeSpan.FromHours(@event.TrackTimeZone)) < - order.LatestTrackTime.Value.Subtract(TimeSpan.FromHours(order.LatestTrackTimeZone ?? 0))) - { - 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; } + + 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 diff --git a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs index ccd4d0d5..e9ed2dec 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs @@ -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 + { + 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) { - // 触发订单主表更新事件 - await _capPublisher.PublishDelayAsync(TimeSpan.FromSeconds(3), - GlobalConst.TrackChangeAfterUpdateOrderSubscribe, - new TrackChangeAfterUpdateOrderEvent - { - 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 + }); } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs index a0df5114..9bd76c5e 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs @@ -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); } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 8059fcf6..a162428a 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -39,10 +39,11 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return; } - await HandleTrackAsync(@event); + var canNextOrderIds = new List(); + 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 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(); diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index b35145a4..a61e4f95 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -6,7 +6,8 @@ public interface ITrackDetailsRepository { Task> GetListAsync(int[] orderIds, string trackCode); - Task> GetListAsync(int[] orderIds); + Task> GetLatestListAsync(int[] eventOrderIds, List trackCodes); + Task AnyAsync(int orderId, DateTime trackTime); @@ -16,17 +17,12 @@ public interface ITrackDetailsRepository Task AddManyAsync(List index); + Task AddAsync(TrackDetailsIndex track); Task> GetListAsync(int orderType, List orderNos); Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input); Task> GetListAsync(int orderId); - /// - /// 删除轨迹 - /// - /// - /// - /// - Task DeleteTrackAsync(int[] orderIds, List trackCodes); + Task DeleteTrackAsync(int[] orderIds, List trackCodes); } \ No newline at end of file diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index 61950329..238a9fd0 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -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> GetListAsync(int[] orderIds) + public async Task> GetLatestListAsync(int[] orderIds, List trackCodes) { + var excludeTrackCodes = trackCodes.ToList(); + excludeTrackCodes.Add(GlobalConst.EmptyString); var data = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd .Field(f => f.TrackEnContent) @@ -40,6 +42,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 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 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) + ) + ) + ) + ) + .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) + ) ) ) ) ) - .Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending)) - .Size(1) ); return data.Documents.ToList(); @@ -98,18 +113,34 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 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 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 DeleteTrackAsync(int[] orderIds, List trackCodes) + public async Task DeleteTrackAsync(int[] orderIds, List trackCodes) { var r = await Context.DeleteByQueryAsync(x => x .Query( @@ -164,8 +195,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 0; } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input)