Merge branch 'S03-轨迹中台优化项' into dev

This commit is contained in:
李鹏军 2024-04-16 11:48:20 +08:00
commit a23e3ea9e7
12 changed files with 171 additions and 159 deletions

View File

@ -5,12 +5,7 @@ public class TrackChangeAfterUpdateOrderEvent
/// <summary> /// <summary>
/// 订单Id /// 订单Id
/// </summary> /// </summary>
public int[] OrderIds { get; set; } public int[] OrderIds { get; set; } = null!;
/// <summary>
/// 是否删除引起的更新
/// </summary>
public bool IsDelete { get; set; }
/// <summary> /// <summary>
/// 这条轨迹是否意味着完成(只有刷新会设为true) /// 这条轨迹是否意味着完成(只有刷新会设为true)

View File

@ -2,6 +2,11 @@
public class TrackPushEvent public class TrackPushEvent
{ {
/// <summary>
/// 重试次数
/// </summary>
public int RetryNum { get; set; }
/// <summary> /// <summary>
/// 操作类型 /// 操作类型
/// </summary> /// </summary>

View File

@ -20,9 +20,8 @@ public class CapFilter : SubscribeFilter
{ {
context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff"); context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff");
// 轨迹增删改要做幂等 // 轨迹增改要做幂等
if (!context.DeliverMessage.Headers["cap-msg-name"] if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe))
.IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackDeleteSubscribe))
return Task.CompletedTask; return Task.CompletedTask;
// 本地调试不做幂等 // 本地调试不做幂等
@ -44,7 +43,7 @@ public class CapFilter : SubscribeFilter
throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数:{argumentJsonStr}"); throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数:{argumentJsonStr}");
RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10, RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10,
TimeSpan.FromHours(1).Seconds); // 1h内不允许一模一样的参数重复消费 TimeSpan.FromMinutes(5).Seconds);
} }
} }
@ -76,11 +75,13 @@ public class CapFilter : SubscribeFilter
/// <returns></returns> /// <returns></returns>
public override Task OnSubscribeExceptionAsync(ExceptionContext context) public override Task OnSubscribeExceptionAsync(ExceptionContext context)
{ {
context.ExceptionHandled = true;
context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
if (context.Exception is BusinessException ex && ex.Code == ErrorCode) if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
{
context.ExceptionHandled = true; // 幂等不做重试
LogHelper.Warn(ex.ToString()); LogHelper.Warn(ex.ToString());
}
else else
LogHelper.Error($"cap-msg-id:{messageId}\n{context.Exception}"); LogHelper.Error($"cap-msg-id:{messageId}\n{context.Exception}");
@ -101,14 +102,14 @@ public class CapFilter : SubscribeFilter
var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture); var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture);
var useTime = DateTime.Now - parsedDate; var useTime = DateTime.Now - parsedDate;
if (useTime.Minutes > 20) if (useTime.Minutes > 10)
{ {
var log = $@" var log = $@"
|-消息编号:{messageId} |-消息编号:{messageId}
|-消费者名:{receiveName} |-消费者名:{receiveName}
|-跳过执行:{(isFromExecptionFliter ? "是" : "否")} |-跳过执行:{(isFromExecptionFliter ? "是" : "否")}
|-执行时间:{receiveTime} |-执行时间:{receiveTime}
|-消费耗时:{useTime.Milliseconds}毫秒 |-消费耗时:{useTime.Seconds}秒
|-请求参数:{arguments}"; |-请求参数:{arguments}";
LogHelper.Error(log); LogHelper.Error(log);
} }

View File

