diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/PushWarningEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/PushWarningEventHandler.cs index 103351c3..de99f575 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/PushWarningEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/PushWarningEventHandler.cs @@ -1,5 +1,6 @@ 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 SJZY.Track.Domain.Consts; @@ -22,22 +23,27 @@ public class PushWarningEventHandler : ITransientDependency, ICapSubscribe [CapSubscribe(GlobalConst.TrackPushWarning)] public async Task Handle(PushWarningEvent @event) { - int total = 0, pageSize = @event.PageSize, waitTime = 1000; - var dateTimes = Enumerable.Range(1, 12).Select(val => DateTime.Now.AddHours(-val)).ToList(); - - foreach (var dateTime in dateTimes) + var hasProcess = await RedisHelper.Client.ExistsAsync(GlobalConst.TrackPushTrackWaringLockKey); + if (hasProcess) { - var rows = 0; - var pageindex = 1; - do - { - rows = await _orderRepository.PushWarnOrdersAsync(pageSize, pageindex, dateTime); - pageindex++; - total += rows; - await Task.Delay(waitTime); - } while (rows != pageSize); + 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); } } \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs index 7f7c6d8f..773b2996 100644 --- a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs @@ -29,13 +29,13 @@ public interface IOrderRepository : IBasicRepository /// 更新推送预警单 /// /// - Task PushWarnOrdersAsync(int pageSize,DateTime createTime); + Task PushWarnOrdersAsync(int pageSize, DateTime createTime); /// /// 更新推送预警单 /// /// - Task PushWarnOrdersAsync(int pageSize, int pageindex, DateTime createTime); + Task PushWarnOrdersAsync(int pageSize, int pageindex); /// /// 更新缺失标记，时区缺失和发生地缺失标记，1表示时区，2表示发生地，3表示两种 diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs index 06972eff..4dec0225 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs @@ -21,44 +21,49 @@ public partial class GlobalConst /// 轨迹拉取配置信息redis key前缀 /// public const string TrackPullContentKeyPrefix = "TrackPullContent"; - + /// /// 预警通知redis key /// public const string ForewarningNotifyKey = "ForewarningNotify"; - + /// /// 预警通知剩余次数redis key /// public const string ForewarningNotifyCntKey = "ForewarningNotifyCnt"; - + /// /// kpi区域redis key前缀 /// public const string KpiAreaKeyPrefix = "KpiArea"; - + /// /// 城市时区redis key前缀 /// public const string CityTimezoneKeyPrefix = "CityTimezone"; - + /// /// 港口时区redis key前缀 /// public const string PortTimezoneKeyPrefix = "PortTimezone"; - + /// /// 发布订阅通知频道Key /// public const string PublishNotifyChannelKey = "PublishNotifyChannel"; - + /// /// 轨迹区域reids key /// public const string TrackAreaKeyPrefix = "TrackArea"; - + /// /// 轨迹预警数据redis key前缀 /// public const string TrackWarningKeyPrefix = "TrackWarning"; + + /// + /// 12小时推送预警轨迹锁 + /// + public const string TrackPushTrackWaringLockKey = "TrackPushTrackWaringLock"; } \ No newline at end of file diff --git a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs index 5460ce3c..5aca5329 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs @@ -1,6 +1,5 @@ using SJZY.Track.Domain; using SJZY.Track.Domain.Aggregates.OrderAggregate; -using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Dtos.Order; namespace SJZY.Track.Repository.Repositories; @@ -81,37 +80,30 @@ AND `a`.`IsDeleted` = 0 AND `b`.`IsDeleted` = 0 limit @pageSize", new { pageSize /// /// /// - public async Task PushWarnOrdersAsync(int pageSize, int pageindex, DateTime createTime) + public async Task PushWarnOrdersAsync(int pageSize, int pageindex) { var ordersToUpdate = await Context.Queryable() - .InnerJoin((b, a) => a.SubSystemId == b.SubSystemId && a.CustomId == b.CustomId) - .Where((b, a) => - !b.IsCompleted - && b.QuestionType != 1 - && b.CreateTime < createTime - && !a.IsDeleted - && !b.IsDeleted) - .Select((b, a) => new { Order = b }) + .Where((b) => !b.IsCompleted + && b.QuestionType != 1 + && !b.IsDeleted) + .Select(b => new { b.LatestPushTime, b.Id, b.QuestionType }) .ToPageListAsync(pageindex, pageSize); - var latestPushTimeIds = new LinkedList(); - var questionZeroIds = new LinkedList(); + var latestPushTimeIds = new List(); + var questionZeroIds = new List(); foreach (var item in ordersToUpdate) { - var order = item.Order; - var questionType = !order.LatestPushTime.HasValue - || order.LatestPushTime < DateTime.Now.AddHours(-12) + var questionType = !item.LatestPushTime.HasValue || item.LatestPushTime < DateTime.Now.AddHours(-12) ? 2 : 0; - if (order.QuestionType == questionType) continue; - if (questionType == 2) latestPushTimeIds.AddLast(order.Id); - else questionZeroIds.AddLast(order.Id); + if (item.QuestionType == questionType) continue; + if (questionType == 2) latestPushTimeIds.AddLast(item.Id); + else questionZeroIds.AddLast(item.Id); } - int rows = 0; - if (latestPushTimeIds.Count > 0) rows += UpdateOrderQuestionType(2, latestPushTimeIds); - if (questionZeroIds.Count > 0) rows += UpdateOrderQuestionType(0, questionZeroIds); - return rows; + if (latestPushTimeIds.Count > 0) UpdateOrderQuestionType(2, latestPushTimeIds); + if (questionZeroIds.Count > 0) UpdateOrderQuestionType(0, questionZeroIds); + return ordersToUpdate.Count; } private int UpdateOrderQuestionType(int questionType, IEnumerable ids)