From 4473936577355a983c7d96bd402195dbbcb73716 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Wed, 15 Jan 2025 10:35:30 +0800 Subject: [PATCH 1/5] review --- .../CAP/EventHandlers/TrackUpsertEventHandler.cs | 2 ++ .../Services/ProvideServices/TrackService.cs | 3 +++ 2 files changed, 5 insertions(+) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 79f29ff8..135bb0bd 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -144,6 +144,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return false; } + // todo: 在这里，判断存在性 + return true; } diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index db991a76..92ce9710 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -55,6 +55,7 @@ public partial class TrackService : ApplicationService, ITrackService [HttpPost("Track/Upsert")] public async Task UpsertTrackAsync(UpsertTrackInput input) { + // todo: 代码太长了 var order = await _orderRepository.GetFirstAsync(x => x.SubSystemId == input.SubSystemId && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, x => new @@ -119,6 +120,7 @@ public partial class TrackService : ApplicationService, ITrackService .Where(x => x.TrackTime == input.TrackTime) .ToList(); + // todo：es落盘延迟是不确定的 //当为更新操作时发现更新的数目为0进行重试操作 if (!input.IsAdd && tracks.Count == 0) { @@ -133,6 +135,7 @@ public partial class TrackService : ApplicationService, ITrackService .ToList(); if (tracks.Count > 0) success = true; + // todo: 最大重试次数应该是可配置的，当出现的时候应该有告警，而不是直接给调用方，指数退避 2^N await Task.Delay(300); } finally From 0b6e20da3f742d3727a3c81fd2a8cb5dc68ddeb8 Mon Sep 17 00:00:00 2001 From: hardstylewyl <827778732@qq.com> Date: Wed, 15 Jan 2025 11:11:55 +0800 Subject: [PATCH 2/5] =?UTF-8?q?=E8=BF=98=E5=8E=9F=E9=B9=8F=E5=86=9B?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 104 +++--------------- 1 file changed, 18 insertions(+), 86 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 07f0380d..03acbc70 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -352,103 +352,35 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId); - //找到历史中推送成功的轨迹 - var trackPushRecords = await _trackPushRecordEsRepository.GetHistorySuccessTrackPushRecordAsync(log.OrderId); - //构建标识符，code+比较字段的值(如果提供的比较字段,否则使用轨迹时间) - var pushIdentitySuccess = trackPushRecords.Select(x => - { - //寻找映射配置 - var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue; - if (string.IsNullOrWhiteSpace(compareFields)) - return code; - - //尝试从扩展中提取目标值 - var extraValue = TextJsonHelper.ExtractValue(x.Extra, compareFields); - return code + extraValue; - }).ToHashSet(); - if (history != default) { body = TextJsonHelper.Deserialize(history.PushContent); - //todo-test:在历史选出推送成功的历史轨迹 - body!.Data = body.Data.Where(x => - { - var code = ((JsonElement) x["code"]!).GetString(); - if (string.IsNullOrWhiteSpace(compareFields)) - return pushIdentitySuccess.Contains(code); - - var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); - var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; - return pushIdentitySuccess.Contains(code + extraValue); - }).ToList(); - - //添加当前要推送的轨迹 - body.Data.Add(cur); - if (config.IsAllPushWithLatest) { - //历史轨迹中同一个code+比较字段联合作为key最新的一个 - var dict = new Dictionary(); - //todo-test:根据code和比较字段值分组，在每组里选出一个创建时间最新的 - foreach (var x in body.Data) + //同code同比较字段保证唯一 + var old = body!.Data + .Where(x => + { + //反序列化出的JsonElement和字典字符串类型做比较 + var a = ((JsonElement) x["code"]!).GetString(); + var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + return a == b; + }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + { + if (x["extra"] == null || log.Extra == null) return false; + return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); + }) + .FirstOrDefault(); + + if (old != null) { - var code = string.Empty; - var dateStr = string.Empty; - var key = string.Empty; - if (x["code"] is JsonElement codeEl) code = codeEl.GetString()!; - else code = x["code"]!.ToString(); - if (x["date"] is JsonElement dateEl) dateStr = dateEl.GetString()!; - else dateStr = x["date"]!.ToString(); - - if (string.IsNullOrWhiteSpace(compareFields)) - key = code; - else - { - var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement); - var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null; - key = code + extraValue; - } - - //只保留最新的一条数据 - if (dict.TryGetValue(key, out var tuple)) - { - //如果当前值创建时间大于历史值，则替换历史值 - if (DateTime.Parse(dateStr) > DateTime.Parse(tuple.dateStr)) - dict[key] = (dateStr, x); - } - else - dict[key] = (dateStr, x); + body.Data.Remove(old); } - // //同code同比较字段保证唯一 - // var old = body!.Data - // .Where(x => - // { - // //反序列化出的JsonElement和字典字符串类型做比较 - // var a = ((JsonElement) x["code"]!).GetString(); - // var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 - // return a == b; - // }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => - // { - // if (x["extra"] == null || log.Extra == null) return false; - // return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); - // }) - // .FirstOrDefault(); - // - // if (old != null) - // { - // body.Data.Remove(old); - // } - - //选出每个组最新的轨迹 - body.Data = dict.Values.Select(x => x.dict).Cast>().ToList(); } - // //添加当前要推送的轨迹 - // body.Data.Add(cur); + body!.Data.Add(cur); } - - //todo-test:按推送映射顺序排序 } log.PushContent = TextJsonHelper.Serialize(body)!; From 08f0ebeeaebf907a16679122a0c4518224ac01ba Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Wed, 15 Jan 2025 15:02:03 +0800 Subject: [PATCH 3/5] =?UTF-8?q?optimize&fix=EF=BC=9Aforce=20push=20event?= =?UTF-8?q?=20case?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 291 +++++++++--------- 1 file changed, 153 insertions(+), 138 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 03acbc70..09d58a78 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -96,130 +96,21 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private async Task DoPushAsync(TrackPushEvent @event) { var items = new List(); - TrackPushContentConfigDto? config; - var isSystemPush = @event.OperationMan == null; - if (isSystemPush && !@event.ForcedPush) + if (@event.OperationMan == null) { - config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); - foreach (var order in _orders) + if (@event.OptionType == OptionType.新增) { - if (config == null) - { - TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); - return; - } - - // 系统不推异常单，只能人工操作 - if (order.QuestionType == (int) TrackProblemType.Error) - { - TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order); - continue; - } - - // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 - var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); - if (resultConfig == null) - { - TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order); - continue; - } - - config = resultConfig; - - // 是否自动推送 - if (!config.IsAutoPush) - { - TrackPushFailedPublicTopic($"没有设置为自动推送，如果要推送请手动推送", @event, order); - continue; - } - - // 配置的类型如何和订单类型不一致则不允许推送 - if (config.OrderType != order.OrderType) - { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); - continue; - } - - var log = new TrackPushRecordIndex - { - OptionType = @event.OptionType.ToString(), - OrderId = order.Id, - OrderNo = order.OrderNo, - TrackStatus = @event.SystemPushInfo.TrackStatus, - TrackCode = @event.TrackCode, - TrackContent = @event.SystemPushInfo.TrackContent, - TrackPlace = @event.SystemPushInfo.TrackPlace, - TrackTime = @event.SystemPushInfo.TrackTime, - TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, - CreateUser = GlobalConst.SystemUser, - SubSystemCode = @event.SubSystemCode, - IsToMq = @event.IsToMq, - Extra = @event.SystemPushInfo.Extra - }; - var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); - item.Log = JsonConvert.SerializeObject(log); - items.Add(item); + await BuiltSystemPushItemsAsync(@event, items); + } + else if (@event.OptionType == OptionType.修改) + { + await BuiltManualPushItemsAsync(@event, items); } } - // 手工或强制推送，不会校验已推和顺序问题，因为可能是更新 else - { - config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); - foreach (var order in _orders) - { - if (config == null) - { - TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); - return; - } - - if (config is { IsAllPush: true }) - { - // 非强制推送需要 推断 本次推送 是否可行 - if (!@event.ForcedPush) - { - var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); - if (resultConfig == null) - { - TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order); - continue; - } - - config = resultConfig; - } - - // tips:强制推送直接使用配置 - } - - - //配置的类型如何和订单类型不一致则不允许推送 - if (config.OrderType != order.OrderType) - { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); - continue; - } - - var log = new TrackPushRecordIndex - { - OptionType = @event.OptionType.ToString(), - OrderId = order.Id, - OrderNo = order.OrderNo, - TrackStatus = @event.SystemPushInfo.TrackStatus, - TrackCode = @event.TrackCode, - TrackContent = @event.SystemPushInfo.TrackContent, - TrackPlace = @event.SystemPushInfo.TrackPlace, - TrackTime = @event.SystemPushInfo.TrackTime, - TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, - CreateUser = @event.OperationMan, - SubSystemCode = @event.SubSystemCode, - IsToMq = @event.IsToMq, - Extra = @event.SystemPushInfo.Extra - }; - var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); - item.Log = JsonConvert.SerializeObject(log); - items.Add(item); - } + // 手工推送，不会校验已推和顺序问题，因为可能是更新 + await BuiltManualPushItemsAsync(@event, items); } if (items.Count == 0) return; @@ -236,12 +127,129 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != GlobalConst.SuccessCode) { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); - throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + items.Cast().ToList()); + throw new BusinessException( + message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } + private async Task BuiltManualPushItemsAsync(TrackPushEvent @event, List items) + { + // tips: 手工强制推送直接使用配置 + var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + foreach (var order in _orders) + { + if (config is { IsAllPush: true }) + { + // 非强制推送需要 推断 本次推送 是否可行 + if (!@event.ForcedPush) + { + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + } + } + + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + return; + } + + + //配置的类型如何和订单类型不一致则不允许推送 + if (config.OrderType != order.OrderType) + { + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); + continue; + } + + var log = new TrackPushRecordIndex + { + OptionType = @event.OptionType.ToString(), + OrderId = order.Id, + OrderNo = order.OrderNo, + TrackStatus = @event.SystemPushInfo.TrackStatus, + TrackCode = @event.TrackCode, + TrackContent = @event.SystemPushInfo.TrackContent, + TrackPlace = @event.SystemPushInfo.TrackPlace, + TrackTime = @event.SystemPushInfo.TrackTime, + TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, + CreateUser = @event.OperationMan, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq, + Extra = @event.SystemPushInfo.Extra + }; + var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); + item.Log = JsonConvert.SerializeObject(log); + items.Add(item); + } + } + + private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List items) + { + var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + foreach (var order in _orders) + { + if (config == null) + { + TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); + return; + } + + // 系统不推异常单，只能人工操作 + if (order.QuestionType == (int) TrackProblemType.Error) + { + TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order); + continue; + } + + // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + continue; + } + + // 是否自动推送 + if (!config.IsAutoPush) + { + TrackPushFailedPublicTopic("没有设置为自动推送，如果要推送请手动推送", @event, order); + continue; + } + + // 配置的类型如何和订单类型不一致则不允许推送 + if (config.OrderType != order.OrderType) + { + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); + continue; + } + + var log = new TrackPushRecordIndex + { + OptionType = @event.OptionType.ToString(), + OrderId = order.Id, + OrderNo = order.OrderNo, + TrackStatus = @event.SystemPushInfo.TrackStatus, + TrackCode = @event.TrackCode, + TrackContent = @event.SystemPushInfo.TrackContent, + TrackPlace = @event.SystemPushInfo.TrackPlace, + TrackTime = @event.SystemPushInfo.TrackTime, + TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, + CreateUser = GlobalConst.SystemUser, + SubSystemCode = @event.SubSystemCode, + IsToMq = @event.IsToMq, + Extra = @event.SystemPushInfo.Extra + }; + var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); + item.Log = JsonConvert.SerializeObject(log); + items.Add(item); + } + } + private async Task DoUpdateAsync(TrackPushEvent @event) { var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); @@ -253,7 +261,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe //配置的类型如何和订单类型不一致则不允许推送 if (config.OrderType != order.OrderType) { - TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order); + TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, + order); continue; } @@ -292,8 +301,10 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (r.Code != GlobalConst.SuccessCode) { - TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); - throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); + TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, + items.Cast().ToList()); + throw new BusinessException( + message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } @@ -363,12 +374,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { //反序列化出的JsonElement和字典字符串类型做比较 var a = ((JsonElement) x["code"]!).GetString(); - var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + var b = cur["code"]! + .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 return a == b; - }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + }) + .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => { if (x["extra"] == null || log.Extra == null) return false; - return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); + return TextJsonHelper.EqByJsonElementField(x["extra"]!, + TextJsonHelper.Deserialize(log.Extra), compareFields!); }) .FirstOrDefault(); @@ -376,8 +390,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { body.Data.Remove(old); } - } + // //添加当前要推送的轨迹 body!.Data.Add(cur); } @@ -392,24 +406,28 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe /// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了 /// 则查询有没有可交换的推送，没有就返回null /// - private static (TrackPushContentConfigDto? config, string nextCode) GetValidPushConfigAsync(List cache, - string? latestPushTrackCode, string? trackCode) + private static TrackPushContentConfigDto? GetValidPushConfigAsync(List cache, + string? latestPushTrackCode, TrackPushEvent @event) { + var trackCode = @event.TrackCode; var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); + if (@event.ForcedPush && trackCode == latestPushTrackCode) + { + //强制推送，允许重复推送当前轨迹 + return cache[idx]; + } + if (idx + 1 >= cache.Count) return default; // 推完了 // find next var matched = cache[idx + 1].TrackCode == trackCode ? cache[idx + 1] - : // find swap - cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); + : cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); //没推完，可能不符合推送顺序 - return matched != null && matched.TrackCode == trackCode - ? (matched, matched.TrackCode) - : (default, matched?.TrackCode ?? cache[idx + 1].TrackCode); + return matched != null && matched.TrackCode == trackCode ? matched : default; } /// @@ -433,10 +451,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } //轨迹推送失败(被拦截) - private void TrackPushFailedPublicTopic( - string failedMessage, - TrackPushEvent @event, - OrderEntity order) + private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order) { _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { From 82bbca3002e0685b437b28742ec029c169ac7476 Mon Sep 17 00:00:00 2001 From: hardstylewyl <827778732@qq.com> Date: Wed, 15 Jan 2025 17:41:06 +0800 Subject: [PATCH 4/5] =?UTF-8?q?update:=E8=B0=83=E6=95=B4=E8=BD=A8=E8=BF=B9?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 98 +++++++++++++------ .../Services/ProvideServices/TrackService.cs | 9 ++ .../IOrderOperationRecordEsRepository.cs | 8 ++ .../ITrackPushRecordEsRepository.cs | 12 ++- .../OrderOperationRecordEsRepository.cs | 14 ++- .../TrackPushRecordEsRepository.cs | 27 ++++- 6 files changed, 132 insertions(+), 36 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 09d58a78..a7cc3d36 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -1,3 +1,4 @@ +using System.Collections.Immutable; using System.Text.Json; using DotNetCore.CAP; using Microsoft.Extensions.Options; @@ -11,6 +12,7 @@ using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Options; using SJZY.Track.Application.Track2InterfaceRelay; using SJZY.Track.Domain.Aggregates.OrderAggregate; +using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; @@ -25,6 +27,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private readonly IOrderRepository _orderRepository; private readonly ITrackPushCacheService _trackPushCacheService; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; + private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private List _orders = null!; private List _pushConfigs = null!; @@ -34,7 +37,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe ITrackPushCacheService trackPushCacheService, ITrackPushRecordEsRepository trackPushRecordEsRepository, IOptionsMonitor options, - IBroadcastingService broadcastingService) + IBroadcastingService broadcastingService, + ITrackDetailsEsRepository trackDetailsEsRepository) { _httpClientProxy = httpClientProxy; _orderRepository = orderRepository; @@ -42,6 +46,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe _trackPushRecordEsRepository = trackPushRecordEsRepository; _options = options; _broadcastingService = broadcastingService; + _trackDetailsEsRepository = trackDetailsEsRepository; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] @@ -206,11 +211,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序 - config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); - if (config == null) + // 如果当前推送code与订单上的最后的code不一致，则考虑推送映射顺序 + if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode)) { - TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); - continue; + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event); + if (config == null) + { + TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order); + continue; + } } // 是否自动推送 @@ -362,38 +371,67 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPush) { - var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId); + var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); if (history != default) { body = TextJsonHelper.Deserialize(history.PushContent); - if (config.IsAllPushWithLatest) - { - //同code同比较字段保证唯一 - var old = body!.Data - .Where(x => - { - //反序列化出的JsonElement和字典字符串类型做比较 - var a = ((JsonElement) x["code"]!).GetString(); - var b = cur["code"]! - .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 - return a == b; - }) - .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => - { - if (x["extra"] == null || log.Extra == null) return false; - return TextJsonHelper.EqByJsonElementField(x["extra"]!, - TextJsonHelper.Deserialize(log.Extra), compareFields!); - }) - .FirstOrDefault(); - if (old != null) + //在推送历史中，选出当前轨迹明细一致的轨迹 + var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId); + //确定轨迹标识符 + var trackIdentifies = tracks.Select(x => { - body.Data.Remove(old); + //寻找映射配置(转换后的code) + var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue; + var timeStr = DateTimeFormat(x.TrackTime, config.TrackTimeZone ?? log.TrackTimeZone, config); + return string.IsNullOrWhiteSpace(compareFields) + ? $"{code}{timeStr}" + : $"{code}{TextJsonHelper.ExtractValue(x.Extra, compareFields)}"; } - } + ).ToImmutableHashSet(); + //在历史中选取在订单当前轨迹明细中的记录 + body!.Data = body.Data.Where(x => + { + var code = ((JsonElement) x["code"]!).GetString(); + var timeStr = ((JsonElement) x["date"]!).GetString(); - // //添加当前要推送的轨迹 - body!.Data.Add(cur); + if (string.IsNullOrWhiteSpace(compareFields)) + return trackIdentifies.Contains($"{code}{timeStr}"); + + if (((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var comparerValueJsonElement)) + return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}"); + + return trackIdentifies.Contains($"{code}{null}"); + }).ToList(); + + // if (config.IsAllPushWithLatest) + // { + // //同code同比较字段保证唯一 + // var old = body!.Data + // .Where(x => + // { + // //反序列化出的JsonElement和字典字符串类型做比较 + // var a = ((JsonElement) x["code"]!).GetString(); + // var b = cur["code"]! + // .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + // return a == b; + // }) + // .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + // { + // if (x["extra"] == null || log.Extra == null) return false; + // return TextJsonHelper.EqByJsonElementField(x["extra"]!, + // TextJsonHelper.Deserialize(log.Extra), compareFields!); + // }) + // .FirstOrDefault(); + // + // if (old != null) + // { + // body.Data.Remove(old); + // } + // } + + // 添加当前要推送的轨迹 + body.Data.Add(cur); } } diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 92ce9710..7e8206f8 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Http; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services.ProvideServices; +using SJZY.Track.Application.Helpers; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackContentAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -120,6 +121,14 @@ public partial class TrackService : ApplicationService, ITrackService .Where(x => x.TrackTime == input.TrackTime) .ToList(); + //根据比较字段选出所需的轨迹 + if (!string.IsNullOrWhiteSpace(input.CompareFields)) + { + var a = input.TrackCode + TextJsonHelper.ExtractValue(input.Extra, input.CompareFields); + tracks = tracks.Where(x => string.Equals(a, + x.TrackCode + TextJsonHelper.ExtractValue(x.Extra, input.CompareFields!))).ToList(); + } + // todo：es落盘延迟是不确定的 //当为更新操作时发现更新的数目为0进行重试操作 if (!input.IsAdd && tracks.Count == 0) diff --git a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs index eb97ddae..18485b5a 100644 --- a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs @@ -16,6 +16,14 @@ public interface IOrderOperationRecordEsRepository /// Task AddManyAsync(List indexes); + /// + /// 根据id删除订单操作记录 + /// + /// + /// + /// + Task DeleteOperationRecordByIdAsync(string id, string indexName); + /// /// 获取订单操作记录 /// diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs index 261599f3..de1f0fa3 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs @@ -11,6 +11,14 @@ public interface ITrackPushRecordEsRepository Task AddManyAsync(List index); + /// + /// 根据id删除轨迹推送记录 + /// + /// + /// + /// + Task DeletePushRecordByIdAsync(string id, string indexName); + /// /// 获取轨迹推送记录 /// @@ -22,9 +30,9 @@ public interface ITrackPushRecordEsRepository /// 获取最新轨迹推送记录 /// /// + /// 只选择成功的记录 /// - Task GetLatestTrackPushRecordAsync(int orderId); - + Task GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false); /// /// 获取历史中推送成功的记录 diff --git a/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs index 0c30c860..7230d375 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs @@ -47,6 +47,18 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract(id, u => u + .Index(indexName) + ); + + if (!r.IsValid) + { + throw new InvalidOperationException($"轨迹操作记录失败，id：{indexName}，原因：{r.DebugInformation}。"); + } + } + public async Task> GetOrderOperationsByIdAsync(int orderId) { var data = await Context.SearchAsync(x => x @@ -57,6 +69,6 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); - return data.Documents.ToList(); + return data.ToIdList(); } } diff --git a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs index b606bc29..3cbef97d 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs @@ -45,6 +45,18 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract(id, u => u + .Index(indexName) + ); + + if (!r.IsValid) + { + throw new InvalidOperationException($"轨迹推送记录失败，id：{indexName}，原因：{r.DebugInformation}。"); + } + } + public async Task> GetTrackPushRecordsAsync(int orderId) { @@ -56,15 +68,24 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract s.Field(f => f.CreateTime, SortOrder.Descending))); - return data.Documents.ToList(); + return data.ToIdList(); } - public async Task GetLatestTrackPushRecordAsync(int orderId) + public async Task GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false) { + var musts = new List, QueryContainer>> + { + mu => mu.Term(t => t.OrderId, orderId), + }; + + if (isSuccess) + musts.Add(mu => mu.Term(t => t.IsSuccess, true)); + var data = await Context.SearchAsync(x => x .Index(GlobalConst.TrackPushRecordIndex) .Query(q => - q.Term(t => t.OrderId, orderId) + q.Bool(b => + b.Must(musts)) ) .Size(1000) .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending))); From 3ca803fe5b5009a3ec1e087ba548f13f7c76777a Mon Sep 17 00:00:00 2001 From: hardstylewyl <827778732@qq.com> Date: Wed, 15 Jan 2025 17:41:58 +0800 Subject: [PATCH 5/5] =?UTF-8?q?update:=E4=BF=9D=E7=95=99=E5=8F=AA=E6=8E=A8?= =?UTF-8?q?=E6=9C=80=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../EventHandlers/TrackPushEventHandler.cs | 51 ++++++++++--------- 1 file changed, 26 insertions(+), 25 deletions(-) diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index a7cc3d36..fc9c5e03 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -404,31 +404,32 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe return trackIdentifies.Contains($"{code}{null}"); }).ToList(); - // if (config.IsAllPushWithLatest) - // { - // //同code同比较字段保证唯一 - // var old = body!.Data - // .Where(x => - // { - // //反序列化出的JsonElement和字典字符串类型做比较 - // var a = ((JsonElement) x["code"]!).GetString(); - // var b = cur["code"]! - // .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 - // return a == b; - // }) - // .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => - // { - // if (x["extra"] == null || log.Extra == null) return false; - // return TextJsonHelper.EqByJsonElementField(x["extra"]!, - // TextJsonHelper.Deserialize(log.Extra), compareFields!); - // }) - // .FirstOrDefault(); - // - // if (old != null) - // { - // body.Data.Remove(old); - // } - // } + + if (config.IsAllPushWithLatest) + { + //同code同比较字段保证唯一 + var old = body!.Data + .Where(x => + { + //反序列化出的JsonElement和字典字符串类型做比较 + var a = ((JsonElement) x["code"]!).GetString(); + var b = cur["code"]! + .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同 + return a == b; + }) + .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => + { + if (x["extra"] == null || log.Extra == null) return false; + return TextJsonHelper.EqByJsonElementField(x["extra"]!, + TextJsonHelper.Deserialize(log.Extra), compareFields!); + }) + .FirstOrDefault(); + + if (old != null) + { + body.Data.Remove(old); + } + } // 添加当前要推送的轨迹 body.Data.Add(cur);