@ -36,36 +36,6 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
if (orders.Count == 0) return; if (orders.Count == 0) return;
// todo:delete可能发生回退,直接去es查最新轨迹,但是可能es还没有刷盘,导致回退失效,可以考虑加个定时任务弥补
if (@event.IsDelete)
{
foreach (var order in orders)
{
var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds);
var track = tracks.Where(x => x.OrderId == order.Id).MaxBy(x => x.TrackTime);
// 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) return;
// tips:订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
}
else
{
// 删除引发的回退更新
order.IsCompleted = @event.IsCompleted;
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
}
}
else // upsert可以直接计算
{
foreach (var order in orders) foreach (var order in orders)
{ {
if (order.LatestTrackTime != null && if (order.LatestTrackTime != null &&
@ -75,15 +45,17 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
continue; continue;
} }
// upsert引发的更新 if (!order.IsCompleted)
{
order.IsCompleted = @event.IsCompleted; order.IsCompleted = @event.IsCompleted;
}
order.LatestTrackContent = @event.TrackContent; order.LatestTrackContent = @event.TrackContent;
order.LatestTrackEnContent = @event.TrackEnContent; order.LatestTrackEnContent = @event.TrackEnContent;
order.LatestTrackPlace = @event.TrackPlace; order.LatestTrackPlace = @event.TrackPlace;
order.LatestTrackTime = @event.TrackTime; order.LatestTrackTime = @event.TrackTime;
order.LatestTrackTimeZone = @event.TrackTimeZone; order.LatestTrackTimeZone = @event.TrackTimeZone;
} }
}
await _orderRepository.UpdateRangeAsync(orders, x => new await _orderRepository.UpdateRangeAsync(orders, x => new
{ {

View File

@ -74,19 +74,50 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
private async Task DeleteTrackAsync(TrackDeleteEvent @event) private async Task DeleteTrackAsync(TrackDeleteEvent @event)
{ {
// 删除轨迹(删的是同子系统下的单,所以code是统一的) // 删除轨迹(删的是同子系统下的单,所以code是统一的)
var isOk = await _trackDetailsRepository.DeleteTrackAsync(@event.OrderIds, @event.TrackCodes); var tracks = await _trackDetailsRepository.GetLatestListAsync(@event.OrderIds, @event.TrackCodes);
if (isOk) await _trackDetailsRepository.DeleteTrackAsync(@event.OrderIds, @event.TrackCodes);
var orders = await _orderRepository.GetListAsync(x => @event.OrderIds.Contains(x.Id),
x => new OrderEntity
{ {
// 触发订单主表更新事件 Id = x.Id,
await _capPublisher.PublishDelayAsync(TimeSpan.FromSeconds(3), LatestTrackContent = x.LatestTrackContent,
GlobalConst.TrackChangeAfterUpdateOrderSubscribe, LatestTrackEnContent = x.LatestTrackEnContent,
new TrackChangeAfterUpdateOrderEvent LatestTrackPlace = x.LatestTrackPlace,
LatestTrackTime = x.LatestTrackTime,
LatestTrackTimeZone = x.LatestTrackTimeZone,
});
if (orders.Count == 0) return;
foreach (var order in orders)
{ {
OrderIds = @event.OrderIds, var track = tracks.FirstOrDefault(x => x.OrderId == order.Id);
TrackCode = @event.TrackCodes.JoinAsString(","),
IsDelete = true // 删除全部轨迹导致的重置行为
if (track == default)
{
if (order.LatestTrackTime == null) return;
// tips:订单是支持从已完成态到未完成态的
order.ResetTrackInfo();
}
else
{
// 删除引发的回退更新
order.LatestTrackContent = track.TrackContent;
order.LatestTrackEnContent = track.TrackEnContent;
order.LatestTrackPlace = track.TrackPlace;
order.LatestTrackTime = track.TrackTime;
order.LatestTrackTimeZone = track.TrackTimeZone;
}
}
await _orderRepository.UpdateRangeAsync(orders, x => new
{
x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
x.LatestTrackTimeZone
}); });
} }
}
} }

View File

@ -20,15 +20,17 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
private readonly ITrackPushCacheService _trackPushCacheService; private readonly ITrackPushCacheService _trackPushCacheService;
private readonly ICapPublisher _capPublisher;
public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy, public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
ITrackPushRecordsRepository trackPushRecordsRepository, ITrackPushRecordsRepository trackPushRecordsRepository,
IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService) IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher)
{ {
_httpClientProxy = httpClientProxy; _httpClientProxy = httpClientProxy;
_trackPushRecordsRepository = trackPushRecordsRepository; _trackPushRecordsRepository = trackPushRecordsRepository;
_orderRepository = orderRepository; _orderRepository = orderRepository;
_trackPushCacheService = trackPushCacheService; _trackPushCacheService = trackPushCacheService;
_capPublisher = capPublisher;
} }
[CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
@ -103,6 +105,19 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
throw new ArgumentOutOfRangeException(); throw new ArgumentOutOfRangeException();
} }
if (!log.IsSuccess)
{
if (@event.RetryNum < 3)
{
// 1, 2, 4min
await _capPublisher.PublishDelayAsync(
TimeSpan.FromSeconds((1 << @event.RetryNum) * 60),
GlobalConst.TrackPushSubscribe, @event);
@event.RetryNum++;
return;
}
}
// 往es 写推送日志 // 往es 写推送日志
await _trackPushRecordsRepository.AddAsync(log); await _trackPushRecordsRepository.AddAsync(log);
@ -169,7 +184,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
Extra = config.ExpandFieldsValue Extra = config.ExpandFieldsValue
}; };
var body = new TrackPushInput() var body = new TrackPushInput
{ {
OrderNo = @event.OrderNo, OrderNo = @event.OrderNo,
}; };

