diff --git a/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs b/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs new file mode 100644 index 00000000..c20366d1 --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Dtos/TrackPull/TrackPullPayload.cs @@ -0,0 +1,9 @@ +namespace SJZY.Track.Application.Contract.Dtos.TrackPull; + +public class TrackPullPayload +{ + /// + /// 订单号 + /// + public string OrderNo { get; set; } = null!; +} \ No newline at end of file diff --git a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj index 749e25a6..82b6c428 100644 --- a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj +++ b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj @@ -15,8 +15,5 @@ - - - diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 402668e3..84b86d8e 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -74,7 +74,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe private async Task DoPushAsync(TrackPushEvent @event) { - var items = new List(); + var items = new List(); TrackPushContentConfigDto? config; if (@event.OperationMan == null) @@ -165,7 +165,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null || !config.IsCanUpdate) return; - var items = new List(); + var items = new List(); foreach (var order in _orders) { var log = new TrackPushRecordIndex diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs b/src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs similarity index 79% rename from src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs rename to src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs index abc24922..9f9dcfbe 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs +++ b/src/SJZY.Track.Application/Rabbit/TrackRefreshEventHandler.cs @@ -1,8 +1,14 @@ using System.Diagnostics; using System.Linq.Expressions; +using System.Text; +using BSI.Stack.Rabbit; +using BSI.Stack.Rabbit.Handler; using DotNetCore.CAP; +using Microsoft.Extensions.Logging; using Newtonsoft.Json; using Newtonsoft.Json.Linq; +using RabbitMQ.Client.Events; +using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.TrackPull; using SJZY.Track.Application.Contract.Events; @@ -15,37 +21,72 @@ using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; -namespace SJZY.Track.Application.CAP.EventHandlers; +namespace SJZY.Track.Application.Rabbit; -public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe +public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler { private readonly ICapPublisher _capPublisher; private readonly IOrderRepository _orderRepository; private readonly ITimezoneQueries _timezoneQueries; + private readonly ILogger _logger; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ITrackPullCacheService _trackPullCacheService; public TrackRefreshEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService, - ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries) + ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries, + ILogger logger) { _orderRepository = orderRepository; _capPublisher = capPublisher; _trackPullCacheService = trackPullCacheService; _trackDetailsRepository = trackDetailsRepository; _timezoneQueries = timezoneQueries; + _logger = logger; + } + + + public bool Enable(ConsumeOptions options) + { + return options.Queue == GlobalConst.TrackRefreshSubscribe; + } + + public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) + { + var msg = Encoding.UTF8.GetString(args.Body.ToArray()); + var @event = System.Text.Json.JsonSerializer.Deserialize(msg); + if (@event == null) + { + _logger.LogWarning( + $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); + return; + } + + await ConsumerAsync(@event); } - [CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)] public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { - if (!@event.IsSuccess) return; + if (!@event.IsSuccess) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹刷新失败", @event.RequestId); + return; + } + var attach = Base64Helper.DeSerialize(@event.Attach); - if (attach is null) return; + if (attach is null || @event.Data == null) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，attach或data解析为空", @event.RequestId); + return; + } var config = await _trackPullCacheService.GetPullParamWithContentConfigAsync(attach.SubSystemId, attach.ServiceProviderId); - if (@event.Data == null || config == null) return; + if (config == null) + { + _logger.LogWarning("请求id: {@RequestId}，轨迹消费失败，找不到拉取配置：{@Attach}", attach.ToJson(), @event.RequestId); + return; + } var data = JObject.Parse(@event.Data); @@ -183,28 +224,22 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe // 轨迹时区 var (trackTimeZone, hasTimezoneBottomLine) = - await GetTimezone(paramConfig, matchedContentConfig, context.OrderId, trackToken); + await GetTimezone(paramConfig, matchedContentConfig, context, trackToken); // 轨迹发生地 var (trackPlace, hasPlaceBottomLine) = - await GetTrackPlace(paramConfig, matchedContentConfig, context.OrderId, trackToken); + await GetTrackPlace(paramConfig, matchedContentConfig, context, trackToken); if (matchedContentConfig is null) { + _logger.LogDebug("订单号：{OrderNo}，未匹配到轨迹内容，轨迹报文：{TrackContent}", + context.OrderNo, trackToken.ToString()); await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, paramConfig); return; } - int deficiencyTagValue = 0; - // 触发了兜底，但是没有兜底住，对订单进行标记 - if (hasPlaceBottomLine || hasTimezoneBottomLine) - { - //1表示时区，2表示发生地，3表示两种 - var placeTagValue = hasPlaceBottomLine ? 2 : 0; - var timezoneTagValue = hasTimezoneBottomLine ? 1 : 0; - deficiencyTagValue = placeTagValue + timezoneTagValue; - } - + var deficiencyTagValue = CalcDeficiencyTagValue(hasPlaceBottomLine, hasTimezoneBottomLine, trackToken, context); + // 新增/更新轨迹 var trackUpsertEvent = new TrackUpsertEvent { @@ -239,6 +274,35 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } + /// + /// 触发了兜底，但是没有兜底住，对订单进行标记 + /// 1表示时区，2表示发生地，3表示两种 + /// + /// + /// + /// + /// + /// + private int CalcDeficiencyTagValue(bool hasPlaceBottomLine, bool hasTimezoneBottomLine, JToken trackToken, + TrackRefreshContext context) + { + var deficiencyTagValue = 0; + + if (hasTimezoneBottomLine) + { + _logger.LogWarning("订单号：{@OrderNo}, 轨迹发生时区无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString()); + deficiencyTagValue += 1; + } + + if (hasPlaceBottomLine) + { + _logger.LogWarning("订单号：{@OrderNo}, 轨迹发生地无法解析，触发了兜底，但是没有兜底住：{@Data}", context.OrderNo, trackToken.ToString()); + deficiencyTagValue += 2; + } + + return deficiencyTagValue; + } + private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, DateTime? trackTime, float? trackTimeZone, TrackPullParamConfigDto paramConfig) { @@ -263,7 +327,9 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe trackToken[paramConfig.TrackContentName!]?.Value() ?? GlobalConst.EmptyString; } - track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); + track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 + ? track.TrackTime + : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); await _trackDetailsRepository.AddAsync(track); } @@ -273,11 +339,11 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe /// /// 配置信息 /// 配置信息 - /// 订单id + /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(float? trackTimeZone, bool hasTimezoneBottomLine)> GetTimezone( - TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, int orderId, + TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生时区 @@ -289,7 +355,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe if (!trackTimeZone.HasValue && matchedContentConfig?.TrackZonePlaceValue is not null) { //兜底没配置的就不需要兜底，拉取的发生地没就没 - var timezone = await GetTimezone(matchedContentConfig, orderId); + var timezone = await GetTimezone(matchedContentConfig, context.OrderId); trackTimeZone = timezone; hasTimezoneBottomLine = !timezone.HasValue; @@ -303,22 +369,21 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe /// /// 配置信息 /// 配置信息 - /// 订单id + /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(string? trackPlace, bool hasPlaceBottomLine)> GetTrackPlace(TrackPullParamConfigDto paramConfig, - TrackPullContentConfigDto? matchedContentConfig, int orderId, JToken trackToken) + TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName == null ? null : trackToken[paramConfig.TrackPlaceName]?.Value(); var hasPlaceBottomLine = false; //执行了兜底但是没有兜底住，需要记录 - if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue != null && - matchedContentConfig.TrackPlaceValue.HasValue) + if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue is not null) { //兜底没配置的就不需要兜底，拉取的发生地没就没 - trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, orderId); + trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, context.OrderId); hasPlaceBottomLine = string.IsNullOrWhiteSpace(trackPlace); } diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index 011dd0d7..5f44d891 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -12,7 +12,8 @@ - + + diff --git a/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs b/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs index 1e6cb204..f0c8edc9 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs @@ -67,16 +67,21 @@ public partial class OrderService x => x.LinkOrderNo); // 计算出未关联的订单 - var noLinkOrderNos = input.LinkOrderNos.Except(linkOrderNos); + var noLinkOrderNos = input.LinkOrderNos.Except(linkOrderNos).ToList(); - // 过滤掉不存在的订单 - var canLinkOrders = await _orderRepository.GetListAsync(x => + // 查询存在的订单 + var existedOrders = await _orderRepository.GetListAsync(x => x.OrderType == input.LinkOrderType && noLinkOrderNos.Contains(x.OrderNo), x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId }); - if (canLinkOrders.Count == 0) return; + // 如果有不存在的单，需要抛异常 + if (existedOrders.Count != noLinkOrderNos.Count) + { + throw new BusinessException( + message: $"关联单：{noLinkOrderNos.Except(existedOrders.Select(x => x.OrderNo)).JoinAsString(", ")}不存在"); + } - await _orderRepository.LinkOrderAsync(canLinkOrders.Select(x => new OrderLinkEntity + await _orderRepository.LinkOrderAsync(existedOrders.Select(x => new OrderLinkEntity { OrderId = orderId, OrderNo = input.OrderNo, @@ -89,13 +94,14 @@ public partial class OrderService var tracks = await _trackDetailsRepository.GetListAsync(orderId); // 补轨迹 - foreach (var group in canLinkOrders.GroupBy(x => x.SubSystemId)) + foreach (var group in existedOrders.GroupBy(x => new {x.SubSystemId, x.CustomId})) { foreach (var track in tracks) { var fillTrackEvent = new TrackUpsertEvent { - SubSystemId = group.Key, + SubSystemId = group.Key.SubSystemId, + CustomId = group.Key.CustomId, OrderIdentifiers = group.Select(x => new OrderIdentifier { OrderId = x.Id, diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs index b781d848..494352a4 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs @@ -1,14 +1,13 @@ using Microsoft.AspNetCore.Authorization; -using Microsoft.Extensions.Logging; using SJZY.Expand.ABP.Core; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; -using SJZY.Track.Application.CAP.EventHandlers; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Dtos.TrackPull; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Helpers; +using SJZY.Track.Application.Rabbit; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.Http.Client.DynamicProxying; @@ -59,11 +58,10 @@ public class TrackPullService : ApplicationService foreach (var group in orders.GroupBy(x => new { x.ServiceProviderId, x.ServiceProviderName, x.SubSystemId, x.OrderType })) { - var data = group.Select(x => new + var data = new List(group.Select(x => new TrackPullPayload { - x.OrderNo - }).ToList(); - + OrderNo = x.OrderNo + })); var serviceProviderCode = group.FirstOrDefault()!.ServiceProviderCode; var attach = Base64Helper.SerializeObject(group.Key); @@ -71,7 +69,7 @@ public class TrackPullService : ApplicationService var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); - + if (res.Code != InterfaceRelayConst.SuccessCode) { LogHelper.Error($"刷新轨迹失败：{res.ToJson()}"); @@ -94,7 +92,7 @@ public class TrackPullService : ApplicationService await Task.WhenAll(tasks); return true; } - + /// /// 轨迹刷新测试 /// @@ -105,7 +103,8 @@ public class TrackPullService : ApplicationService #if DEBUG [AllowAnonymous] #endif - public async Task TrackRefreshEventTest([FromServices]TrackRefreshEventHandler trackRefreshEventHandler, InterfaceRelayForwardReplayEvent input) + public async Task TrackRefreshEventTest([FromServices] TrackRefreshEventHandler trackRefreshEventHandler, + InterfaceRelayForwardReplayEvent input) { await trackRefreshEventHandler.ConsumerAsync(input); @@ -121,7 +120,7 @@ public class TrackPullService : ApplicationService } var page = 1; - var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page, 2000); + var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page); while (orders.Any()) { var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; @@ -129,9 +128,15 @@ public class TrackPullService : ApplicationService 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)); - var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); + var data = new List(group.Select(x => new TrackPullPayload + { + OrderNo = x.OrderNo + })); + var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, + serviceProviderName!, group.Key.SubSystemId, group.Key.OrderType)); + var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, + GlobalConst.TrackRefreshSubscribe, attach); if (res.Code != InterfaceRelayConst.SuccessCode) { diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.Query.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.Query.cs index a055513f..2ce8b135 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.Query.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.Query.cs @@ -27,7 +27,24 @@ public partial class TrackService var res = new List(); input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空 - var details = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos); + + object[]? searchAfter = null; + List details = new(); + int limit = 1000, cnt, time = 10; + + do + { + var (currentResults, nextSearchAfter) = + await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, searchAfter, limit); + cnt = currentResults.Count; + details.AddRange(currentResults); + + // 更新 searchAfter 以获取下一批数据 + searchAfter = nextSearchAfter; + time--; + } while (cnt > 0 && time > 0); + + if (details.Count == limit * time) throw new BusinessException(message: "查询轨迹量过大，请缩小查询范围"); foreach (var group in details.GroupBy(x => x.OrderId)) { @@ -42,8 +59,12 @@ public partial class TrackService Content = x.TrackContent, OriginContent = x.OriginTrackContent, Location = x.TrackPlace, - Time = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? x.TrackTime: x.ZeroTrackTime.AddHours(input.TimeZone ?? 8), - TimeZone = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1? null: input.TimeZone ?? 8 + Time = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 + ? x.TrackTime + : x.ZeroTrackTime.AddHours(input.TimeZone ?? 8), + TimeZone = Math.Abs(x.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 + ? null + : input.TimeZone ?? 8 }).ToList(); res.Add(item); diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 3d76bd98..ed053823 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -66,6 +66,7 @@ public partial class TrackService : ApplicationService, ITrackService // 新增/更新轨迹 var @event = ObjectMapper.Map(input); @event.SubSystemId = order.SubSystemId; + @event.CustomId = order.CustomId; @event.OrderIdentifiers = new[] { new OrderIdentifier diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 0dfe7b1e..50dc5959 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -1,6 +1,8 @@ -using DotNetCore.CAP; +using BSI.Stack.Rabbit; +using DotNetCore.CAP; using DotNetCore.CAP.Internal; using Microsoft.Extensions.Configuration; +using RabbitMQ.Client; using Savorboard.CAP.InMemoryMessageQueue; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; @@ -9,6 +11,7 @@ using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.CAP; using SJZY.Track.Application.Contract; using SJZY.Track.Application.Localization; +using SJZY.Track.Domain.Consts; using SJZY.UserCenter.Contracts; using SJZY.UserCenter.SharedModel; using SJZY.UserCenter.SharedModel.Pub; @@ -121,6 +124,29 @@ namespace SJZY.Track.Application } }; }).AddSubscribeFilter(); + + + context.Services.AddRabbitBoot(options => + { + options.ConnectionString = + $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; + options.PoolSize = configuration.GetValue("RabbitMQ:PoolSize"); + options.Consumes = new List + { + new() + { + Queue = GlobalConst.TrackRefreshSubscribe, + FetchCount = configuration.GetValue("RabbitMQ:.TrackRefreshFetchCount"), + FailedRequeue = false, + Declaration = o => + { + o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true); + o.QueueDeclare(GlobalConst.TrackRefreshSubscribe, true); + o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange, GlobalConst.TrackRefreshSubscribe); + } + } + }; + }, typeof(SjzyTrackApplicationModule).Assembly); } public override void PreConfigureServices(ServiceConfigurationContext context) diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index 639f92ef..cec7351b 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -8,6 +8,8 @@ public interface ITrackDetailsRepository Task> GetLatestListAsync(int[] eventOrderIds, List trackCodes); + Task<(List results, object[]? nextSearchAfter)> GetListAsync(int orderType, List orderNos, object[]? searchAfter = null, int limit = 1000); + Task AnyAsync(int orderId, DateTime trackTime); Task AnyAsync(int orderId, string originTrackDataHash); @@ -20,8 +22,6 @@ public interface ITrackDetailsRepository Task AddAsync(TrackDetailsIndex track); - Task> GetListAsync(int orderType, List orderNos); - Task<(List list, long total)> GetListAsync(TrackDetailsQueryInput input); Task> GetListAsync(int orderId); diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 515afe91..a6c4af1f 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -12,11 +12,6 @@ public partial class GlobalConst /// public const string TrackUpsertForRefreshSubscribe = "SJZY.Track.TrackUpsertForRefresh"; - /// - /// 无法映射的轨迹新增 - /// - public const string TrackInsertForNotMappedEventSubscribe = "SJZY.Track.TrackInsertForNotMapped"; - /// /// 轨迹删除 /// @@ -36,6 +31,11 @@ public partial class GlobalConst /// 轨迹人工手动推送 /// public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush"; + + /// + /// 接口通交换机 + /// + public const string InterfaceRelayExchange = "sjzy.interface.relay.direct"; /// /// 轨迹推送完成（接口通回调） @@ -47,16 +47,6 @@ public partial class GlobalConst /// public const string TrackChangeAfterUpdateOrderSubscribe = "SJZY.Track.TrackChangeAfterUpdateOrder"; - /// - /// 轨迹推送后触发更新订单主表 - /// - public const string TrackPushAfterUpdateOrderSubscribe = "SJZY.Track.TrackPushAfterUpdateOrder"; - - /// - /// 关联订单后补轨迹 - /// - public const string OrderLinkAfterFillTrack = "SJZY.Track.OrderLinkAfterFillTrack"; - /// /// 推送预警任务 /// diff --git a/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs b/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs index 24743a5b..1d66b92f 100644 --- a/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs +++ b/src/SJZY.Track.Domain/Consts/InterfaceRelayConst.cs @@ -13,7 +13,7 @@ public class InterfaceRelayConst /// /// 获取轨迹接口代码 /// - public const string GetTrackInterfaceCode = "A012"; + public const string GetTrackInterfaceCode = "TEST"; /// /// 推送轨迹接口代码 diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 60043689..8ca8c9a6 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -122,7 +122,7 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, return await Context.Queryable() .RightJoin((a, b) => a.Id == b.OrderId) .LeftJoin((a, b, c) => b.ServiceProviderId == c.Id) - .Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull==1) + .Where((a, b) => b.ServiceProviderId == serviceProvider && !a.IsCompleted && a.IsTriggerPull == 1) .Select((a, b, c) => new OrderWithServiceProviderVO { OrderNo = a.OrderNo, @@ -136,12 +136,12 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, .Take(pageSize) .ToListAsync(); } - + public async Task> FindOrderWithServiceProviderAsync(int[] orderIds) { return await Context.Queryable() .InnerJoin((a, b) => a.Id == b.OrderId) - .Where((a, b) => orderIds.Contains( a.Id) && !a.IsCompleted) + .Where((a, b) => orderIds.Contains(a.Id) && !a.IsCompleted) .Select((a, b) => new OrderTransportDto { OrderId = a.Id, diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index cb82921e..c3f0b0a9 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -210,9 +210,9 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int orderType, List orderNos) + public async Task<(List results, object[]? nextSearchAfter)> GetListAsync(int orderType, List orderNos, object[]? searchAfter = null, int limit = 1000) { - if (orderNos.Count < 1) return new List(); + if (orderNos.Count < 1) return (new List(), null); var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd @@ -226,12 +226,26 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract f.TrackTimeZone) )) .Query(q => - q.Terms(t => t - .Field(f => f.OrderNo.Suffix("keyword")) - .Terms(orderNos)) - && q.Term(f => f.OrderType, orderType) + q.Bool(b => b + .Must(mu => mu + .Terms(t => t + .Field(f => f.OrderNo.Suffix("keyword")) + .Terms(orderNos) + ), + mu => mu + .Term(f => f.OrderType, orderType) + ) + .MustNot(mn => mn + .Term(t => t.TrackCode.Suffix("keyword"), GlobalConst.EmptyString) + ) + ) ) - .Size(1000) + .Sort(s => s + .Ascending(f => f.TrackTime) // 需要根据某个字段排序 + .Ascending(f => f.OrderNo) // 可以添加更多排序字段以确保唯一性 + ) + .Size(limit) + .SearchAfter(searchAfter) ); if (!r.IsValid) @@ -239,7 +253,10 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int orderId)