From 52b70cb2133e1d7e599d1a711b50c2a869289499 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 09:45:02 +0800 Subject: [PATCH 01/27] =?UTF-8?q?fix=EF=BC=9A=E6=8E=A8=E9=80=81=E6=9B=B4?= =?UTF-8?q?=E6=96=B0attach=20ascii?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 3d7ae977..95a1088e 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -144,9 +144,9 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } } - string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); - if (items.Count == 0) return; + + string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.PushTrackInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); @@ -179,9 +179,11 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } if (items.Count == 0) return; + + string asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.TrackUpdateInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, - attach: OptionType.修改.ToString()); + attach: asciiString); } private TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) From 96b3486956600852f8ed3879ed79a0d750757a5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 10:23:09 +0800 Subject: [PATCH 02/27] =?UTF-8?q?fix=EF=BC=9A=E6=89=8B=E5=8A=A8=E6=8E=A8?= =?UTF-8?q?=E9=80=81=20=E5=AD=90=E7=B3=BB=E7=BB=9F=EF=BC=8C=E5=AE=A2?= =?UTF-8?q?=E6=88=B7=20=E5=88=86=E7=BB=84=E6=8E=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ProvideServices/TrackPushService.cs | 30 ++++++++++++------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs index 8b43e04f..d4bcb72d 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs @@ -39,25 +39,33 @@ public class TrackPushService : ApplicationService { input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空 - var orderIds = await _orderRepository.GetArrayAsync(x => - x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id); + var orders = await _orderRepository.GetArrayAsync(x => + x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new + { + x.Id, x.SubSystemId, x.CustomId + }); - if (!orderIds.Any()) + if (!orders.Any()) { throw new BusinessException(message: "一个订单都不存在"); } - var trackPushEvent = new TrackPushEvent + foreach (var group in orders.GroupBy(x => new { x.SubSystemId, x.CustomId })) { - OptionType = OptionType.新增, - OrderIds = orderIds, - TrackCode = input.TrackCode, - OperationMan = CurrentUserRequest.CurrentUser.Name, - }; + var trackPushEvent = new TrackPushEvent + { + OptionType = OptionType.新增, + SubSystemId = group.Key.SubSystemId, + CustomId = group.Key.CustomId, + OrderIds = group.Select(x => x.Id).ToArray(), + TrackCode = input.TrackCode, + OperationMan = CurrentUserRequest.CurrentUser.Name, + }; - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + } - return $"即将推送轨迹代码{input.TrackCode}给{orderIds.Length}个订单"; + return $"即将推送轨迹代码{input.TrackCode}给{orders.Length}个订单"; } From 4cb5f465da2475bf081d637d46c9521ea02f2ce9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 10:43:39 +0800 Subject: [PATCH 03/27] =?UTF-8?q?opt=EF=BC=9A=E6=8E=A5=E5=8F=A3=E9=80=9A?= =?UTF-8?q?=E8=AF=B7=E6=B1=82=E5=A4=B1=E8=B4=A5=E6=89=93=E5=8D=B0=E5=BC=82?= =?UTF-8?q?=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../TrackPushCompletedEventHandler.cs | 35 ++++++++++--------- .../EventHandlers/TrackPushEventHandler.cs | 21 ++++++++--- 2 files changed, 35 insertions(+), 21 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs index 022c1b36..a1024118 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs @@ -65,8 +65,14 @@ public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscrib private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log) { - var order = new OrderEntity(); - order.Id = log.OrderId; + // tips：只有新增会影响推送进度？ + if (log.OptionType != OptionType.新增.ToString()) return; + + var order = new OrderEntity + { + Id = log.OrderId + }; + /* * 异常：最近一条推送是失败 */ @@ -74,27 +80,22 @@ public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscrib { 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; - if (log.OptionType == OptionType.新增.ToString()) + order.LatestPushTrackCode = log.TrackCode; + order.LatestPushTime = log.CreateTime; + await _orderRepository.UpdateColumnAsync(order, x => new { - // tips：只有新增会影响推送进度？ - order.LatestPushTrackCode = log.TrackCode; - order.LatestPushTime = DateTime.Now; - await _orderRepository.UpdateColumnAsync(order, x => new - { - x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError - }); - return; - } + x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError + }); } - - await _orderRepository.UpdateColumnAsync(order, x => new - { - x.QuestionType, x.LatestPushError - }); } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 95a1088e..67fb60d9 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -44,9 +44,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (pushConfigs == null || pushConfigs.Count == 0) return; _pushConfigs = pushConfigs; - var errorType = (int)TrackProblemType.Error; var orders = await _orderRepository.GetListAsync(x => - @event.OrderIds.Contains(x.Id) && x.QuestionType != errorType, x => new OrderEntity + @event.OrderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode, OrderType = x.OrderType @@ -80,6 +79,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { foreach (var order in _orders) { + if (order.QuestionType == (int)TrackProblemType.Error) continue; // 系统不推异常单，只能人工操作 // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行 config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); if (config == null) continue; @@ -114,6 +114,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (tracks.Count == 0) return; // 一个能推的轨迹都不存在 + // tips：人工推送只要能推就推了，不会校验已推和顺序问题，因为可能是更新 foreach (var track in tracks) { var order = _orders.First(x => x.Id == track.OrderId); @@ -147,9 +148,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (items.Count == 0) return; string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); - await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.PushTrackInterfaceCode, + var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.PushTrackInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); + + if (r.Code != InterfaceRelayConst.SuccessCode) + { + throw new BusinessException(message: r.Msg); + } } private async Task DoUpdateAsync(TrackPushEvent @event) @@ -181,9 +188,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (items.Count == 0) return; string asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); - await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.TrackUpdateInterfaceCode, + var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.TrackUpdateInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); + + if (r.Code != InterfaceRelayConst.SuccessCode) + { + throw new BusinessException(message: r.Msg); + } } private TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) From c6e99193c3590a7837022077de57329306f38011 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 17:00:27 +0800 Subject: [PATCH 04/27] =?UTF-8?q?fix=EF=BC=9A=E8=BD=A8=E8=BF=B9=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=BC=95=E5=8F=91=E7=9A=84=E4=B8=BB=E8=A1=A8=E7=BB=B4?= =?UTF-8?q?=E6=8A=A4=E4=B8=8D=E5=AE=8C=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackDeleteEventHandler.cs | 3 ++- .../Repositories/TrackDetailsRepository.cs | 5 ++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs index 46b87f8b..bd85a916 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs @@ -109,13 +109,14 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe // 删除全部轨迹导致的重置行为 if (track == default) { - if (order.LatestTrackTime == null) return; + if (order.LatestTrackTime == null) continue; // tips：订单是支持从已完成态到未完成态的 order.ResetTrackInfo(); } else { + if (order.LatestTrackContent == track.TrackContent) continue; // 删除引发的回退更新 order.LatestTrackContent = track.TrackContent; order.LatestTrackEnContent = track.TrackEnContent; diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index af29d3fd..be3e4d85 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -236,17 +236,16 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract trackCodes) { var r = await Context.DeleteByQueryAsync(x => x - .Conflicts(Conflicts.Abort) + .Conflicts(Conflicts.Proceed) .Query( q => q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) && q.Term(t => t.IsCanDelete, true) && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) - .WaitForCompletion() ); if (!r.IsValid) - throw new Exception($"轨迹数据删除失败，原因：{r.DebugInformation}"); + throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input) From d8f594f19318a33935a7bfc89a91f6ab126ad3a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 17:22:41 +0800 Subject: [PATCH 05/27] try abort --- .../Repositories/TrackDetailsRepository.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index be3e4d85..97133bfb 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -236,12 +236,13 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract trackCodes) { var r = await Context.DeleteByQueryAsync(x => x - .Conflicts(Conflicts.Proceed) + .Conflicts(Conflicts.Abort) .Query( q => q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) && q.Term(t => t.IsCanDelete, true) && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) + .WaitForCompletion() ); if (!r.IsValid) From 9d3ef9e4a3cf7c6a730f145b2c62f11e2051af03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 18:21:02 +0800 Subject: [PATCH 06/27] =?UTF-8?q?change=EF=BC=9A=E8=BD=A8=E8=BF=B9?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E9=99=8D=E7=BA=A7=E4=B8=BA=E6=BB=9A=E5=8A=A8?= =?UTF-8?q?=E5=88=A0=E9=99=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ITrackDetailsRepository.cs | 1 - .../Repositories/TrackDetailsRepository.cs | 49 ++++++++++++++----- 2 files changed, 38 insertions(+), 12 deletions(-) diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index 0d06b2fd..cc4c033a 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -8,7 +8,6 @@ public interface ITrackDetailsRepository Task> GetLatestListAsync(int[] eventOrderIds, List trackCodes); - Task AnyAsync(int orderId, DateTime trackTime); Task AnyAsync(int orderId, string originTrackData); diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index 97133bfb..c48a0ec0 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -1,4 +1,3 @@ -using Elasticsearch.Net; using Nest; using SJZY.Expand.ElasticSearch.Services; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -235,18 +234,46 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract trackCodes) { - var r = await Context.DeleteByQueryAsync(x => x - .Conflicts(Conflicts.Abort) - .Query( - q => - q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) - && q.Term(t => t.IsCanDelete, true) - && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) - .WaitForCompletion() + var search = await Context.SearchAsync(s => s + .Source(fd => fd.Includes(f => f.Field("indexName"))) + .Size(1000) + .Query(q => q + .Bool(b => b + .Must( + mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), + mu => mu.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) + ) + ) + ) + .Scroll("1m") // 设置滚动时间 ); - if (!r.IsValid) - throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); + + while (search.IsValid && search.Documents.Any()) + { + // 处理滚动查询结果 + var tracks = search.ToIdList(); + + foreach (var track in tracks) + { + var r = await Context.DeleteAsync(track.Id, u => u + .Index(track.IndexName) + ); + + if (!r.IsValid) + throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); + } + + var scrollId = search.ScrollId; // 更新滚动ID + + // 滚动查询 + search = await Context.ScrollAsync("1m", scrollId); + } + + if (!search.IsValid) + { + throw new Exception($"轨迹查询异常，原因：{search.DebugInformation}。"); + } } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input) From 399106ec0a17887a52d80b05535cc1d11149ba47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 18:27:59 +0800 Subject: [PATCH 07/27] =?UTF-8?q?retry=EF=BC=9A=E6=BB=9A=E5=8A=A8=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E5=8F=AA=E8=83=BD500=EF=BC=8C=E5=9B=A0?= =?UTF-8?q?=E6=AD=A4=E5=9B=9E=E9=80=80=E5=88=B0=E6=99=AE=E9=80=9A=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Repositories/TrackDetailsRepository.cs | 53 ++++++++----------- 1 file changed, 21 insertions(+), 32 deletions(-) diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index c48a0ec0..74666a61 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -234,46 +234,35 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract trackCodes) { - var search = await Context.SearchAsync(s => s - .Source(fd => fd.Includes(f => f.Field("indexName"))) - .Size(1000) - .Query(q => q - .Bool(b => b - .Must( - mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), - mu => mu.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) - ) + var search = await Context.SearchAsync(x => x + .Source(s => s.Includes(fd => fd.Field(f => f.IndexName))) + .Size(1000) + .Query(q => q + .Bool(b => b + .Must( + mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), + mu => mu.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) ) ) - .Scroll("1m") // 设置滚动时间 + ) ); - - while (search.IsValid && search.Documents.Any()) - { - // 处理滚动查询结果 - var tracks = search.ToIdList(); - - foreach (var track in tracks) - { - var r = await Context.DeleteAsync(track.Id, u => u - .Index(track.IndexName) - ); - - if (!r.IsValid) - throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); - } - - var scrollId = search.ScrollId; // 更新滚动ID - - // 滚动查询 - search = await Context.ScrollAsync("1m", scrollId); - } - if (!search.IsValid) { throw new Exception($"轨迹查询异常，原因：{search.DebugInformation}。"); } + + var tracks = search.ToIdList(); + + foreach (var track in tracks) + { + var r = await Context.DeleteAsync(track.Id, u => u + .Index(track.IndexName) + ); + + if (!r.IsValid) + throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); + } } public async Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input) From d075c02d5e1c8a4f0bfa2ed07b12c47ad94726b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 18:58:38 +0800 Subject: [PATCH 08/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E5=AF=B9=E6=8E=A5=E6=8E=A5=E5=8F=A3=E9=80=9A=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 67fb60d9..3b42b3db 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -155,7 +155,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != InterfaceRelayConst.SuccessCode) { - throw new BusinessException(message: r.Msg); + throw new BusinessException(message: $"requestId: {r.Data}; {r.Msg}"); } } @@ -195,7 +195,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != InterfaceRelayConst.SuccessCode) { - throw new BusinessException(message: r.Msg); + throw new BusinessException(message: $"requestId: {r.Data}; {r.Msg}"); } } From 62ba69aed92dc89e04401660aa8b82d53aa1adef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 19:07:55 +0800 Subject: [PATCH 09/27] attach --- .../Repositories/TrackDetailsRepository.cs | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index 74666a61..d2b2e3cc 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -236,14 +236,10 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(x => x .Source(s => s.Includes(fd => fd.Field(f => f.IndexName))) - .Size(1000) - .Query(q => q - .Bool(b => b - .Must( - mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)), - mu => mu.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) - ) - ) + .Size(orderIds.Length * trackCodes.Count * 2) + .Query(q => + q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) + && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)) ) ); From 148434dda016263cfb380a6f9377047ae316f33b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 25 Apr 2024 20:23:12 +0800 Subject: [PATCH 10/27] =?UTF-8?q?fix=EF=BC=9A=E5=88=A0=E9=99=A4=E4=B8=8D?= =?UTF-8?q?=E5=AD=98=E5=9C=A8=E6=96=87=E6=A1=A3response.IsValid=E4=B8=BAfa?= =?UTF-8?q?lse=E5=BC=95=E8=B5=B7=E7=9A=84=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Repositories/TrackDetailsRepository.cs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index d2b2e3cc..e4c08278 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -252,12 +252,9 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(track.Id, u => u + await Context.DeleteAsync(track.Id, u => u .Index(track.IndexName) ); - - if (!r.IsValid) - throw new Exception($"轨迹数据删除失败，原因：{r.ServerError.Error}"); } } From de7d1daf81715e4982280ae9f486758b69b699b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 10:51:38 +0800 Subject: [PATCH 11/27] =?UTF-8?q?change=EF=BC=9A=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E4=B8=8D=E5=81=9A=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/CAPFilter.cs | 64 +++---------------- 1 file changed, 9 insertions(+), 55 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs index 4c0014c6..1ce7c51b 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs @@ -1,5 +1,4 @@ -using System.Globalization; -using DotNetCore.CAP.Filter; +using DotNetCore.CAP.Filter; using SJZY.Expand.ABP.Core; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; @@ -20,6 +19,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.ContainsKey("cap-exception")) return Task.CompletedTask; + // 轨迹增改要做幂等 if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe)) return Task.CompletedTask; @@ -31,18 +33,19 @@ public class CapFilter : SubscribeFilter } var argumentJsonStr = context.Arguments.ToJson(); - context.DeliverMessage.Headers["sjzy-arguments"] = argumentJsonStr; + var hash = argumentJsonStr.GetHash(); + + context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; if (!string.IsNullOrEmpty(argumentJsonStr)) { - var hash = argumentJsonStr.GetHash(); using (RedisHelper.Client.Lock("Filter:" + hash, 10)) { - var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExcuteFilter:{hash}")); + var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}")); if (isSkip) throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); - RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10, + RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, TimeSpan.FromMinutes(5).Seconds); } } @@ -50,24 +53,6 @@ public class CapFilter : SubscribeFilter return Task.CompletedTask; } - /// - /// 执行后 - /// - /// - /// - public override Task OnSubscribeExecutedAsync(ExecutedContext context) - { - // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); - // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); - // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); - // context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); - // - // BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!); - - return Task.CompletedTask; - } - - /// /// 消费异常事件 /// @@ -85,37 +70,6 @@ public class CapFilter : SubscribeFilter else LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}"); - - // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); - // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); - // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); - // - // BuildLog(receiveTime!, receiveName!, arguments, true, messageId!); - return Task.CompletedTask; } - - - private void BuildLog(string receiveTime, string receiveName, string? arguments, bool isFromExecptionFliter, - string messageId) - { - var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture); - var useTime = DateTime.Now - parsedDate; - - if (useTime.Minutes > 10) - { - var log = $@" -|-消息编号：{messageId} -|-消费者名：{receiveName} -|-跳过执行：{(isFromExecptionFliter ? "是" : "否")} -|-执行时间：{receiveTime} -|-消费耗时：{useTime.Seconds}秒 -|-请求参数：{arguments}"; - LogHelper.Error(log); - } - else - { - //LogHelper.Info(log); - } - } } \ No newline at end of file From 78c0aa11918b8a52a2d0dcf93495cea4a6417ea0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 10:53:52 +0800 Subject: [PATCH 12/27] =?UTF-8?q?change=EF=BC=9A=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E5=88=86=E7=BB=84=E5=8F=AF=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/SjzyTrackApplicationModule.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 4e8f5fff..1fa422f6 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -92,7 +92,7 @@ namespace SJZY.Track.Application options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); options.ConsumerThreadCount = configuration.GetValue("CAP:ConsumerThreadCount"); options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); - options.UseDispatchingPerGroup = true; + options.UseDispatchingPerGroup = configuration.GetValue("CAP:UseDispatchingPerGroup"); options.EnableConsumerPrefetch = true; options.UseDashboard(dashboardOptions => From 49eda2a3aa7b553a1ecb09595b98a0fcc2c8c123 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 11:28:55 +0800 Subject: [PATCH 13/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E8=BD=A8?= =?UTF-8?q?=E8=BF=B9=E6=96=B0=E5=A2=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 6 ++++-- .../Dtos/TrackDetails/MatchTrackContentDto.cs | 9 +++++++-- src/SJZY.Track.Domain/Queries/IProvideQueries.cs | 4 ++-- src/SJZY.Track.Repository/Queries/ProvideQueries.cs | 7 ++++--- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 937c0a21..8817a331 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -106,6 +106,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode); + var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray(); + var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode); + TrackDetailsIndex? firstTrack = null; var insertTracks = new List(); @@ -136,9 +139,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe CreateTime = DateTime.Now }; - //todo：可以优化? // 根据轨迹code在订单模板内容中寻找匹配的 - var trackContent = await _provideQueries.MatchTrackContentAsync(order.OrderId, @event.TrackCode); + var trackContent = trackContents.FirstOrDefault(x => x.OrderId == order.OrderId); if (trackContent == default) { // 订单所绑定的模板，没有一条内容可以匹配发生的轨迹 diff --git a/src/SJZY.Track.Domain/Dtos/TrackDetails/MatchTrackContentDto.cs b/src/SJZY.Track.Domain/Dtos/TrackDetails/MatchTrackContentDto.cs index 58429cdb..23c27bfe 100644 --- a/src/SJZY.Track.Domain/Dtos/TrackDetails/MatchTrackContentDto.cs +++ b/src/SJZY.Track.Domain/Dtos/TrackDetails/MatchTrackContentDto.cs @@ -3,9 +3,14 @@ public class MatchTrackContentDto { /// - /// 轨迹内容id + /// 订单Id /// - public string TrackCode { get; set; } + public int OrderId { get; set; } + + /// + /// 轨迹内容code + /// + public string TrackCode { get; set; } = null!; /// /// 轨迹内容 diff --git a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs index 7d043a9f..ab50db31 100644 --- a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs +++ b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs @@ -27,10 +27,10 @@ public interface IProvideQueries /// /// 根据订单id找到订单模板，匹配模板命中的轨迹内容 /// - /// 订单id + /// 订单ids /// 轨迹代码 /// - Task MatchTrackContentAsync(int orderId, string code); + Task> MatchTrackContentAsync(int[] orderIds, string code); /// /// 获取可删除的订单轨迹代码 diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 82e4993f..c2013762 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -50,22 +50,23 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, .ToListAsync(); } - public async Task MatchTrackContentAsync(int orderId, string code) + public async Task> MatchTrackContentAsync(int[] orderIds, string code) { return await Context.Queryable() .LeftJoin((x, a) => x.TrackTemplateId == a.TrackTemplateId) .LeftJoin((x, a, b) => a.TrackContentId == b.Id) .LeftJoin((x, a, b, c) => b.TrackStatusId == c.Id) - .Where((x, a, b) => x.OrderId == orderId && b.Code == code) + .Where((x, a, b) => orderIds.Contains(x.OrderId) && b.Code == code) .Select((x, a, b, c) => new MatchTrackContentDto { + OrderId = x.OrderId, TrackCode = b.Code, Content = b.Content, EnContent = b.EnContent, Status = c.Status, IsCanDelete = a.IsCanDelete }) - .FirstAsync(); + .ToListAsync(); } public async Task> GetCanDeleteTrackCodesAsync(List orderIds, From 5cda1270fbe6602c42bd4ac2af2e120e622019c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 11:45:44 +0800 Subject: [PATCH 14/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=E8=BD=A8=E8=BF=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 2 +- .../ITrackDetailsRepository.cs | 2 +- .../Repositories/TrackDetailsRepository.cs | 46 ++++++++++++++----- 3 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index 8817a331..edeadec2 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -104,7 +104,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe 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); + var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode, true); var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray(); var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode); diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index cc4c033a..4296713a 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -4,7 +4,7 @@ namespace SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; public interface ITrackDetailsRepository { - Task> GetListAsync(int[] orderIds, string trackCode); + Task> GetListAsync(int[] orderIds, string trackCode, bool onlyIndexName = false); Task> GetLatestListAsync(int[] eventOrderIds, List trackCodes); diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index e4c08278..ad8bba7d 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -13,20 +13,35 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int[] orderIds, string trackCode) + public async Task> GetListAsync(int[] orderIds, string trackCode, + bool onlyIndexName = false) { - var r = await Context.SearchAsync(s => s - .Source() - .Size(1000) - .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)) + var r = !onlyIndexName + ? await Context.SearchAsync(s => s + .Source() + .Size(1000) + .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)) + ) ) ) ) - ); + : await Context.SearchAsync(s => s + .Source(s2 => s2.Includes(fd => fd + .Field(f => f.IndexName))) + .Size(1000) + .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)) + ) + ) + ) + ); if (!r.IsValid) { @@ -144,9 +159,16 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(track.Id, u => u + var r = await Context.UpdateAsync(track.Id, u => u .Index(track.IndexName) - .Doc(track) + .Doc(new + { + track.TrackPlace, + track.TrackTime, + track.TrackTimeZone, + track.UpdateUser, + UpdateTime = DateTime.Now, + }) ); if (!r.IsValid) From 761446c06933c4f1bf1de25738b289831ec96e83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 12:01:02 +0800 Subject: [PATCH 15/27] attach --- .../TrackDetailsAggregate/ITrackDetailsRepository.cs | 2 +- src/SJZY.Track.Repository/Queries/ProvideQueries.cs | 1 + .../Repositories/TrackDetailsRepository.cs | 7 ++++--- 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index 4296713a..4e4a0b9c 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -4,7 +4,7 @@ namespace SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; public interface ITrackDetailsRepository { - Task> GetListAsync(int[] orderIds, string trackCode, bool onlyIndexName = false); + Task> GetListAsync(int[] orderIds, string trackCode, bool forUpsert = false); Task> GetLatestListAsync(int[] eventOrderIds, List trackCodes); diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index c2013762..7e80e4e8 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -52,6 +52,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, public async Task> MatchTrackContentAsync(int[] orderIds, string code) { + if (orderIds.Length == 0) return new List(); return await Context.Queryable() .LeftJoin((x, a) => x.TrackTemplateId == a.TrackTemplateId) .LeftJoin((x, a, b) => a.TrackContentId == b.Id) diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index ad8bba7d..64a426d2 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -14,9 +14,9 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int[] orderIds, string trackCode, - bool onlyIndexName = false) + bool forUpsert = false) { - var r = !onlyIndexName + var r = !forUpsert ? await Context.SearchAsync(s => s .Source() .Size(1000) @@ -31,7 +31,8 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(s => s .Source(s2 => s2.Includes(fd => fd - .Field(f => f.IndexName))) + .Field(f => f.IndexName) + .Field(f => f.OrderId))) .Size(1000) .Query(q => q .Bool(b => b From eff09f1cb008c6e7a9fec82170e433ab930de0b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 14:59:36 +0800 Subject: [PATCH 16/27] =?UTF-8?q?upg=EF=BC=9A=E5=8D=87=E7=BA=A7=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E9=80=9A7.2.1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 4 ++-- src/SJZY.Track.Application/SJZY.Track.Application.csproj | 2 +- src/SJZY.Track.Application/SjzyTrackApplicationModule.cs | 5 ++--- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 3b42b3db..5edf8f4a 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -155,7 +155,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != InterfaceRelayConst.SuccessCode) { - throw new BusinessException(message: $"requestId: {r.Data}; {r.Msg}"); + throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}"); } } @@ -195,7 +195,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != InterfaceRelayConst.SuccessCode) { - throw new BusinessException(message: $"requestId: {r.Data}; {r.Msg}"); + throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}"); } } diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index fba5f4d2..f9d34258 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -12,7 +12,7 @@ - + diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 1fa422f6..ff8ecefe 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -1,5 +1,4 @@ -using System.Text.Json; -using DotNetCore.CAP; +using DotNetCore.CAP; using Microsoft.Extensions.Configuration; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; @@ -109,7 +108,7 @@ namespace SJZY.Track.Application |-CAP失败回调事件： |-处理类型：{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")} |-头部信息：{faileCalback.Message.Headers.ToJson()} - |-消息内容：{(faileCalback.Message.Value is string ? faileCalback.Message.Value : ((JsonElement?)faileCalback.Message.Value)?.GetRawText())}"); + |-消息内容：{(faileCalback.Message.Value is string ? faileCalback.Message.Value : faileCalback.Message.Value?.ToJson())}"); } catch (Exception ex) { From 56f8d95e48db1f5769ca07dde46e123725c416ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 15:45:59 +0800 Subject: [PATCH 17/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96upsert?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackUpsertEventHandler.cs | 24 +++++++++---------- .../Dtos/TrackContents/TrackContentVO.cs | 19 +++++++++++++++ .../Queries/IProvideQueries.cs | 3 +++ .../Queries/ProvideQueries.cs | 15 ++++++++++++ .../Repositories/TrackDetailsRepository.cs | 3 ++- 5 files changed, 51 insertions(+), 13 deletions(-) create mode 100644 src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index edeadec2..201c4efc 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -109,8 +109,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray(); var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode); - TrackDetailsIndex? firstTrack = null; - var insertTracks = new List(); foreach (var order in @event.OrderIdentifiers) @@ -168,30 +166,32 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe } canNextOrderIds.Add(order.OrderId); - - firstTrack ??= track; } + if (canNextOrderIds.Count == 0) return; + + var content = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode); + // 批量add to es await _trackDetailsRepository.AddManyAsync(insertTracks); - if (firstTrack == null) return; - // 触发订单主表更新事件（异步） await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { - OrderIds = orderIds, + OrderIds = canNextOrderIds.ToArray(), IsCompleted = @event.TrackRefreshData.IsCompleted, - TrackContent = firstTrack.TrackContent, - TrackEnContent = firstTrack.TrackEnContent, + TrackContent = content.TrackContent, + TrackEnContent = content.TrackEnContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone }); // 触发轨迹推送事件（异步） - foreach (var group in @event.OrderIdentifiers.GroupBy(x => x.CustomId)) + foreach (var group in @event.OrderIdentifiers + .Where(x => x.OrderId.IsIn(canNextOrderIds)) + .GroupBy(x => x.CustomId)) { var trackPushEvent = new TrackPushEvent { @@ -202,8 +202,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackCode = @event.TrackCode, SystemPushInfo = new SystemPushStruct { - TrackStatus = firstTrack.TrackStatus, - TrackContent = firstTrack.TrackContent, + TrackStatus = content.TrackStatus, + TrackContent = content.TrackContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone diff --git a/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs new file mode 100644 index 00000000..d44e75c8 --- /dev/null +++ b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs @@ -0,0 +1,19 @@ +namespace SJZY.Track.Domain.Dtos.TrackContents; + +public class TrackContentVO +{ + /// + /// 轨迹状态 + /// + public string TrackStatus { get; set; } = null!; + + /// + /// 轨迹内容 + /// + public string TrackContent { get; set; } = null!; + + /// + /// 轨迹内容英文 + /// + public string TrackEnContent { get; set; } = null!; +} \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs index ab50db31..34e99ae9 100644 --- a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs +++ b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs @@ -1,4 +1,5 @@ using SJZY.Track.Domain.Dtos.Order; +using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Dtos.TrackDetails; using SJZY.Track.Domain.Dtos.TrackTemplate; @@ -65,4 +66,6 @@ public interface IProvideQueries /// Task> FindOrderWithServiceProviderAsync(int serviceProvider, int pageIndex, int pageSize = 1000); + + Task GetTrackContentAsync(int subSystemId, string trackCode); } \ No newline at end of file diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 7e80e4e8..f2a9f25c 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -3,6 +3,7 @@ using SJZY.Track.Domain.Aggregates.TrackContentAggregate; using SJZY.Track.Domain.Aggregates.TrackStatusAggregate; using SJZY.Track.Domain.Aggregates.TrackTemplateAggregate; using SJZY.Track.Domain.Dtos.Order; +using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Dtos.TrackDetails; using SJZY.Track.Domain.Dtos.TrackTemplate; using SJZY.Track.Domain.Queries; @@ -135,4 +136,18 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, .Take(pageSize) .ToListAsync(); } + + public async Task GetTrackContentAsync(int subSystemId, string trackCode) + { + return await Context.Queryable() + .LeftJoin((x, y) => x.TrackStatusId == y.Id) + .Where(x => x.SubSystemId == subSystemId && x.Code == trackCode) + .Select((x, y) => new TrackContentVO + { + TrackStatus = y.Status, + TrackContent = x.Content, + TrackEnContent = x.EnContent + }) + .FirstAsync(); + } } \ 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 64a426d2..4f798cb2 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -32,7 +32,8 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(s => s .Source(s2 => s2.Includes(fd => fd .Field(f => f.IndexName) - .Field(f => f.OrderId))) + .Field(f => f.OrderId) + )) .Size(1000) .Query(q => q .Bool(b => b From 5b12642faf9199cee7bf66166aed6c45491af28c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Thu, 9 May 2024 17:49:33 +0800 Subject: [PATCH 18/27] =?UTF-8?q?change=EF=BC=9Acap=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E4=B8=8D=E6=89=93=E6=B6=88=E6=81=AF=E5=86=85=E5=AE=B9=EF=BC=88?= =?UTF-8?q?=E5=8E=BBcap=E7=9C=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/SjzyTrackApplicationModule.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index ff8ecefe..59827ae9 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -107,8 +107,7 @@ namespace SJZY.Track.Application LogHelper.Error($@" |-CAP失败回调事件： |-处理类型：{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")} - |-头部信息：{faileCalback.Message.Headers.ToJson()} - |-消息内容：{(faileCalback.Message.Value is string ? faileCalback.Message.Value : faileCalback.Message.Value?.ToJson())}"); + |-头部信息：{faileCalback.Message.Headers.ToJson()}"); } catch (Exception ex) { From 7482521e7553eacb3cbd6f58350b49e0a8825a21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Sat, 11 May 2024 10:13:00 +0800 Subject: [PATCH 19/27] =?UTF-8?q?opt=EF=BC=9A=E5=8D=87=E7=BA=A7cap?= =?UTF-8?q?=E5=88=B08.1.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/SJZY.Track.Application.csproj | 8 ++++---- src/SJZY.Track.Application/SjzyTrackApplicationModule.cs | 4 +--- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index f9d34258..b2801d4a 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -17,10 +17,10 @@ - - - - + + + + diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 59827ae9..2984ac45 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -89,10 +89,8 @@ namespace SJZY.Track.Application // options.TopicNamePrefix = "v1"; options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); - options.ConsumerThreadCount = configuration.GetValue("CAP:ConsumerThreadCount"); options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); - options.UseDispatchingPerGroup = configuration.GetValue("CAP:UseDispatchingPerGroup"); - options.EnableConsumerPrefetch = true; + options.EnableSubscriberParallelExecute = true; options.UseDashboard(dashboardOptions => { From 4b0acb1978c89bdca231b08ad3da0398f50b8d92 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, 14 May 2024 17:17:14 +0800 Subject: [PATCH 20/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E8=BD=A8?= =?UTF-8?q?=E8=BF=B9=E5=88=B7=E6=96=B0=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E7=AD=89=E5=BE=85=E8=B6=85=E6=97=B6=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ProvideServices/TrackPullService.cs | 68 ++++++++++--------- 1 file changed, 36 insertions(+), 32 deletions(-) diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs index d7f71038..cb7cd5e1 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs @@ -78,46 +78,50 @@ public class TrackPullService : ApplicationService /// [AllowAnonymous] [HttpPost("Track/RefreshForTask")] - public async Task RefreshForTaskAsync() + public bool RefreshForTaskAsync() { - var configs = await _trackPullCacheService.GetAllPullConfigsAsync(); - foreach (var config in configs) + Task.Run(async () => { - var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId); - if (await RedisHelper.Client.ExistsAsync(redisKey)) - { - continue; - } + var configs = await _trackPullCacheService.GetAllPullConfigsAsync(); - var page = 1; - var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page); - while (orders.Any()) + foreach (var config in configs) { - var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; - var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName; - - foreach (var group in orders.GroupBy(x => - new { x.SubSystemId, x.OrderType })) + var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId); + if (await RedisHelper.Client.ExistsAsync(redisKey)) { - var data = group.Select(x => new - { - x.OrderNo - }).ToList(); - - var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, - serviceProviderName!, - group.Key.SubSystemId, group.Key.OrderType)); - // 批量获取轨迹 - await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( - InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, - GlobalConst.TrackRefreshSubscribe, attach); + continue; } - orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page); - } + var page = 1; + var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page); + while (orders.Any()) + { + var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; + var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName; - await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count); - } + foreach (var group in orders.GroupBy(x => + new { x.SubSystemId, x.OrderType })) + { + var data = group.Select(x => new + { + x.OrderNo + }).ToList(); + + var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, + serviceProviderName!, + group.Key.SubSystemId, group.Key.OrderType)); + // 批量获取轨迹 + await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, + GlobalConst.TrackRefreshSubscribe, attach); + } + + orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page); + } + + await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count); + } + }); return true; } From 505d8e7dc1ed16e2ec52a38046869fd26ec9bb95 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, 14 May 2024 17:17:37 +0800 Subject: [PATCH 21/27] =?UTF-8?q?add=EF=BC=9A=E6=B7=BB=E5=8A=A0cap?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E5=86=85=E5=AD=98debug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/SJZY.Track.Application.csproj | 2 ++ src/SJZY.Track.Application/SjzyTrackApplicationModule.cs | 7 ++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index b2801d4a..982969a4 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -21,6 +21,8 @@ + + diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 2984ac45..fa522c48 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -1,5 +1,6 @@ using DotNetCore.CAP; using Microsoft.Extensions.Configuration; +using Savorboard.CAP.InMemoryMessageQueue; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; @@ -76,6 +77,10 @@ namespace SJZY.Track.Application context.Services.AddCapTableNameService(); context.Services.AddCap(options => { +#if DEBUG + options.UseInMemoryStorage(); + options.UseInMemoryMessageQueue(); +#else options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!); options.UseRabbitMQ(configure => { @@ -86,7 +91,7 @@ namespace SJZY.Track.Application configure.BasicQosOptions = new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount")); }); - // options.TopicNamePrefix = "v1"; +#endif options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); From b5f7f866646fb0f64be1d62c24ed34a77e3e96e0 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, 27 May 2024 11:51:34 +0800 Subject: [PATCH 22/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E5=B9=82=E7=AD=89=EF=BC=8C=E4=BC=98=E5=8C=96cap?= =?UTF-8?q?=E9=87=8D=E8=AF=95=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Api/appsettings.json | 2 +- .../{EventHandlers => CAP}/CAPFilter.cs | 54 ++-- ...TrackChangeAfterUpdateOrderEventHandler.cs | 6 +- .../EventHandlers/TrackDeleteEventHandler.cs | 4 +- .../TrackInsertForNotMappedEventHandler.cs | 2 +- .../TrackPushCompletedEventHandler.cs | 2 +- .../EventHandlers/TrackPushEventHandler.cs | 3 +- .../EventHandlers/TrackRefreshEventHandler.cs | 2 +- .../EventHandlers/TrackUpsertEventHandler.cs | 2 +- .../CAP/LoggerExtensions.cs | 89 ++++++ .../CAP/MySubscribeExecutor.cs | 268 ++++++++++++++++++ .../CAP/SubscriberExecutionFailedException.cs | 15 + .../ProvideServices/TrackPushService.cs | 2 +- .../SjzyTrackApplicationModule.cs | 4 +- .../Consts/GlobalConst.MQ.cs | 5 + 15 files changed, 431 insertions(+), 29 deletions(-) rename src/SJZY.Track.Application/{EventHandlers => CAP}/CAPFilter.cs (59%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs (92%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackDeleteEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackInsertForNotMappedEventHandler.cs (97%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackPushCompletedEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackPushEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackRefreshEventHandler.cs (99%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackUpsertEventHandler.cs (99%) create mode 100644 src/SJZY.Track.Application/CAP/LoggerExtensions.cs create mode 100644 src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs create mode 100644 src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs diff --git a/src/SJZY.Track.Api/appsettings.json b/src/SJZY.Track.Api/appsettings.json index 676973d6..d69cb0c1 100644 --- a/src/SJZY.Track.Api/appsettings.json +++ b/src/SJZY.Track.Api/appsettings.json @@ -5,7 +5,7 @@ "Apollo": { "AppId": "SJZY.Track.Configuration", "Env": "DEV", - "Namespace": "Dev", + "Namespace": "Test", "MetaServer": "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080", "ConfigServer": [ "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080" ] } diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs similarity index 59% rename from src/SJZY.Track.Application/EventHandlers/CAPFilter.cs rename to src/SJZY.Track.Application/CAP/CAPFilter.cs index 1ce7c51b..a4e1f841 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/src/SJZY.Track.Application/CAP/CAPFilter.cs @@ -1,10 +1,12 @@ -using DotNetCore.CAP.Filter; +using System.Security.Cryptography; +using System.Text; +using DotNetCore.CAP.Filter; using SJZY.Expand.ABP.Core; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Domain.Consts; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP; public class CapFilter : SubscribeFilter { @@ -17,21 +19,11 @@ public class CapFilter : SubscribeFilter /// public override Task OnSubscribeExecutingAsync(ExecutingContext context) { - context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff"); - - // 异常重试不做幂等 - if (context.DeliverMessage.Headers.ContainsKey("cap-exception")) return Task.CompletedTask; - - // 轨迹增改要做幂等 - if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe)) + // 轨迹新增、更新要做幂等 + if (!context.DeliverMessage.Headers["cap-msg-name"] + .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe)) return Task.CompletedTask; - // 本地调试不做幂等 - if (System.Diagnostics.Debugger.IsAttached) - { - return Task.CompletedTask; - } - var argumentJsonStr = context.Arguments.ToJson(); var hash = argumentJsonStr.GetHash(); @@ -44,15 +36,23 @@ public class CapFilter : SubscribeFilter var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}")); if (isSkip) throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); - - RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, - TimeSpan.FromMinutes(5).Seconds); } } return Task.CompletedTask; } + public override Task OnSubscribeExecutedAsync(ExecutedContext context) + { + context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash); + if (!hash.IsNullOrEmpty()) + { + RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, TimeSpan.FromMinutes(10).Seconds); + } + + return Task.CompletedTask; + } + /// /// 消费异常事件 /// @@ -72,4 +72,22 @@ public class CapFilter : SubscribeFilter return Task.CompletedTask; } + + + /// 获取字符串的hash值 + /// + /// + public static string GetHash(string str) + { + // 将字符串转换为字节 + byte[] data = Encoding.UTF8.GetBytes(str); + + // 创建SHA256实例 + using SHA256 sha256 = SHA256.Create(); + // 计算哈希值 + byte[] hash = sha256.ComputeHash(data); + + // 将哈希值转换为16进制字符串 + return BitConverter.ToString(hash); + } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs similarity index 92% rename from src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index 3e3b1d07..c8f184ae 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs @@ -5,7 +5,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe { @@ -36,6 +36,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa if (orders.Count == 0) return; + var updateOrders = new List(); foreach (var order in orders) { if (order.LatestTrackContent != @event.TrackContent) @@ -58,9 +59,10 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa order.LatestTrackPlace = @event.TrackPlace; order.LatestTrackTime = @event.TrackTime; order.LatestTrackTimeZone = @event.TrackTimeZone; + updateOrders.Add(order); } - await _orderRepository.UpdateRangeAsync(orders, x => new + await _orderRepository.UpdateRangeAsync(updateOrders, x => new { x.IsCompleted, x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime, x.LatestTrackTimeZone diff --git a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index bd85a916..fa2751bf 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -8,7 +8,7 @@ using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe { @@ -29,6 +29,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] public async Task ConsumerAsync(TrackDeleteEvent @event) { + throw new Exception(); + var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); if (dup.Any()) { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackInsertForNotMappedEventHandler.cs similarity index 97% rename from src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackInsertForNotMappedEventHandler.cs index 9bd76c5e..b2f59331 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackInsertForNotMappedEventHandler.cs @@ -4,7 +4,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSubscribe { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushCompletedEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackPushCompletedEventHandler.cs index a1024118..4658aa44 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushCompletedEventHandler.cs @@ -9,7 +9,7 @@ using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscribe { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 5edf8f4a..734246eb 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -14,7 +14,7 @@ using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { @@ -38,6 +38,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] + [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] public async Task ConsumerAsync(TrackPushEvent @event) { var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId); diff --git a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs similarity index 99% rename from src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs index 0e9fd334..f982343e 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs @@ -14,7 +14,7 @@ using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs similarity index 99% rename from src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 201c4efc..46443ffb 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -8,7 +8,7 @@ using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.EventHandlers; +namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { diff --git a/src/SJZY.Track.Application/CAP/LoggerExtensions.cs b/src/SJZY.Track.Application/CAP/LoggerExtensions.cs new file mode 100644 index 00000000..8b430825 --- /dev/null +++ b/src/SJZY.Track.Application/CAP/LoggerExtensions.cs @@ -0,0 +1,89 @@ +using Microsoft.Extensions.Logging; + +namespace SJZY.Track.Application.CAP; + +internal static class LoggerExtensions +{ + public static void ConsumerExecutedAfterThreshold(this ILogger logger, string messageId, int retries) + { + logger.LogWarning( + $"The Subscriber of the message({messageId}) still fails after {retries}th executions and we will stop retrying."); + } + + public static void SenderAfterThreshold(this ILogger logger, string messageId, int retries) + { + logger.LogWarning( + $"The Publisher of the message({messageId}) still fails after {retries}th sends and we will stop retrying."); + } + + public static void ExecutedThresholdCallbackFailed(this ILogger logger, Exception ex) + { + logger.LogWarning(ex, "FailedThresholdCallback action raised an exception:" + ex.Message); + } + + public static void ConsumerDuplicates(this ILogger logger, string subscriber, string group) + { + logger.LogWarning( + $"We detected that you have duplicate subscribers ({subscriber}) in same group ({group}), this will cause diversity behavior."); + } + + public static void ConsumerExecutionRetrying(this ILogger logger, string messageId, int retries) + { + logger.LogWarning($"The {retries}th retrying consume a message failed. message id: {messageId}"); + } + + public static void SenderRetrying(this ILogger logger, string messageId, int retries) + { + logger.LogWarning($"The {retries}th retrying send a message failed. message id: {messageId} "); + } + + public static void MessageReceived(this ILogger logger, string messageId, string name) + { + logger.LogDebug($"Received message. id:{messageId}, name: {name}"); + } + + public static void MessagePublishException(this ILogger logger, string? messageId, string reason, Exception? ex) + { + logger.LogError(ex, + $"An exception occurred while publishing a message, reason:{reason}. message id:{messageId}"); + } + + public static void ConsumerExecuting(this ILogger logger, string className, string methodName, string group) + { + logger.LogInformation($"Executing subscriber method '{className}.{methodName}' on group '{group}'"); + } + + public static void ConsumerExecuted(this ILogger logger, string className, string methodName, string group, + double milliseconds, string? instance) + { + logger.LogInformation( + $"Executed subscriber method '{className}.{methodName}' on group '{group}' with instance '{instance}' in {milliseconds}ms"); + } + + public static void ConsumerExecuteFailed(this ILogger logger, string topic, string id, string? instance, + Exception? ex) + { + logger.LogError(ex, + $"An exception occurred while executing the subscription method. Topic:{topic}, Id:{id}, Instance: {instance}"); + } + + public static void ServerStarting(this ILogger logger) + { + logger.LogInformation("Starting the processing server."); + } + + public static void ProcessorsStartedError(this ILogger logger, Exception ex) + { + logger.LogError(ex, "Starting the processors throw an exception."); + } + + public static void ServerShuttingDown(this ILogger logger) + { + logger.LogInformation("Shutting down the processing server..."); + } + + public static void ExpectedOperationCanceledException(this ILogger logger, Exception ex) + { + logger.LogWarning(ex, $"Expected an OperationCanceledException, but found '{ex.Message}'."); + } +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs b/src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs new file mode 100644 index 00000000..c9ff478b --- /dev/null +++ b/src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs @@ -0,0 +1,268 @@ +using System.Diagnostics; +using System.Reflection; +using DotNetCore.CAP; +using DotNetCore.CAP.Diagnostics; +using DotNetCore.CAP.Internal; +using DotNetCore.CAP.Messages; +using DotNetCore.CAP.Persistence; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace SJZY.Track.Application.CAP; + +public class MySubscribeExecutor : ISubscribeExecutor +{ + // diagnostics listener + // ReSharper disable once InconsistentNaming + private static readonly DiagnosticListener s_diagnosticListener = + new(CapDiagnosticListenerNames.DiagnosticListenerName); + + private readonly IDataStorage _dataStorage; + private readonly string? _hostName; + private readonly ILogger _logger; + private readonly CapOptions _options; + private readonly IServiceProvider _provider; + + public MySubscribeExecutor( + ILogger logger, + IOptions options, + IServiceProvider provider) + { + _provider = provider; + _logger = logger; + _options = options.Value; + + _dataStorage = _provider.GetRequiredService(); + Invoker = _provider.GetRequiredService(); + _hostName = Helper.GetInstanceHostname(); + } + + private ISubscribeInvoker Invoker { get; } + + public async Task ExecuteAsync(MediumMessage message, ConsumerExecutorDescriptor? descriptor = null, + CancellationToken cancellationToken = default) + { + if (descriptor == null) + { + var selector = _provider.GetRequiredService(); + if (!selector.TryGetTopicExecutor(message.Origin.GetName(), message.Origin.GetGroup()!, out descriptor)) + { + var error = + $"Message (Name:{message.Origin.GetName()},Group:{message.Origin.GetGroup()}) can not be found subscriber." + + $"{Environment.NewLine} see: https://github.com/dotnetcore/CAP/issues/63"; + _logger.LogError(error); + + TracingError(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), message.Origin, null, + new Exception(error)); + + var ex = new SubscriberNotFoundException(error); + await SetFailedState(message, ex); + return OperateResult.Failed(ex); + } + } + + bool retry; + OperateResult result; + + //record instance id + message.Origin.Headers[Headers.ExecutionInstanceId] = _hostName; + + do + { + var (shouldRetry, operateResult) = await ExecuteWithoutRetryAsync(message, descriptor, cancellationToken) + .ConfigureAwait(false); + result = operateResult; + if (result.Equals(OperateResult.Success)) return result; + retry = shouldRetry; + } while (retry); + + return result; + } + + private async Task<(bool, OperateResult)> ExecuteWithoutRetryAsync(MediumMessage message, + ConsumerExecutorDescriptor descriptor, CancellationToken cancellationToken) + { + if (message == null) throw new ArgumentNullException(nameof(message)); + + cancellationToken.ThrowIfCancellationRequested(); + + try + { + _logger.ConsumerExecuting(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name, + descriptor.Attribute.Group); + + var sp = Stopwatch.StartNew(); + + await InvokeConsumerMethodAsync(message, descriptor, cancellationToken).ConfigureAwait(false); + + sp.Stop(); + + await SetSuccessfulState(message).ConfigureAwait(false); + + CapEventCounterSource.Log.WriteInvokeTimeMetrics(sp.Elapsed.TotalMilliseconds); + _logger.ConsumerExecuted(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name, + descriptor.Attribute.Group, sp.Elapsed.TotalMilliseconds, message.Origin.GetExecutionInstanceId()); + + return (false, OperateResult.Success); + } + catch (Exception ex) + { + _logger.ConsumerExecuteFailed(message.Origin.GetName(), message.DbId, + message.Origin.GetExecutionInstanceId(), ex); + + return (await SetFailedState(message, ex).ConfigureAwait(false), OperateResult.Failed(ex)); + } + } + + private Task SetSuccessfulState(MediumMessage message) + { + message.ExpiresAt = DateTime.Now.AddSeconds(_options.SucceedMessageExpiredAfter); + + return _dataStorage.ChangeReceiveStateAsync(message, StatusName.Succeeded); + } + + private async Task SetFailedState(MediumMessage message, Exception ex) + { + if (ex is SubscriberNotFoundException) + message.Retries = _options.FailedRetryCount; // not retry if SubscriberNotFoundException + + var needRetry = UpdateMessageForRetry(message); + + message.Origin.AddOrUpdateException(ex); + message.ExpiresAt = message.Added.AddSeconds(_options.FailedMessageExpiredAfter); + + await _dataStorage.ChangeReceiveStateAsync(message, StatusName.Failed).ConfigureAwait(false); + + return needRetry; + } + + private bool UpdateMessageForRetry(MediumMessage message) + { + var retries = ++message.Retries; + + if (retries == _options.FailedRetryCount) + { + try + { + _options.FailedThresholdCallback?.Invoke(new FailedInfo + { + ServiceProvider = _provider, + MessageType = MessageType.Subscribe, + Message = message.Origin + }); + + _logger.ConsumerExecutedAfterThreshold(message.DbId, _options.FailedRetryCount); + } + catch (Exception ex) + { + _logger.ExecutedThresholdCallbackFailed(ex); + } + } + + return false; + } + + private async Task InvokeConsumerMethodAsync(MediumMessage message, ConsumerExecutorDescriptor descriptor, + CancellationToken cancellationToken) + { + var consumerContext = new ConsumerContext(descriptor, message); + var tracingTimestamp = TracingBefore(message.Origin, descriptor.MethodInfo); + try + { + var ret = await Invoker.InvokeAsync(consumerContext, cancellationToken).ConfigureAwait(false); + + TracingAfter(tracingTimestamp, message.Origin, descriptor.MethodInfo); + + if (!string.IsNullOrEmpty(ret.CallbackName)) + { + var header = new Dictionary + { + [Headers.CorrelationId] = message.Origin.GetId(), + [Headers.CorrelationSequence] = (message.Origin.GetCorrelationSequence() + 1).ToString() + }; + + if (message.Origin.Headers.TryGetValue(Headers.TraceParent, out var traceparent)) + header[Headers.TraceParent] = traceparent; + + await _provider.GetRequiredService() + .PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + //ignore + } + catch (Exception ex) + { + var e = new SubscriberExecutionFailedException(ex.Message, ex); + + TracingError(tracingTimestamp, message.Origin, descriptor.MethodInfo, e); + + e.ReThrow(); + } + } + + #region tracing + + private long? TracingBefore(Message message, MethodInfo method) + { + if (s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.BeforeSubscriberInvoke)) + { + var eventData = new CapEventDataSubExecute + { + OperationTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), + Operation = message.GetName(), + Message = message, + MethodInfo = method + }; + + s_diagnosticListener.Write(CapDiagnosticListenerNames.BeforeSubscriberInvoke, eventData); + + return eventData.OperationTimestamp; + } + + return null; + } + + private void TracingAfter(long? tracingTimestamp, Message message, MethodInfo method) + { + CapEventCounterSource.Log.WriteInvokeMetrics(); + if (tracingTimestamp != null && + s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.AfterSubscriberInvoke)) + { + var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + var eventData = new CapEventDataSubExecute + { + OperationTimestamp = now, + Operation = message.GetName(), + Message = message, + MethodInfo = method, + ElapsedTimeMs = now - tracingTimestamp.Value + }; + + s_diagnosticListener.Write(CapDiagnosticListenerNames.AfterSubscriberInvoke, eventData); + } + } + + private void TracingError(long? tracingTimestamp, Message message, MethodInfo? method, Exception ex) + { + if (tracingTimestamp != null && + s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.ErrorSubscriberInvoke)) + { + var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + var eventData = new CapEventDataSubExecute + { + OperationTimestamp = now, + Operation = message.GetName(), + Message = message, + MethodInfo = method, + ElapsedTimeMs = now - tracingTimestamp.Value, + Exception = ex + }; + + s_diagnosticListener.Write(CapDiagnosticListenerNames.ErrorSubscriberInvoke, eventData); + } + } + + #endregion +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs b/src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs new file mode 100644 index 00000000..31b183af --- /dev/null +++ b/src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs @@ -0,0 +1,15 @@ +namespace SJZY.Track.Application.CAP; + +internal class SubscriberExecutionFailedException : Exception +{ + private readonly Exception _originException; + + public SubscriberExecutionFailedException(string message, Exception ex) : base(message, ex) + { + _originException = ex; + } + + public override string? StackTrace => _originException.StackTrace; + + public override string? Source => _originException.Source; +} \ 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 d4bcb72d..6897451e 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs @@ -62,7 +62,7 @@ public class TrackPushService : ApplicationService OperationMan = CurrentUserRequest.CurrentUser.Name, }; - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent); } return $"即将推送轨迹代码{input.TrackCode}给{orders.Length}个订单"; diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index fa522c48..efa5aa41 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -1,12 +1,13 @@ using DotNetCore.CAP; +using DotNetCore.CAP.Internal; using Microsoft.Extensions.Configuration; using Savorboard.CAP.InMemoryMessageQueue; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; +using SJZY.Track.Application.CAP; using SJZY.Track.Application.Contract; -using SJZY.Track.Application.EventHandlers; using SJZY.Track.Application.Localization; using SJZY.UserCenter.Contracts; using SJZY.UserCenter.SharedModel; @@ -74,6 +75,7 @@ namespace SJZY.Track.Application ); //CAP + context.Services.AddSingleton(); context.Services.AddCapTableNameService(); context.Services.AddCap(options => { diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 93fd1c53..378f0b89 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -32,6 +32,11 @@ public partial class GlobalConst /// public const string TrackPushSubscribe = "SJZY.Track.TrackPush"; + /// + /// 轨迹人工手动推送 + /// + public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush"; + /// /// 轨迹推送完成（接口通回调） /// From 33c36bed103e050eaf01a7a7b9858f6b87386319 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, 27 May 2024 16:52:08 +0800 Subject: [PATCH 23/27] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96cap=20filte?= =?UTF-8?q?r?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/CAP/CAPFilter.cs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs index a4e1f841..110395fc 100644 --- a/src/SJZY.Track.Application/CAP/CAPFilter.cs +++ b/src/SJZY.Track.Application/CAP/CAPFilter.cs @@ -19,24 +19,21 @@ public class CapFilter : SubscribeFilter /// public override Task OnSubscribeExecutingAsync(ExecutingContext context) { + if (!context.DeliverMessage.Headers["sjzy-arguments-hash"].IsNullOrEmpty()) return Task.CompletedTask; + // 轨迹新增、更新要做幂等 if (!context.DeliverMessage.Headers["cap-msg-name"] .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe)) return Task.CompletedTask; var argumentJsonStr = context.Arguments.ToJson(); - var hash = argumentJsonStr.GetHash(); - - context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; if (!string.IsNullOrEmpty(argumentJsonStr)) { - using (RedisHelper.Client.Lock("Filter:" + hash, 10)) - { - var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}")); - if (isSkip) - throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); - } + var hash = GetHash(argumentJsonStr); + context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; + if (!string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}"))) + throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); } return Task.CompletedTask; From 5712f567dfbf9d5168f2e9b6a3828baefd0fe1d2 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, 27 May 2024 17:00:39 +0800 Subject: [PATCH 24/27] =?UTF-8?q?opt=EF=BC=9A=E6=8F=90=E5=8F=96=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E9=87=8D=E8=AF=95=E5=86=B7=E5=8D=B4=E7=AA=97=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/CAP/CAPFilter.cs | 35 +++++++++++-------- .../SjzyTrackApplicationModule.cs | 2 ++ 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs index 110395fc..6522599b 100644 --- a/src/SJZY.Track.Application/CAP/CAPFilter.cs +++ b/src/SJZY.Track.Application/CAP/CAPFilter.cs @@ -17,37 +17,40 @@ public class CapFilter : SubscribeFilter /// /// /// - public override Task OnSubscribeExecutingAsync(ExecutingContext context) + public override async Task OnSubscribeExecutingAsync(ExecutingContext context) { - if (!context.DeliverMessage.Headers["sjzy-arguments-hash"].IsNullOrEmpty()) return Task.CompletedTask; - // 轨迹新增、更新要做幂等 if (!context.DeliverMessage.Headers["cap-msg-name"] .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe)) - return Task.CompletedTask; + { + return; + } var argumentJsonStr = context.Arguments.ToJson(); if (!string.IsNullOrEmpty(argumentJsonStr)) { - var hash = GetHash(argumentJsonStr); - context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; - if (!string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}"))) - throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); - } + var hash = context.DeliverMessage.Headers["sjzy-arguments-hash"]; + if (hash.IsNullOrEmpty()) // 计算参数hash + { + hash = GetHash(argumentJsonStr); + context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash; + } - return Task.CompletedTask; + if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}")) + { + throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}"); + } + } } - public override Task OnSubscribeExecutedAsync(ExecutedContext context) + public override async Task OnSubscribeExecutedAsync(ExecutedContext context) { context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash); if (!hash.IsNullOrEmpty()) { - RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, TimeSpan.FromMinutes(10).Seconds); + await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}", 1, TimeSpan.FromMinutes(10).Seconds); } - - return Task.CompletedTask; } /// @@ -65,7 +68,11 @@ public class CapFilter : SubscribeFilter LogHelper.Warn(ex.ToString()); } else + { + // 发飞书，异常继续上抛，进入重试机制 + // tips：重试有冷却期（默认4分钟） LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}"); + } return Task.CompletedTask; } diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index efa5aa41..10394daa 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -97,6 +97,8 @@ namespace SJZY.Track.Application options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); + options.FallbackWindowLookbackSeconds = + configuration.GetValue("CAP:FallbackWindowLookbackSeconds"); options.EnableSubscriberParallelExecute = true; options.UseDashboard(dashboardOptions => From b4bcb377eccef946eba125789469f32c4727e18e 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, 27 May 2024 17:29:51 +0800 Subject: [PATCH 25/27] =?UTF-8?q?fix=EF=BC=9Ahash=20key=E4=B8=8D=E5=AD=98?= =?UTF-8?q?=E5=9C=A8=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Application/CAP/CAPFilter.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/SJZY.Track.Application/CAP/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs index 6522599b..036c1e34 100644 --- a/src/SJZY.Track.Application/CAP/CAPFilter.cs +++ b/src/SJZY.Track.Application/CAP/CAPFilter.cs @@ -30,7 +30,7 @@ public class CapFilter : SubscribeFilter if (!string.IsNullOrEmpty(argumentJsonStr)) { - var hash = context.DeliverMessage.Headers["sjzy-arguments-hash"]; + context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash); if (hash.IsNullOrEmpty()) // 计算参数hash { hash = GetHash(argumentJsonStr); From c63bf09c085e135ff3a80ec56cf2147404869edf 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, 28 May 2024 09:05:39 +0800 Subject: [PATCH 26/27] =?UTF-8?q?=E8=BF=98=E5=8E=9F=E8=B0=83=E8=AF=95?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../CAP/EventHandlers/TrackDeleteEventHandler.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index fa2751bf..ae24d7e3 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -29,8 +29,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] public async Task ConsumerAsync(TrackDeleteEvent @event) { - throw new Exception(); - var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); if (dup.Any()) { From 47ea3b8f82d6efe1acec5af9d904a3b70c7d3946 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, 25 Jun 2024 17:53:56 +0800 Subject: [PATCH 27/27] =?UTF-8?q?token=E8=A7=A3=E6=9E=90=E5=8F=98=E6=9B=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ProvideServices/TrackPushService.cs | 17 ++++++- .../Services/ProvideServices/TrackService.cs | 50 +++++++++++++++++-- 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs index 6897451e..41a91a72 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs @@ -6,6 +6,7 @@ using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Consts; using SJZY.UserCenter.HttpApi.Services; +using Volo.Abp.Http.Client.DynamicProxying; namespace SJZY.Track.Application.Services.ProvideServices; @@ -19,14 +20,17 @@ public class TrackPushService : ApplicationService private readonly IOrderRepository _orderRepository; private readonly ICapPublisher _capPublisher; private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; + private readonly IHttpClientProxy _requestCurrentUserService; public TrackPushService(IOrderRepository orderRepository , ICapPublisher capPublisher - , ITrackPushRecordsRepository trackPushRecordsRepository) + , ITrackPushRecordsRepository trackPushRecordsRepository, + IHttpClientProxy requestCurrentUserService) { _orderRepository = orderRepository; _capPublisher = capPublisher; _trackPushRecordsRepository = trackPushRecordsRepository; + _requestCurrentUserService = requestCurrentUserService; } /// @@ -50,6 +54,15 @@ public class TrackPushService : ApplicationService throw new BusinessException(message: "一个订单都不存在"); } + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != "200" || userInfo.Data is null) + { + //TODO: log || ex + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}"); + } + + var name = userInfo.Data.UserName; + foreach (var group in orders.GroupBy(x => new { x.SubSystemId, x.CustomId })) { var trackPushEvent = new TrackPushEvent @@ -59,7 +72,7 @@ public class TrackPushService : ApplicationService CustomId = group.Key.CustomId, OrderIds = group.Select(x => x.Id).ToArray(), TrackCode = input.TrackCode, - OperationMan = CurrentUserRequest.CurrentUser.Name, + OperationMan = name, }; await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent); diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 6c5d81ba..f3d63ca2 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -7,6 +7,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using SJZY.UserCenter.HttpApi.Services; +using Volo.Abp.Http.Client.DynamicProxying; namespace SJZY.Track.Application.Services.ProvideServices; @@ -20,14 +21,17 @@ public partial class TrackService : ApplicationService, ITrackService private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ICapPublisher _capPublisher; + private readonly IHttpClientProxy _requestCurrentUserService; public TrackService(IOrderRepository orderRepository, IProvideQueries provideQueries, ICapPublisher capPublisher, - ITrackDetailsRepository trackDetailsRepository) + ITrackDetailsRepository trackDetailsRepository, + IHttpClientProxy requestCurrentUserService) { _orderRepository = orderRepository; _provideQueries = provideQueries; _provideQueries = provideQueries; _capPublisher = capPublisher; + _requestCurrentUserService = requestCurrentUserService; _trackDetailsRepository = trackDetailsRepository; } @@ -50,6 +54,15 @@ public partial class TrackService : ApplicationService, ITrackService throw new BusinessException(message: $"订单：{input.OrderNo}不存在"); } + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != "200" || userInfo.Data is null) + { + //TODO: log || ex + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}"); + } + + var name = userInfo.Data.UserName; + // 新增/更新轨迹 var @event = ObjectMapper.Map(input); @event.SubSystemId = order.SubSystemId; @@ -63,7 +76,7 @@ public partial class TrackService : ApplicationService, ITrackService CustomId = order.CustomId } }; - @event.OperationMan = CurrentUserRequest.CurrentUser.Name; + @event.OperationMan = name; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event); return new UnifyResultDto(true, true); } @@ -89,6 +102,15 @@ public partial class TrackService : ApplicationService, ITrackService x.Id, x.OrderNo, x.SubSystemId, x.CustomId }); + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != "200" || userInfo.Data is null) + { + //TODO: log || ex + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}"); + } + + var name = userInfo.Data.UserName; + foreach (var group in orders.GroupBy(x => x.SubSystemId)) { // 新增/更新轨迹 @@ -107,7 +129,7 @@ public partial class TrackService : ApplicationService, ITrackService TrackPlace = input.TrackPlace, TrackTime = input.TrackTime!.Value, TrackTimeZone = input.TrackTimeZone!.Value, - OperationMan = CurrentUserRequest.CurrentUser.Name + OperationMan = name }; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event); } @@ -143,13 +165,22 @@ public partial class TrackService : ApplicationService, ITrackService throw new BusinessException(message: "没有一个轨迹代码是可删的"); } + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != "200" || userInfo.Data is null) + { + //TODO: log || ex + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}"); + } + + var name = userInfo.Data.UserName; + // 删除轨迹 var @event = new TrackDeleteEvent { OrderIds = new[] { order.Id }, TrackCodes = codes.Select(x => x.TrackCode).ToList(), SubSystemId = order.SubSystemId, - OperationMan = CurrentUserRequest.CurrentUser.Name + OperationMan = name }; await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event); @@ -182,6 +213,15 @@ public partial class TrackService : ApplicationService, ITrackService throw new BusinessException(message: "没有一个轨迹代码是可删的"); } + var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync(); + if (userInfo.Code != "200" || userInfo.Data is null) + { + //TODO: log || ex + throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}"); + } + + var name = userInfo.Data.UserName; + // 删除轨迹 foreach (var group in codes.GroupBy(x => x.OrderId)) { @@ -190,7 +230,7 @@ public partial class TrackService : ApplicationService, ITrackService OrderIds = group.Select(x => x.OrderId).ToArray(), TrackCodes = input.TrackCodes, SubSystemId = orders.First(x => x.Id == group.Key).SubSystemId, - OperationMan = CurrentUserRequest.CurrentUser.Name + OperationMan = name }; await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event); }