View File

@ -60,9 +60,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
}).ToArray(); }).ToArray();
nextEvent.TrackRefreshData = new TrackRefreshStruct nextEvent.TrackRefreshData = new TrackRefreshStruct
{ {
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString, ServiceProviderName = @event.TrackRefreshData.ServiceProviderName,
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent,
OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString, OriginTrackData = @event.TrackRefreshData.OriginTrackData,
// tips:一定要重置为false,父级订单的完成不会使得关联订单完成,关联单也必须由刷新来完成或手动设置完成!! // tips:一定要重置为false,父级订单的完成不会使得关联订单完成,关联单也必须由刷新来完成或手动设置完成!!
IsCompleted = false IsCompleted = false
}; };
@ -80,7 +80,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return; return;
} }
@event.TrackCode = mapCode; nextEvent.TrackCode = mapCode;
} }
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent);
@ -105,7 +105,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
if (track != null && @event.IsAdd!.Value) return; // 存在不让创建 if (track != null && @event.IsAdd!.Value) return; // 存在不让创建
} }
// step1、upsert es轨迹
if (track == default) if (track == default)
{ {
track = new TrackDetailsIndex track = new TrackDetailsIndex
@ -118,9 +117,9 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
TrackPlace = @event.TrackPlace, TrackPlace = @event.TrackPlace,
TrackTime = @event.TrackTime, TrackTime = @event.TrackTime,
TrackTimeZone = @event.TrackTimeZone, TrackTimeZone = @event.TrackTimeZone,
ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString,
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString,
OriginTrackData = @event.TrackRefreshData.OriginTrackData, OriginTrackData = @event.TrackRefreshData.OriginTrackData ?? GlobalConst.EmptyString,
IsCanDelete = false, IsCanDelete = false,
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
CreateTime = DateTime.Now CreateTime = DateTime.Now
@ -159,8 +158,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
var trackPushEvent = new TrackPushEvent var trackPushEvent = new TrackPushEvent
{ {
OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改, OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改,
OrderId = track.OrderId, OrderId = order.OrderId,
OrderNo = track.OrderNo,
TrackStatus = track.TrackStatus, TrackStatus = track.TrackStatus,
TrackCode = track.TrackCode, TrackCode = track.TrackCode,
TrackContent = track.TrackContent, TrackContent = track.TrackContent,
@ -170,22 +168,26 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
}; };
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
}
var firstTrack = insertTracks.FirstOrDefault();
if (firstTrack == null) return;
// step1、upsert es轨迹
await _trackDetailsRepository.AddManyAsync(insertTracks);
// step3、触发订单主表更新事件(异步) // step3、触发订单主表更新事件(异步)
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
new TrackChangeAfterUpdateOrderEvent new TrackChangeAfterUpdateOrderEvent
{ {
OrderIds = new[] { track.OrderId }, OrderIds = orderIds,
IsCompleted = @event.TrackRefreshData.IsCompleted, IsCompleted = @event.TrackRefreshData.IsCompleted,
TrackCode = track.TrackCode, TrackCode = firstTrack.TrackCode,
TrackContent = track.TrackContent, TrackContent = firstTrack.TrackContent,
TrackEnContent = track.TrackEnContent, TrackEnContent = firstTrack.TrackEnContent,
TrackPlace = track.TrackPlace, TrackPlace = firstTrack.TrackPlace,
TrackTime = track.TrackTime, TrackTime = firstTrack.TrackTime,
TrackTimeZone = track.TrackTimeZone TrackTimeZone = firstTrack.TrackTimeZone
}); });
} }
await _trackDetailsRepository.AddManyAsync(insertTracks);
}
} }

