ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

轨迹刷新算法更新

Browse Source
This commit is contained in:
xiaolipro 2024-09-10 09:43:33 +08:00
parent a52ddf7b8a
commit e78427521c
18 changed files with 245 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningNotifyCacheEntry.cs
View File

@ -14,6 +14,7 @@ public class WarningNotifyCacheEntry
var values = notifyValue.Split('_');
RuleId = int.Parse(values[0]);
OrderId = int.Parse(values[1]);
Counter = int.Parse(values[2]);
}
/// <summary>

2
src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs
View File

@ -18,5 +18,5 @@ public interface IForewarningNotifyService
/// <summary>
/// 发送通知。
/// </summary>
Task SendNotificationAsync(int subSystemId);
Task<int> SendNotificationAsync(int subSystemId);
}

26
src/SJZY.Track.Application/BackgroundServices/OrderPullStatusScanner.cs Normal file
View File

@ -0,0 +1,26 @@
using Microsoft.Extensions.Hosting;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
namespace SJZY.Track.Application.BackgroundServices;
public class OrderPullStatusScanner : BackgroundService
{
private readonly IOrderRepository _orderRepository;
public OrderPullStatusScanner(IOrderRepository orderRepository)
{
_orderRepository = orderRepository;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var count = await _orderRepository.UpdatePullStatusBySilenceAsync();
if (count > 0)
{
LogHelper.Info($"{nameof(OrderPullStatusScanner)}本次扫描更新了{count}条订单的拉取状态");
}
await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
}
}

48
src/SJZY.Track.Application/CAP/EventHandlers/PushWarningEventHandler.cs
View File

@ -1,48 +0,0 @@
using DotNetCore.CAP;
using Microsoft.Extensions.Logging;
using SJZY.Expand.ABP.Core;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.CAP.EventHandlers;
public class PushWarningEventHandler : ITransientDependency, ICapSubscribe
{
private readonly ILogger<PushWarningEventHandler> _logger;
private readonly IOrderRepository _orderRepository;
public PushWarningEventHandler(IOrderRepository orderRepository,
ILogger<PushWarningEventHandler> logger)
{
_orderRepository = orderRepository;
_logger = logger;
}
[CapSubscribe(GlobalConst.TrackPushWarning)]
public async Task Handle(PushWarningEvent @event)
{
var hasProcess = await RedisHelper.Client.ExistsAsync(GlobalConst.TrackPushTrackWaringLockKey);
if (hasProcess)
{
return;
}
await RedisHelper.Client.SetAsync(GlobalConst.TrackPushTrackWaringLockKey, 1, TimeSpan.FromHours(3));
int total = 0, pageSize = @event.PageSize, waitTime = 1000;
var rows = 0;
var pageindex = 0;
do
{
rows = await _orderRepository.PushWarnOrdersAsync(pageSize, pageindex);
pageindex++;
total += rows;
await Task.Delay(waitTime);
} while (rows == pageSize);
await RedisHelper.Client.DelAsync(GlobalConst.TrackPushTrackWaringLockKey);
_logger.LogInformation(message: "本次预警扫描了：{total}个订单", total);
}
}

11
src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -196,8 +196,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
if (canNextOrderIds.Count == 0) return;
// 批量add to es
await _trackDetailsEsRepository.AddManyAsync(insertTracks);
if (tracks.Any())
{
// 批量add to es
await _trackDetailsEsRepository.AddManyAsync(insertTracks);
// add track触发订单拉取状态更新
await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(),
@event.TrackCode);
}
// 触发订单主表更新事件（异步）
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,

7
src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs
View File

@ -99,7 +99,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep
await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成
}
public async Task SendNotificationAsync(int subSystemId)
public async Task<int> SendNotificationAsync(int subSystemId)
{
var redisKey = GetForewarningNotifySetKeyRedisKey(subSystemId);
var timestamp = GetTimestamp(DateTime.Now);
@ -108,7 +108,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep
if (set.IsNullOrEmpty())
{
LogHelper.Info("没有运输时效需要预警");
return;
return 0;
}
var entries = set.Select(x => new WarningNotifyCacheEntry(x)).ToList();
@ -142,6 +142,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep
CustomName = rule.CustomName,
RuleId = entry.RuleId,
Rule = rule.ToString(),
Counter = entry.Counter,
OrderId = entry.OrderId,
WarningStatus = WarningStatus.UnProcessed,
NotifyUser = rule.NotifyUser,
@ -170,6 +171,8 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep
}
await RedisHelper.Client.ZRemRangeByScoreAsync(redisKey, 0, timestamp);
return warningRecords.Count;
}
private string GetForewarningNotifySetKeyRedisKey(int subSystemId) =>

