From dc44378f454b23aceaf2224870391b24b891e81a Mon Sep 17 00:00:00 2001 From: dingenwen <543338760@qq.com> Date: Wed, 10 Apr 2024 17:40:09 +0800 Subject: [PATCH 1/7] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Api/appsettings.json | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/SJZY.Track.Api/appsettings.json b/src/SJZY.Track.Api/appsettings.json index b88d44ad..676973d6 100644 --- a/src/SJZY.Track.Api/appsettings.json +++ b/src/SJZY.Track.Api/appsettings.json @@ -7,9 +7,7 @@ "Env": "DEV", "Namespace": "Dev", "MetaServer": "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080", - "ConfigServer": [ - "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080" - ] + "ConfigServer": [ "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080" ] } }, "Kestrel": { From d95c5da96b1c97ee39089a276da77f937ef610df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 15 Apr 2024 11:08:04 +0800 Subject: [PATCH 2/7] =?UTF-8?q?fix=EF=BC=9A=E8=B7=A8=E5=AD=90=E7=B3=BB?= =?UTF-8?q?=E7=BB=9F=E6=98=A0=E5=B0=84=E8=B5=8B=E5=80=BC=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 192a1b87..278667aa 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -80,7 +80,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return; } - @event.TrackCode = mapCode; + nextEvent.TrackCode = mapCode; } await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent); From a28a9ac5dbe6ec5a8b20ef61d0a2e341c2a0dda4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 15 Apr 2024 14:09:12 +0800 Subject: [PATCH 3/7] =?UTF-8?q?change=EF=BC=9A=E8=BD=A8=E8=BF=B9=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E4=B8=BA--=E7=9A=84=E4=B8=8D=E5=8F=82=E4=B8=8E?= =?UTF-8?q?=E8=AE=A1=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 12 +++---- .../Repositories/TrackDetailsRepository.cs | 31 ++++++++++++------- 2 files changed, 25 insertions(+), 18 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 278667aa..277ad96b 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -60,9 +60,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe }).ToArray(); nextEvent.TrackRefreshData = new TrackRefreshStruct { - ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString, - OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, - OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString, + ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, + OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, + OriginTrackData = @event.TrackRefreshData.OriginTrackData, // tips：一定要重置为false，父级订单的完成不会使得关联订单完成，关联单也必须由刷新来完成或手动设置完成！！ IsCompleted = false }; @@ -118,9 +118,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, - ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, - OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, - OriginTrackData = @event.TrackRefreshData.OriginTrackData, + ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString, + OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, + OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString, IsCanDelete = false, CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index b692a3f2..d1345e90 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -2,7 +2,6 @@ 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; @@ -18,12 +17,13 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(s => s .Source() - .Query(q => - q.Terms(t => t - .Field(f => f.OrderId) - .Terms(orderIds)) - && q.Term(f => f.TrackCode.Suffix("keyword"), trackCode) - && q.Term(f => f.ServiceProviderName.Suffix("keyword"), GlobalConst.EmptyString) + .Query(q => q + .Bool(b => b + .Must( + mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), + mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode)) + ) + ) ) ); @@ -41,11 +41,18 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract f.TrackTimeZone) .Field(f => f.OrderId) )) - .Query(q => - q.Terms(t => t - .Field(f => f.OrderId) - .Terms(orderIds)) - && q.Term(f => f.ServiceProviderName.Suffix("keyword"), GlobalConst.EmptyString) + .Query(q => q + .Bool(b => b + .Must( + mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) + ) + .MustNot(mn => mn + .Term(t => t + .Field(f => f.TrackCode.Suffix("keyword")) + .Value("--") + ) + ) + ) ) .Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending)) .Size(1) From 0fb2690efdfa4669d06f7050f724e4c0f3413284 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 15 Apr 2024 14:47:52 +0800 Subject: [PATCH 4/7] =?UTF-8?q?opt=EF=BC=9A=E5=8E=8B=E7=BC=A9=E8=AE=A2?= =?UTF-8?q?=E5=8D=95=E4=B8=BB=E8=A1=A8=E6=9B=B4=E6=96=B0=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 30 ++++++++++--------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 277ad96b..661848b9 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -170,22 +170,24 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); - - // step3、触发订单主表更新事件（异步） - await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, - new TrackChangeAfterUpdateOrderEvent - { - OrderIds = new[] { track.OrderId }, - IsCompleted = @event.TrackRefreshData.IsCompleted, - TrackCode = track.TrackCode, - TrackContent = track.TrackContent, - TrackEnContent = track.TrackEnContent, - TrackPlace = track.TrackPlace, - TrackTime = track.TrackTime, - TrackTimeZone = track.TrackTimeZone - }); } + var firstTrack = insertTracks.FirstOrDefault(); + if (firstTrack == null) return; + // step3、触发订单主表更新事件（异步） + await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, + new TrackChangeAfterUpdateOrderEvent + { + OrderIds = orderIds, + IsCompleted = @event.TrackRefreshData.IsCompleted, + TrackCode = firstTrack.TrackCode, + TrackContent = firstTrack.TrackContent, + TrackEnContent = firstTrack.TrackEnContent, + TrackPlace = firstTrack.TrackPlace, + TrackTime = firstTrack.TrackTime, + TrackTimeZone = firstTrack.TrackTimeZone + }); + await _trackDetailsRepository.AddManyAsync(insertTracks); } } \ No newline at end of file From 932ca3d1c553647c6cdc530872bc86ab25f1f2e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 15 Apr 2024 19:02:01 +0800 Subject: [PATCH 5/7] opt --- .../TrackChangeAfterUpdateOrderEvent.cs | 2 +- .../Events/TrackPushEvent.cs | 5 ++++ .../EventHandlers/TrackPushEventHandler.cs | 19 ++++++++++-- .../EventHandlers/TrackUpsertEventHandler.cs | 10 +++---- .../ProvideServices/TrackPushService.cs | 14 ++++----- .../ITrackDetailsRepository.cs | 2 -- src/SJZY.Track.Domain/Queries/IBasicQuery.cs | 3 ++ .../Queries/BasicQuery.cs | 6 ++++ .../Repositories/TrackDetailsRepository.cs | 29 ------------------- 9 files changed, 43 insertions(+), 47 deletions(-) diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs index 544c98b1..47a507d2 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeAfterUpdateOrderEvent.cs @@ -5,7 +5,7 @@ public class TrackChangeAfterUpdateOrderEvent /// /// 订单Id /// - public int[] OrderIds { get; set; } + public int[] OrderIds { get; set; } = null!; /// /// 是否删除引起的更新 diff --git a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs index 9d98dcf1..43156556 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs @@ -2,6 +2,11 @@ public class TrackPushEvent { + /// + /// 重试次数 + /// + public int RetryNum { get; set; } + /// /// 操作类型 /// diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 65e60991..1334e3ce 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -20,15 +20,17 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; private readonly IOrderRepository _orderRepository; private readonly ITrackPushCacheService _trackPushCacheService; + private readonly ICapPublisher _capPublisher; public TrackPushEventHandler(IHttpClientProxy httpClientProxy, ITrackPushRecordsRepository trackPushRecordsRepository, - IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService) + IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher) { _httpClientProxy = httpClientProxy; _trackPushRecordsRepository = trackPushRecordsRepository; _orderRepository = orderRepository; _trackPushCacheService = trackPushCacheService; + _capPublisher = capPublisher; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] @@ -103,6 +105,19 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe throw new ArgumentOutOfRangeException(); } + if (!log.IsSuccess) + { + if (@event.RetryNum < 3) + { + // 1, 2, 4min + await _capPublisher.PublishDelayAsync( + TimeSpan.FromSeconds((1 << @event.RetryNum) * 60), + GlobalConst.TrackPushSubscribe, @event); + @event.RetryNum++; + return; + } + } + // 往es 写推送日志 await _trackPushRecordsRepository.AddAsync(log); @@ -169,7 +184,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe Extra = config.ExpandFieldsValue }; - var body = new TrackPushInput() + var body = new TrackPushInput { OrderNo = @event.OrderNo, }; diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 661848b9..8059fcf6 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -105,7 +105,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (track != null && @event.IsAdd!.Value) return; // 存在不让创建 } - // step1、upsert es轨迹 if (track == default) { track = new TrackDetailsIndex @@ -159,8 +158,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var trackPushEvent = new TrackPushEvent { OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改, - OrderId = track.OrderId, - OrderNo = track.OrderNo, + OrderId = order.OrderId, TrackStatus = track.TrackStatus, TrackCode = track.TrackCode, TrackContent = track.TrackContent, @@ -174,6 +172,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var firstTrack = insertTracks.FirstOrDefault(); if (firstTrack == null) return; + + // step1、upsert es轨迹 + await _trackDetailsRepository.AddManyAsync(insertTracks); + // step3、触发订单主表更新事件（异步） await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent @@ -187,7 +189,5 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackTime = firstTrack.TrackTime, TrackTimeZone = firstTrack.TrackTimeZone }); - - await _trackDetailsRepository.AddManyAsync(insertTracks); } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs index ac2d5277..4a0ef260 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs @@ -43,18 +43,16 @@ public class TrackPushService : ApplicationService { input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空 - var orders = await _orderRepository.GetListAsync(x => - x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new - { - x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId - }); + var orderIds = await _orderRepository.GetArrayAsync(x => + x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id); - if (orders.Count == 0) + if (!orderIds.Any()) { - throw new BusinessException(message: "一个推送配置都不存在"); + throw new BusinessException(message: "一个订单都不存在"); } - var tracks = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, input.TrackCode); + var tracks = await _trackDetailsRepository.GetListAsync(orderIds, input.TrackCode); + tracks = tracks.DistinctBy(x => x.OrderId).ToList(); if (tracks.Count == 0) { diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index d32ed593..b35145a4 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -19,8 +19,6 @@ public interface ITrackDetailsRepository Task> GetListAsync(int orderType, List orderNos); - Task> GetListAsync(int orderType, List orderNos, string trackCode); - Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input); Task> GetListAsync(int orderId); diff --git a/src/SJZY.Track.Domain/Queries/IBasicQuery.cs b/src/SJZY.Track.Domain/Queries/IBasicQuery.cs index 460e58be..d2bddef3 100644 --- a/src/SJZY.Track.Domain/Queries/IBasicQuery.cs +++ b/src/SJZY.Track.Domain/Queries/IBasicQuery.cs @@ -22,6 +22,9 @@ public interface IBasicQuery where T : IEntity Task> GetListAsync(Expression> whereExpression, Expression> resultExpression); + Task GetArrayAsync(Expression> whereExpression, + Expression> resultExpression); + /// /// count /// diff --git a/src/SJZY.Track.Repository/Queries/BasicQuery.cs b/src/SJZY.Track.Repository/Queries/BasicQuery.cs index d8003be8..23a3abf1 100644 --- a/src/SJZY.Track.Repository/Queries/BasicQuery.cs +++ b/src/SJZY.Track.Repository/Queries/BasicQuery.cs @@ -24,4 +24,10 @@ public class BasicQuery : SimpleClient, IBasicQuery where T : class, IE { return await Context.Queryable().Where(whereExpression).Select(resultExpression).ToListAsync(); } + + public async Task GetArrayAsync(Expression> whereExpression, + Expression> resultExpression) + { + return await Context.Queryable().Where(whereExpression).Select(resultExpression).ToArrayAsync(); + } } \ 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 d1345e90..61950329 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -104,8 +104,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract tracks) { - if (!tracks.Any()) return; - // 按天写入es var r = await Context.IndexManyAsync(tracks, tracks.First().IndexName); @@ -141,33 +139,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int orderType, List orderNos, string trackCode) - { - if (orderNos.Count < 1) return new List(); - - var data = await Context.SearchAsync(x => x - .Source(s => s.Includes(fd => fd - .Field(f => f.OrderId) - .Field(f => f.OrderNo) - .Field(f => f.TrackStatus) - .Field(f => f.TrackCode) - .Field(f => f.TrackContent) - .Field(f => f.TrackPlace) - .Field(f => f.TrackTime) - .Field(f => f.TrackTimeZone) - )) - .Query(q => - q.Terms(t => t - .Field(f => f.OrderNo.Suffix("keyword")) - .Terms(orderNos)) - && q.Term(f => f.OrderType, orderType) - && q.Term(f => f.TrackCode.Suffix("keyword"), trackCode) - ) - ); - - return data.Documents.ToList(); - } - public async Task> GetListAsync(int orderId) { var data = await Context.SearchAsync(x => x From ad563dda97d4c901e89b1934e043a4c92b2ba281 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 15 Apr 2024 19:14:41 +0800 Subject: [PATCH 6/7] =?UTF-8?q?opt=EF=BC=9A=E5=88=A0=E4=B8=8D=E5=81=9A?= =?UTF-8?q?=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/EventHandlers/CAPFilter.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs index 06bdeca3..20d168f2 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs @@ -20,9 +20,9 @@ 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, GlobalConst.TrackDeleteSubscribe)) + .IsIn(GlobalConst.TrackUpsertSubscribe)) return Task.CompletedTask; // 本地调试不做幂等 @@ -44,7 +44,7 @@ public class CapFilter : SubscribeFilter throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10, - TimeSpan.FromHours(1).Seconds); // 1h内不允许一模一样的参数重复消费 + TimeSpan.FromMinutes(5).Seconds); } } From 64ab6757a99d873be02f79d9309f9e131b4bc88e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Tue, 16 Apr 2024 11:47:21 +0800 Subject: [PATCH 7/7] opt --- .../TrackChangeAfterUpdateOrderEvent.cs | 5 -- .../EventHandlers/CAPFilter.cs | 11 ++-- ...TrackChangeAfterUpdateOrderEventHandler.cs | 54 +++++-------------- .../EventHandlers/TrackDeleteEventHandler.cs | 53 ++++++++++++++---- .../ITrackDetailsRepository.cs | 11 ++-- .../Repositories/TrackDetailsRepository.cs | 37 ++++++++----- 6 files changed, 89 insertions(+), 82 deletions(-) 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.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index b35145a4..f44f298f 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); @@ -22,11 +23,5 @@ public interface ITrackDetailsRepository 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..75ecc85a 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(0) + .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,7 +113,7 @@ 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 +179,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 0; } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input)