diff --git a/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs index a3d1091e..782126bb 100644 --- a/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs +++ b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs @@ -12,7 +12,7 @@ public interface IBroadcastingService : ITransientDependency /// /// /// - void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode); + void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode); /// /// 轨迹变更广播（删除） @@ -20,7 +20,7 @@ public interface IBroadcastingService : ITransientDependency /// /// /// - void TrackChangePublicTopicAsync(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode); + void TrackChangePublicTopic(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode); /// /// 轨迹变更广播（删除） @@ -28,7 +28,7 @@ public interface IBroadcastingService : ITransientDependency /// /// /// - void TrackChangePublicTopicAsync(TrackChangeEto eto, string subSystemCode); + void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode); /// /// 轨迹推送完成广播（推送完成回调） diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 1265a4d9..41956e2d 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -47,7 +47,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe //当轨迹删除，根据轨迹标记/状态上IsToMq选择进行广播 if (trackList.Count != 0) - await PublicTrackTopicAsync(@event, trackList); + _ = PublicTrackTopicAsync(@event, trackList); // 根据订单关联关系异步向下刷（异步） await DownByOrderLink(@event); @@ -121,89 +121,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe return true; } - private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List trackList) - { - var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray(); - var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray(); - - //从轨迹内容中查询出标记是否需要mq广播 - var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes); - var toMqDict = contentList - .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq); - - //选出需要广播的轨迹,并按code和订单号分组 - var trackGroup = trackList - .Where(x => toMqDict[x.SubSystemId + x.TrackCode]) - .GroupBy(m => new - { - m.TrackCode, - m.OrderNo - }) - .ToList(); - if (trackGroup.Count == 0) return; - - foreach (var dataItem in trackGroup) - { - var item = dataItem.First(); - var content = contentList.First(m => m.TrackContentCode == item.TrackCode); - - // _broadcastingService.TrackChangePublicTopicAsync(track, - // dataItem.Select(m => new OrderIdentifier - // { - // OrderId = m.OrderId, - // OrderNo = m.OrderNo, - // OrderType = m.OrderType - // }).ToArray(), - // item.TrackCode, - // item.TrackStatusCode, - // content.SubSystemCode - // ); - - //将每个订单的每个轨迹的删除变更发布 - foreach (var track in dataItem) - { - _broadcastingService.TrackChangePublicTopicAsync(new TrackChangeEto - { - EstimatedDeliveryDate = track.EstimatedDeliveryDate, - TrackPlacePostcode = track.TrackPlacePostcode, - TrackPlaceNation = track.TrackPlaceNation, - TrackPlaceNationId = track.TrackPlaceNationId, - TrackPlaceProvince = track.TrackPlaceProvince, - TrackPlaceProvinceId = track.TrackPlaceProvinceId, - TrackPlaceCity = track.TrackPlaceCity, - TrackPlaceCityId = track.TrackPlaceCityId, - Files = track.Files?.Select(x => new FileItemDto() - { - FileName = x.FileName, - OssKey = x.OssKey - }).ToList(), - Remarks = track.Remarks, - Source = track.Source, - IsVisibility = track.IsVisibility, - TrackTimeZone = track.TrackTimeZone, - TrackTime = track.TrackTime, - TrackPlace = track.TrackPlace, - TrackStatusCode = track.TrackStatusCode, - EnStatus = track.TrackEnContent, - Content = track.TrackContent, - TrackCode = track.TrackCode, - OperationMan = @event.OperationMan, - OrderIdentifiers = new[] - { - new OrderIdentifier() - { - OrderId = track.OrderId, - OrderNo = track.OrderNo, - OrderType = track.OrderType - } - }, - SubSystemId = @event.SubSystemId, - Extra = track.Extra - }, content.SubSystemCode); - } - } - } - private async Task> DeleteTrackAsync(TrackDeleteEvent @event) { List orderIds; @@ -264,7 +181,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe throw new BusinessException(message: $"当前轨迹不存在扩展信息，无法利用扩展字段：{@event.CompareFields}"); } - var extraDict = GetExtraDictForTextJson(x.Extra); + var extraDict = TextJsonHelper.DeserializeDictionary(x.Extra); if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) { throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}"); @@ -361,68 +278,88 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe return deleteTracks; } - private Dictionary? GetExtraDictForTextJson(string? extraStr) + private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List trackList) { - return string.IsNullOrWhiteSpace(extraStr) - ? null - : System.Text.Json.JsonSerializer.Deserialize>(extraStr); - } + var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray(); + var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray(); - //json Element和object的比较 - private bool EqJsonElementObject2(object? a, object? b) - { - if (a == null && b == null) - { - return true; - } + //从轨迹内容中查询出标记是否需要mq广播 + var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes); + var toMqDict = contentList + .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq); - if (a is JsonElement aJsonElement) - { - switch (aJsonElement.ValueKind) + //选出需要广播的轨迹,并按code和订单号分组 + var trackGroup = trackList + .Where(x => toMqDict[x.SubSystemId + x.TrackCode]) + .GroupBy(m => new { - case JsonValueKind.Undefined: - return false; - case JsonValueKind.Object: - return false; - case JsonValueKind.Array: - return false; - case JsonValueKind.String: - return string.Equals(aJsonElement.GetString(), b as string); - case JsonValueKind.Number: + m.TrackCode, + m.OrderNo + }) + .ToList(); + if (trackGroup.Count == 0) return; + + //增加延时确保写盘成功 + await Task.Delay(1000); + foreach (var dataItem in trackGroup) + { + var item = dataItem.First(); + var content = contentList.First(m => m.TrackContentCode == item.TrackCode); + + // _broadcastingService.TrackChangePublicTopic(track, + // dataItem.Select(m => new OrderIdentifier + // { + // OrderId = m.OrderId, + // OrderNo = m.OrderNo, + // OrderType = m.OrderType + // }).ToArray(), + // item.TrackCode, + // item.TrackStatusCode, + // content.SubSystemCode + // ); + + //将每个订单的每个轨迹的删除变更发布 + foreach (var track in dataItem) + { + _broadcastingService.TrackChangePublicTopic(new TrackChangeEto + { + EstimatedDeliveryDate = track.EstimatedDeliveryDate, + TrackPlacePostcode = track.TrackPlacePostcode, + TrackPlaceNation = track.TrackPlaceNation, + TrackPlaceNationId = track.TrackPlaceNationId, + TrackPlaceProvince = track.TrackPlaceProvince, + TrackPlaceProvinceId = track.TrackPlaceProvinceId, + TrackPlaceCity = track.TrackPlaceCity, + TrackPlaceCityId = track.TrackPlaceCityId, + Files = track.Files?.Select(x => new FileItemDto() { - if (b == null) return false; - - if (b is int bInt) - { - var aInt = aJsonElement.GetInt32(); - return aInt == bInt; - } - - if (b is long bLong) - { - var aLong = aJsonElement.GetInt64(); - return aLong == bLong; - } - - return false; - } - case JsonValueKind.True: - case JsonValueKind.False: + FileName = x.FileName, + OssKey = x.OssKey + }).ToList(), + Remarks = track.Remarks, + Source = track.Source, + IsVisibility = track.IsVisibility, + TrackTimeZone = track.TrackTimeZone, + TrackTime = track.TrackTime, + TrackPlace = track.TrackPlace, + TrackStatusCode = track.TrackStatusCode, + EnStatus = track.TrackEnContent, + Content = track.TrackContent, + TrackCode = track.TrackCode, + OperationMan = @event.OperationMan, + OrderIdentifiers = new[] { - if (b is bool bl) + new OrderIdentifier() { - return aJsonElement.GetBoolean() == bl; + OrderId = track.OrderId, + OrderNo = track.OrderNo, + OrderType = track.OrderType } - - return false; - } - case JsonValueKind.Null: - { - return false; - } + }, + SubSystemId = @event.SubSystemId, + Extra = track.Extra + }, content.SubSystemCode); } } - - return false; } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index fc9c5e03..cd792aa6 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -408,7 +408,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.IsAllPushWithLatest) { //同code同比较字段保证唯一 - var old = body!.Data + var old = body.Data .Where(x => { //反序列化出的JsonElement和字典字符串类型做比较 diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 135bb0bd..1cd62f13 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -56,7 +56,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe // 处理广播事件 if (content != null && (content.IsToMq || @event.IsToMq)) - _broadcastingService.TrackChangePublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode); + _ = EnsureWriteAfterPublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode); if (canNextOrderIds.Count == 0) return; // 根据订单关联关系异步向下刷（异步） @@ -277,7 +277,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe else { Debug.Assert(@event.CompareFields != null, "@event.CompareFields != null"); - var extraDict = GetExtraDictForTextJson(@event.Extra); + var extraDict = TextJsonHelper.DeserializeDictionary(@event.Extra); if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) { throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}"); @@ -285,7 +285,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe existTracks = existTracks.Where(x => { - var oldExtraDict = GetExtraDictForTextJson(x.Extra); + var oldExtraDict = TextJsonHelper.DeserializeDictionary(x.Extra); // tips：注意兼容性，历史轨迹的字段可能为空 if (oldExtraDict == null || !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue)) { @@ -432,13 +432,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return track; } - private Dictionary? GetExtraDictForTextJson(string? extraStr) - { - return string.IsNullOrWhiteSpace(extraStr) - ? null - : System.Text.Json.JsonSerializer.Deserialize>(extraStr); - } - private OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order) { return new OrderOperationRecordIndex @@ -452,4 +445,16 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe CreateTime = DateTime.Now }; } + + private async Task EnsureWriteAfterPublicTopicAsync(TrackUpsertEvent @event, string subSystemCode, string statusCode) + { + //理想情况在写入之后1s会刷盘 https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html + //增加一个版本号来确定是否真的变更了 + //没有刷盘成功事件回调 + //方案1：!!【轮询等待静默刷盘】查出版本号确实发生了变更在发送广播 + //方案2：考虑强制刷盘（对es压力大）后发送 + //方案3：采用一个较大延时，不查询直接发 + await Task.Delay(1000); + _broadcastingService.TrackChangePublicTopic(@event, subSystemCode, statusCode); + } } diff --git a/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs b/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs index e63b0360..b7c833cf 100644 --- a/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs +++ b/src/SJZY.Track.Application/Helpers/TextJsonHelper.cs @@ -18,11 +18,11 @@ public static class TextJsonHelper return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize(json); } - public static Dictionary? DeserializeDictionary(string? dictString) + public static Dictionary? DeserializeDictionary(string? dictJsonString) { - return string.IsNullOrWhiteSpace(dictString) + return string.IsNullOrWhiteSpace(dictJsonString) ? null - : System.Text.Json.JsonSerializer.Deserialize>(dictString); + : System.Text.Json.JsonSerializer.Deserialize>(dictJsonString); } /// diff --git a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs index 4add29c6..7322f296 100644 --- a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs +++ b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs @@ -26,7 +26,7 @@ public class BroadcastingService : IBroadcastingService /// 子系统代码 /// 轨迹状态代码 /// - public void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode) + public void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode) { var data = new TrackChangeEto { @@ -65,7 +65,7 @@ public class BroadcastingService : IBroadcastingService Encoding.UTF8.GetBytes(data.ToJson())); } - public void TrackChangePublicTopicAsync(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode) + public void TrackChangePublicTopic(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode) { var data = new TrackChangeEto { @@ -81,7 +81,7 @@ public class BroadcastingService : IBroadcastingService Encoding.UTF8.GetBytes(data.ToJson())); } - public void TrackChangePublicTopicAsync(TrackChangeEto eto, string subSystemCode) + public void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode) { _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", Encoding.UTF8.GetBytes(eto.ToJson())); diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 7e8206f8..31ab4fcc 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -255,11 +255,8 @@ public partial class TrackService : ApplicationService, ITrackService } var extraDict = System.Text.Json.JsonSerializer.Deserialize>(y.Extra); - if (extraDict!.TryGetValue(input.CompareFields!, out var compareValue)) - { - //todo:需要修改JsonElement和object的逻辑 - return compareValue.Equals(input.CompareValue); - } + if (extraDict!.TryGetValue(input.CompareFields!, out var compareValueJsonElement)) + return TextJsonHelper.EqJsonElementAndObject(compareValueJsonElement, input.CompareValue); return false; }) @@ -378,12 +375,9 @@ public partial class TrackService : ApplicationService, ITrackService throw new BusinessException(message: "轨迹extra为空无法使用比较字段比较，请检查轨迹"); } - var extraDict = System.Text.Json.JsonSerializer.Deserialize>(y.Extra); - if (extraDict!.TryGetValue(input.CompareFields!, out var compareValue)) - { - //todo:需要修改JsonElement和object的逻辑 - return compareValue.Equals(input.CompareValue); - } + var extraDict = TextJsonHelper.DeserializeDictionary(y.Extra); + if (extraDict!.TryGetValue(input.CompareFields!, out var compareValueJsonElement)) + return TextJsonHelper.EqJsonElementAndObject(compareValueJsonElement, input.CompareValue); return false; })