74
src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs
View File

@ -91,8 +91,12 @@ public class TrackPullService : ApplicationService
public async Task<bool> RefreshForTaskAsync()
{
var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
var tasks = configs.Select(RefreshForConfigAsync).ToArray();
await Task.WhenAll(tasks);
foreach (var config in configs)
{
await RefreshForConfigAsync(config);
}
return true;
}
@ -122,38 +126,50 @@ public class TrackPullService : ApplicationService
return;
}
var page = 1;
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
_ = Task.Run(async () =>
{
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
foreach (var group in orders.GroupBy(x => new
try
{
x.SubSystemId,
x.OrderType
}))
{
var data = new List<object>(group.Select(x => new TrackPullPayload
var page = 1;
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{
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);
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
if (res.Code != InterfaceRelayConst.SuccessCode)
{
LogHelper.Error($"刷新轨迹失败：{res.ToJson()}");
foreach (var group in orders.GroupBy(x => new
{
x.SubSystemId,
x.OrderType
}))
{
var data = new List<object>(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)
{
LogHelper.Error($"刷新轨迹失败：{res.ToJson()}");
}
}
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
}
}
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
}
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
catch (Exception e)
{
LogHelper.Error(e.ToString());
}
finally
{
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, 1);
}
});
}
}

39
src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs
View File

