Merge branch 'refs/heads/S04轨迹预警' into test

This commit is contained in:
xiaolipro 2024-08-07 15:03:51 +08:00
commit ff43491704
4 changed files with 48 additions and 45 deletions

View File

@ -1,5 +1,6 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using SJZY.Expand.ABP.Core;
using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Consts;
@ -22,22 +23,27 @@ public class PushWarningEventHandler : ITransientDependency, ICapSubscribe
[CapSubscribe(GlobalConst.TrackPushWarning)] [CapSubscribe(GlobalConst.TrackPushWarning)]
public async Task Handle(PushWarningEvent @event) public async Task Handle(PushWarningEvent @event)
{ {
int total = 0, pageSize = @event.PageSize, waitTime = 1000; var hasProcess = await RedisHelper.Client.ExistsAsync(GlobalConst.TrackPushTrackWaringLockKey);
var dateTimes = Enumerable.Range(1, 12).Select(val => DateTime.Now.AddHours(-val)).ToList(); if (hasProcess)
foreach (var dateTime in dateTimes)
{ {
return;
}
await RedisHelper.Client.SetAsync(GlobalConst.TrackPushTrackWaringLockKey, 1, TimeSpan.FromHours(3));
int total = 0, pageSize = @event.PageSize, waitTime = 1000;
var rows = 0; var rows = 0;
var pageindex = 1; var pageindex = 0;
do do
{ {
rows = await _orderRepository.PushWarnOrdersAsync(pageSize, pageindex, dateTime); rows = await _orderRepository.PushWarnOrdersAsync(pageSize, pageindex);
pageindex++; pageindex++;
total += rows; total += rows;
await Task.Delay(waitTime); await Task.Delay(waitTime);
} while (rows != pageSize); } while (rows == pageSize);
}
await RedisHelper.Client.DelAsync(GlobalConst.TrackPushTrackWaringLockKey);
_logger.LogInformation(message: "本次预警扫描了:{total}个订单", total); _logger.LogInformation(message: "本次预警扫描了:{total}个订单", total);
} }
} }

View File

@ -35,7 +35,7 @@ public interface IOrderRepository : IBasicRepository<OrderEntity>
/// 更新推送预警单 /// 更新推送预警单
/// </summary> /// </summary>
/// <returns></returns> /// <returns></returns>
Task<int> PushWarnOrdersAsync(int pageSize, int pageindex, DateTime createTime); Task<int> PushWarnOrdersAsync(int pageSize, int pageindex);
/// <summary> /// <summary>
/// 更新缺失标记,时区缺失和发生地缺失标记,1表示时区,2表示发生地,3表示两种 /// 更新缺失标记,时区缺失和发生地缺失标记,1表示时区,2表示发生地,3表示两种

View File

@ -61,4 +61,9 @@ public partial class GlobalConst
/// 轨迹预警数据redis key前缀 /// 轨迹预警数据redis key前缀
/// </summary> /// </summary>
public const string TrackWarningKeyPrefix = "TrackWarning"; public const string TrackWarningKeyPrefix = "TrackWarning";
/// <summary>
/// 12小时推送预警轨迹锁
/// </summary>
public const string TrackPushTrackWaringLockKey = "TrackPushTrackWaringLock";
} }

View File

@ -1,6 +1,5 @@
using SJZY.Track.Domain; using SJZY.Track.Domain;
using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate;
using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
using SJZY.Track.Domain.Dtos.Order; using SJZY.Track.Domain.Dtos.Order;
namespace SJZY.Track.Repository.Repositories; namespace SJZY.Track.Repository.Repositories;
@ -81,37 +80,30 @@ AND `a`.`IsDeleted` = 0 AND `b`.`IsDeleted` = 0 limit @pageSize", new { pageSize
/// <param name="pageindex"></param> /// <param name="pageindex"></param>
/// <param name="createTime"></param> /// <param name="createTime"></param>
/// <returns></returns> /// <returns></returns>
public async Task<int> PushWarnOrdersAsync(int pageSize, int pageindex, DateTime createTime) public async Task<int> PushWarnOrdersAsync(int pageSize, int pageindex)
{ {
var ordersToUpdate = await Context.Queryable<OrderEntity>() var ordersToUpdate = await Context.Queryable<OrderEntity>()
.InnerJoin<TrackPushConfigEntity>((b, a) => a.SubSystemId == b.SubSystemId && a.CustomId == b.CustomId) .Where((b) => !b.IsCompleted
.Where((b, a) =>
!b.IsCompleted
&& b.QuestionType != 1 && b.QuestionType != 1
&& b.CreateTime < createTime
&& !a.IsDeleted
&& !b.IsDeleted) && !b.IsDeleted)
.Select((b, a) => new { Order = b }) .Select(b => new { b.LatestPushTime, b.Id, b.QuestionType })
.ToPageListAsync(pageindex, pageSize); .ToPageListAsync(pageindex, pageSize);
var latestPushTimeIds = new LinkedList<int>(); var latestPushTimeIds = new List<int>();
var questionZeroIds = new LinkedList<int>(); var questionZeroIds = new List<int>();
foreach (var item in ordersToUpdate) foreach (var item in ordersToUpdate)
{ {
var order = item.Order; var questionType = !item.LatestPushTime.HasValue || item.LatestPushTime < DateTime.Now.AddHours(-12)
var questionType = !order.LatestPushTime.HasValue
|| order.LatestPushTime < DateTime.Now.AddHours(-12)
? 2 ? 2
: 0; : 0;
if (order.QuestionType == questionType) continue; if (item.QuestionType == questionType) continue;
if (questionType == 2) latestPushTimeIds.AddLast(order.Id); if (questionType == 2) latestPushTimeIds.AddLast(item.Id);
else questionZeroIds.AddLast(order.Id); else questionZeroIds.AddLast(item.Id);
} }
int rows = 0; if (latestPushTimeIds.Count > 0) UpdateOrderQuestionType(2, latestPushTimeIds);
if (latestPushTimeIds.Count > 0) rows += UpdateOrderQuestionType(2, latestPushTimeIds); if (questionZeroIds.Count > 0) UpdateOrderQuestionType(0, questionZeroIds);
if (questionZeroIds.Count > 0) rows += UpdateOrderQuestionType(0, questionZeroIds); return ordersToUpdate.Count;
return rows;
} }
private int UpdateOrderQuestionType(int questionType, IEnumerable<int> ids) private int UpdateOrderQuestionType(int questionType, IEnumerable<int> ids)