View File

@ -43,18 +43,16 @@ public class TrackPushService : ApplicationService
{ {
input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空 input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空
var orders = await _orderRepository.GetListAsync(x => var orderIds = await _orderRepository.GetArrayAsync(x =>
x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id);
{
x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId
});
if (orders.Count == 0) if (!orderIds.Any())
{ {
throw new BusinessException(message: "一个推送配置都不存在"); throw new BusinessException(message: "一个订单都不存在");
} }
var tracks = await _trackDetailsRepository.GetListAsync(input.OrderType, input.OrderNos, input.TrackCode); var tracks = await _trackDetailsRepository.GetListAsync(orderIds, input.TrackCode);
tracks = tracks.DistinctBy(x => x.OrderId).ToList();
if (tracks.Count == 0) if (tracks.Count == 0)
{ {

View File

@ -6,7 +6,8 @@ public interface ITrackDetailsRepository
{ {
Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode); Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode);
Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds); Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] eventOrderIds, List<string> trackCodes);
Task<bool> AnyAsync(int orderId, DateTime trackTime); Task<bool> AnyAsync(int orderId, DateTime trackTime);
@ -19,16 +20,8 @@ public interface ITrackDetailsRepository
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos); Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos);
Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode);
Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input); Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input);
Task<List<TrackDetailsIndex>> GetListAsync(int orderId); Task<List<TrackDetailsIndex>> GetListAsync(int orderId);
/// <summary> Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes);
/// 删除轨迹
/// </summary>
/// <param name="orderIds"></param>
/// <param name="trackCodes"></param>
/// <returns></returns>
Task<bool> DeleteTrackAsync(int[] orderIds, List<string> trackCodes);
} }

View File

@ -22,6 +22,9 @@ public interface IBasicQuery<T> where T : IEntity
Task<List<TResult>> GetListAsync<TResult>(Expression<Func<T, bool>> whereExpression, Task<List<TResult>> GetListAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression); Expression<Func<T, TResult>> resultExpression);
Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression);
/// <summary> /// <summary>
/// count /// count
/// </summary> /// </summary>

View File

@ -24,4 +24,10 @@ public class BasicQuery<T> : SimpleClient<T>, IBasicQuery<T> where T : class, IE
{ {
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToListAsync(); return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToListAsync();
} }
public async Task<TResult[]> GetArrayAsync<TResult>(Expression<Func<T, bool>> whereExpression,
Expression<Func<T, TResult>> resultExpression)
{
return await Context.Queryable<T>().Where(whereExpression).Select(resultExpression).ToArrayAsync();
}
} }

View File