@ -1,5 +1,7 @@
using DotNetCore.CAP;
using Microsoft.AspNetCore.Authorization;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
@ -91,12 +93,43 @@ public class TrackPushService : ApplicationService
[HttpPost("Track/PushWarnForTask")]
public async Task<string> PushWarnForTaskAsync()
{
await _capPublisher.PublishAsync(GlobalConst.TrackPushWarning, new PushWarningEvent
var hasProcess = await RedisHelper.Client.ExistsAsync(GlobalConst.TrackPushTrackWaringLockKey);
if (hasProcess)
{
PageSize = 1000
return "冷却中";
}
await RedisHelper.Client.SetAsync(GlobalConst.TrackPushTrackWaringLockKey, 1, TimeSpan.FromHours(3));
_ = Task.Run(async () =>
{
try
{
int total = 0, pageSize = 1000, waitTime = 1000;
int rows;
var pageindex = 0;
do
{
rows = await _orderRepository.PushWarnOrdersAsync(pageSize, pageindex);
pageindex++;
total += rows;
await Task.Delay(waitTime);
} while (rows == pageSize);
LogHelper.Info($"本次预警扫描了：{total}个订单");
}
catch (Exception e)
{
LogHelper.Error(e.ToString());
}
finally
{
await RedisHelper.Client.DelAsync(GlobalConst.TrackPushTrackWaringLockKey);
}
});
return "推送预警任务已启动";
return "预警中";
}
/// <summary>

14
src/SJZY.Track.Application/Services/TrackOrders/OrderWarningService.cs
View File

@ -32,7 +32,6 @@ public class OrderWarningService
public async Task<UnifyPageResultDto<GetOrderWarningPageOutput>> GetOrderWarningPageListAsync(
GetOrderWarningPageListVo inputVo)
{
// todo：产品要的，排序字段，查询方式需要讨论
var result = await _warningRecordEsRepository.GetOrderWarningPageListAsync(inputVo);
var orderIds = result.list.Select(x => x.OrderId).ToList();
var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new
@ -52,6 +51,7 @@ public class OrderWarningService
var item = new GetOrderWarningPageOutput
{
Rule = x.Rule,
Counter = x.Counter,
WarningName = x.WarningName,
WarningStatus = x.WarningStatus,
CreateTime = x.CreateTime,
@ -95,21 +95,25 @@ public class OrderWarningService
continue;
}
// 异步发送通知
await RedisHelper.Client.SetExAsync(redisKey, 30 * 60, 1);
// todo：改mq分流？
_ = Task.Run(async () =>
{
try
{
await _forewarningNotifyService.SendNotificationAsync(item.id);
var cnt = await _forewarningNotifyService.SendNotificationAsync(item.id);
LogHelper.Info($"预警通知发送成功，新增预警记录{cnt}条");
}
catch (Exception e)
{
LogHelper.Error(e.ToString());
}
finally
{
await RedisHelper.Client.DelAsync(redisKey);
}
});
await RedisHelper.Client.SetExAsync(redisKey, configs[item.code] * 60, 1);
}
return true;

5
src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/WarningRecordIndex.cs
View File

@ -42,6 +42,11 @@ public class WarningRecordIndex : IdBaseIndex
/// </summary>
public string Rule { get; set; } = null!;
/// <summary>
/// 预警次数
/// </summary>
public int Counter { get; set; }
/// <summary>
/// 预警状态
/// </summary>

16
src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs
View File

@ -53,6 +53,22 @@ public interface IOrderRepository : IBasicRepository<OrderEntity>
/// <returns></returns>
Task<List<(string code, int id)>> GetSubSystemListByCodesAsync(List<string> codes);
/// <summary>
/// 更新订单拉取状态为静默，Init->SilenceBeforeStart，Pulling->SilenceAfterEnd
/// </summary>
/// <param name="orderIds"></param>
/// <param name="trackCode"></param>
/// <returns></returns>
Task UpdatePullStatusToSilenceAsync(List<int> orderIds, string trackCode);
/// <summary>
/// 更新订单拉取状态根据静默，SilenceBeforeStart->Pulling，SilenceAfterEnd->Completed
/// </summary>
/// <returns></returns>
Task<int> UpdatePullStatusBySilenceAsync();
#region 订单关联
/// <summary>

8
src/SJZY.Track.Domain/Aggregates/OrderAggregate/OrderTrackPullStatusEntity.cs
View File

@ -4,7 +4,7 @@ namespace SJZY.Track.Domain.Aggregates.OrderAggregate;
/// 订单轨迹拉取状态表（1：N）
/// </summary>
[SugarTable("t_order_pull_status")]
public class OrderPullStatusEntity : ICreateBaicEntity, IEntity
public class OrderPullStatusEntity : ICreateBaicEntity, IUpdateBasicEntity, IEntity
{
/// <summary>
/// 自增主键
@ -77,4 +77,10 @@ public class OrderPullStatusEntity : ICreateBaicEntity, IEntity
/// </summary>
[SugarColumn(ColumnName = "CreateTime")]
public DateTime CreateTime { get; set; }
[SugarColumn(ColumnName = "UpdateTime")]
public DateTime? UpdateTime { get; set; }
[SugarColumn(ColumnName = "UpdateUserId")]
public int? UpdateUserId { get; set; }
}

5
src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs
View File

@ -47,11 +47,6 @@ public partial class GlobalConst
/// </summary>
public const string TrackChangeAfterUpdateOrderSubscribe = "SJZY.Track.TrackChangeAfterUpdateOrder";
/// <summary>
/// 推送预警任务
/// </summary>
public const string TrackPushWarning = "SJZY.Track.TrackPushWarning";
/// <summary>
/// 预警计算任务
/// </summary>

5
src/SJZY.Track.Domain/Dtos/Forewarning/GetOrderWarningPageOutput.cs
View File

@ -19,6 +19,11 @@ public class GetOrderWarningPageOutput
/// </summary>
public string Rule { get; set; } = null!;
/// <summary>
/// 预警次数
/// </summary>
public int Counter { get; set; }
/// <summary>
/// 预警状态
/// </summary>

7
src/SJZY.Track.Domain/Dtos/Order/OrderWithServiceProviderVO.cs
View File

@ -1,3 +1,5 @@
using SJZY.Track.Domain.Aggregates.OrderAggregate;
namespace SJZY.Track.Domain.Dtos.Order;
public class OrderWithServiceProviderVO
@ -23,4 +25,9 @@ public class OrderWithServiceProviderVO
/// 子系统id
/// </summary>
public int SubSystemId { get; set; }
/// <summary>
/// 轨迹拉取状态
/// </summary>
public TrackPullStatus TrackPullStatus { get; set; }
}

6
src/SJZY.Track.Repository/Queries/ProvideQueries.cs
View File

@ -58,14 +58,16 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
return await Context.Queryable<OrderEntity>()
.RightJoin<OrderPullStatusEntity>((a, b) => a.Id == b.OrderId)
.LeftJoin<VServiceProviderEntity>((a, b, c) => b.ServiceProviderCode == c.Code)
.Where((a, b, c) => c.Id == serviceProviderId && !a.IsCompleted)
.Where((a, b, c) => c.Id == serviceProviderId && !a.IsCompleted
&& b.StartTime <= DateTime.Now && b.EndTime >= DateTime.Now)
.Select((a, b, c) => new OrderWithServiceProviderVO
{
OrderNo = a.OrderNo,
OrderType = a.OrderType,
ServiceProviderCode = c.Code,
ServiceProviderName = c.Name,
SubSystemId = a.SubSystemId
SubSystemId = a.SubSystemId,
TrackPullStatus = b.PullStatus
})
.Skip((pageIndex - 1) * pageSize)
.Take(pageSize)

72
src/SJZY.Track.Repository/Repositories/OrderRepository.cs
View File

@ -1,3 +1,4 @@
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Track.Domain;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Dtos.Order;
@ -120,9 +121,7 @@ public class OrderRepository : BasicRepository<OrderEntity>, IOrderRepository, I
public async Task<int> PushWarnOrdersAsync(int pageSize, int pageindex)
{
var ordersToUpdate = await Context.Queryable<OrderEntity>()
.Where(b => !b.IsCompleted
&& b.QuestionType != 1
&& !b.IsDeleted)
.Where(b => !b.IsCompleted && b.QuestionType != 1)
.Select(b => new
{
b.LatestPushTime,
@ -162,6 +161,73 @@ public class OrderRepository : BasicRepository<OrderEntity>, IOrderRepository, I
return res.Select(x => (x.Key, x.Id)).ToList();
}
public async Task UpdatePullStatusToSilenceAsync(List<int> orderIds, string trackCode)
{
// var status = await Context.Queryable<OrderPullStatusEntity>()
// .Where(x => orderIds.Contains(x.OrderId) && x.StartTrackCode == trackCode)
// .Where(x => x.PullStatus == TrackPullStatus.Init || x.PullStatus == TrackPullStatus.Pulling)
// .Select(x => new
// {
// x.Id,
// x.PullStatus,
// x.SilenceBeforeStart,
// x.SilenceAfterEnd,
// x.StartTime,
// x.EndTime
// })
// .ToListAsync();
//
// // 初始态-》静默前、拉取中-》静默后
// var changes = status.Select(x => new
// {
// x.Id,
// PullStatus = x.PullStatus == TrackPullStatus.Init
// ? x.SilenceBeforeStart > 0
// ? TrackPullStatus.SilenceBeforeStart
// : TrackPullStatus.Pulling
// : x.SilenceAfterEnd > 0
// ? TrackPullStatus.SilenceAfterEnd
// : TrackPullStatus.Completed,
// StartTime =
// x.PullStatus == TrackPullStatus.Init ? DateTime.Now.AddHours(x.SilenceBeforeStart) : x.StartTime,
// EndTime = x.PullStatus == TrackPullStatus.Pulling ? DateTime.Now.AddHours(x.SilenceAfterEnd) : x.EndTime
// });
// await Context.Updateable<OrderPullStatusEntity>(changes).ExecuteCommandAsync();
// 初始态-》静默前
await Context.Updateable<OrderPullStatusEntity>()
.SetColumns(x => x.PullStatus == TrackPullStatus.SilenceBeforeStart)
.SetColumns(x => x.StartTime == DateTime.Now.AddHours(x.SilenceBeforeStart))
.Where(x => orderIds.Contains(x.OrderId) && x.StartTrackCode == trackCode)
.Where(x => x.PullStatus == TrackPullStatus.Init)
.ExecuteCommandAsync();
// 拉取中-》静默后
await Context.Updateable<OrderPullStatusEntity>()
.SetColumns(x => x.PullStatus == TrackPullStatus.SilenceAfterEnd)
.SetColumns(x => x.EndTime == DateTime.Now.AddHours(x.SilenceAfterEnd)) // >0才有意义
.Where(x => orderIds.Contains(x.OrderId) && x.EndTrackCode == trackCode)
.Where(x => x.PullStatus == TrackPullStatus.Pulling)
.ExecuteCommandAsync();
}
public async Task<int> UpdatePullStatusBySilenceAsync()
{
var cnt1 = await Context.Updateable<OrderPullStatusEntity>()
.SetColumns(x => x.PullStatus == TrackPullStatus.Pulling)
.Where(x => x.PullStatus == TrackPullStatus.SilenceBeforeStart && x.StartTime > DateTime.Now)
.ExecuteCommandAsync();
LogHelper.Info($"{cnt1}条订单的拉取状态由SilenceBeforeStart更新为Pulling");
var cnt2 = await Context.Updateable<OrderPullStatusEntity>()
.SetColumns(x => x.PullStatus == TrackPullStatus.Completed)
.Where(x => x.PullStatus == TrackPullStatus.SilenceAfterEnd && x.EndTime < DateTime.Now)
.ExecuteCommandAsync();
LogHelper.Info($"{cnt2}条订单的拉取状态由SilenceAfterEnd更新为Completed");
return cnt1 + cnt2;
}
private int UpdateOrderQuestionType(int questionType, IEnumerable<int> ids)
{
var idsString = string.Join(",", ids);

1
src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs
View File

@ -152,7 +152,6 @@ internal class DetailsEsRepository : ElasticSearchContextAbstract<TrackDetailsIn
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
{
if (!tracks.Any()) return;
// 按天写入es
var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}";
tracks.ForEach(x => x.IndexName = indexName);