@ -1,5 +1,4 @@
using Nest; using Nest;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ElasticSearch.Services; using SJZY.Expand.ElasticSearch.Services;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Consts;
@ -18,20 +17,23 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
{ {
var r = await Context.SearchAsync<TrackDetailsIndex>(s => s var r = await Context.SearchAsync<TrackDetailsIndex>(s => s
.Source() .Source()
.Query(q => .Query(q => q
q.Terms(t => t .Bool(b => b
.Field(f => f.OrderId) .Must(
.Terms(orderIds)) mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
&& q.Term(f => f.TrackCode.Suffix("keyword"), trackCode) mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
&& q.Term(f => f.ServiceProviderName.Suffix("keyword"), GlobalConst.EmptyString) )
)
) )
); );
return r.ToIdList(); return r.ToIdList();
} }
public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds) public async Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] orderIds, List<string> trackCodes)
{ {
var excludeTrackCodes = trackCodes.ToList();
excludeTrackCodes.Add(GlobalConst.EmptyString);
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd .Source(s => s.Includes(fd => fd
.Field(f => f.TrackEnContent) .Field(f => f.TrackEnContent)
@ -40,15 +42,35 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
.Field(f => f.TrackTime) .Field(f => f.TrackTime)
.Field(f => f.TrackTimeZone) .Field(f => f.TrackTimeZone)
.Field(f => f.OrderId) .Field(f => f.OrderId)
.Field(f => f.TrackCode)
)) ))
.Query(q => .Query(q => q
q.Terms(t => t .Bool(b => b
.Field(f => f.OrderId) .Must(
.Terms(orderIds)) mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
&& q.Term(f => f.ServiceProviderName.Suffix("keyword"), GlobalConst.EmptyString)
) )
.Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending)) .MustNot(mn => mn
.Terms(t => t
.Field(f => f.TrackCode.Suffix("keyword"))
.Terms(excludeTrackCodes)
)
)
)
)
.Size(0)
.Aggregations(a => a
.Terms("orders", t => t
.Field(f => f.OrderId)
.Aggregations(aa => aa
.TopHits("latest_track", th => th
.Size(1) .Size(1)
.Sort(so => so
.Field(f => f.ZeroTrackTime, SortOrder.Descending)
)
)
)
)
)
); );
return data.Documents.ToList(); return data.Documents.ToList();
@ -91,20 +113,18 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid) if (!r.IsValid)
{ {
LogHelper.Error($"轨迹更新失败,原因:{r.DebugInformation}。"); throw new Exception($"轨迹更新失败,原因:{r.DebugInformation}。");
} }
} }
public async Task AddManyAsync(List<TrackDetailsIndex> tracks) public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
{ {
if (!tracks.Any()) return;
// 按天写入es // 按天写入es
var r = await Context.IndexManyAsync(tracks, tracks.First().IndexName); var r = await Context.IndexManyAsync(tracks, tracks.First().IndexName);
if (!r.IsValid) if (!r.IsValid)
{ {
LogHelper.Error($"{tracks.Count}条轨迹新增失败,原因:{r.DebugInformation}。"); throw new Exception($"{tracks.Count}条轨迹新增失败,原因:{r.DebugInformation}。");
} }
} }
@ -134,33 +154,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
return data.Documents.ToList(); return data.Documents.ToList();
} }
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderType, List<string> orderNos, string trackCode)
{
if (orderNos.Count < 1) return new List<TrackDetailsIndex>();
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
.Source(s => s.Includes(fd => fd
.Field(f => f.OrderId)
.Field(f => f.OrderNo)
.Field(f => f.TrackStatus)
.Field(f => f.TrackCode)
.Field(f => f.TrackContent)
.Field(f => f.TrackPlace)
.Field(f => f.TrackTime)
.Field(f => f.TrackTimeZone)
))
.Query(q =>
q.Terms(t => t
.Field(f => f.OrderNo.Suffix("keyword"))
.Terms(orderNos))
&& q.Term(f => f.OrderType, orderType)
&& q.Term(f => f.TrackCode.Suffix("keyword"), trackCode)
)
);
return data.Documents.ToList();
}
public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId) public async Task<List<TrackDetailsIndex>> GetListAsync(int orderId)
{ {
var data = await Context.SearchAsync<TrackDetailsIndex>(x => x var data = await Context.SearchAsync<TrackDetailsIndex>(x => x
@ -168,12 +161,12 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
q.Term(t => t.OrderId, orderId) q.Term(t => t.OrderId, orderId)
) )
.Size(1000) .Size(1000)
.Sort(s => s.Field(f => f.TrackTime, SortOrder.Descending))); .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending)));
return data.Documents.ToList(); return data.Documents.ToList();
} }
public async Task<bool> DeleteTrackAsync(int[] orderIds, List<string> trackCodes) public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
{ {
var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
.Query( .Query(
@ -186,8 +179,6 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
if (!r.IsValid) if (!r.IsValid)
throw new Exception($"轨迹数据删除失败,原因:{r.DebugInformation}"); throw new Exception($"轨迹数据删除失败,原因:{r.DebugInformation}");
return r.Deleted > 0;
} }
public async Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input) public async Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input)