diff --git a/src/SJZY.Track.Application/Services/TimerTaskss/TrackPullTask.cs b/src/SJZY.Track.Application/Services/TimerTaskss/TrackPullTask.cs index dbab9348..5ddc1b3c 100644 --- a/src/SJZY.Track.Application/Services/TimerTaskss/TrackPullTask.cs +++ b/src/SJZY.Track.Application/Services/TimerTaskss/TrackPullTask.cs @@ -9,6 +9,7 @@ using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Track2InterfaceRelay; using SJZY.Track.Domain.Dtos.TrackPull; using SJZY.Track.Domain.Queries; +using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; namespace SJZY.Track.Application.Services.TimerTaskss; @@ -17,20 +18,8 @@ namespace SJZY.Track.Application.Services.TimerTaskss; /// 轨迹拉取定时任务 /// [ApiExplorerSettings(GroupName = "定时任务")] -public class TrackPullTask : ApplicationService +public class TrackPullTask : ApplicationService, ISingletonDependency { - private readonly IHttpClientProxy _httpClientProxy; - private readonly IProvideQueries _provideQueries; - private readonly ITrackPullCacheService _trackPullCacheService; - - public TrackPullTask(IProvideQueries provideQueries, ITrackPullCacheService trackPullCacheService, - IHttpClientProxy httpClientProxy) - { - _provideQueries = provideQueries; - _trackPullCacheService = trackPullCacheService; - _httpClientProxy = httpClientProxy; - } - /// /// 定时任务刷新轨迹 /// @@ -39,19 +28,25 @@ public class TrackPullTask : ApplicationService [HttpPost("/TimerTask/Refresh")] public async Task RefreshForTaskAsync() { - var configs = await _provideQueries.GetAllPullConfigsAsync(); + using var serviceScope = LazyServiceProvider.CreateScope(); + + var provideQueries = serviceScope.ServiceProvider.GetRequiredService(); + var configs = await provideQueries.GetAllPullConfigsAsync(); foreach (var config in configs) { - await RefreshForConfigAsync(config); + await RefreshForConfigAsync(serviceScope, config); } + await Task.Delay(1000); // life-time possible dispose + return true; } - private async Task RefreshForConfigAsync(TrackPullConfigVo config) + private async Task RefreshForConfigAsync(IServiceScope serviceScope, TrackPullConfigVo config) { - var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderCode); + var trackPullCacheService = LazyServiceProvider.LazyGetRequiredService(); + var redisKey = trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderCode); if (await RedisHelper.Client.ExistsAsync(redisKey)) { return; @@ -63,38 +58,10 @@ public class TrackPullTask : ApplicationService { try { - using (Track2InterfaceRelaySubSystemContext.Use( - new Track2InterfaceRelaySubSystemContext(config.SubSystemCode))) - { - var page = 1; - var orders = - await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode, page); - while (orders.Any()) - { - var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; - - foreach (var group in orders.GroupBy(x => x.OrderType)) - { - var data = new List(group.Select(x => new TrackPullPayload - { - OrderNo = x.OrderNo - })); - var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderCode, - config.SubSystemId, group.Key)); - 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.ServiceProviderCode, - ++page); - } - } + var provideQueries = serviceScope.ServiceProvider.GetRequiredService(); + var httpClientProxy = serviceScope.ServiceProvider + .GetRequiredService>(); + await DoRefreshAsync(config, provideQueries, httpClientProxy); } catch (Exception e) { @@ -103,4 +70,41 @@ public class TrackPullTask : ApplicationService } }); } + + private static async Task DoRefreshAsync(TrackPullConfigVo config, IProvideQueries provideQueries, + IHttpClientProxy httpClientProxy) + { + using (Track2InterfaceRelaySubSystemContext.Use( + new Track2InterfaceRelaySubSystemContext(config.SubSystemCode))) + { + var page = 1; + var orders = + await provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode, page); + while (orders.Any()) + { + var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; + + foreach (var group in orders.GroupBy(x => x.OrderType)) + { + var data = new List(group.Select(x => new TrackPullPayload + { + OrderNo = x.OrderNo + })); + var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderCode, + config.SubSystemId, group.Key)); + 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.ServiceProviderCode, + ++page); + } + } + } } diff --git a/src/SJZY.Track.Application/Services/TimerTaskss/TrackPushTask.cs b/src/SJZY.Track.Application/Services/TimerTaskss/TrackPushTask.cs index 161283fd..928b54e9 100644 --- a/src/SJZY.Track.Application/Services/TimerTaskss/TrackPushTask.cs +++ b/src/SJZY.Track.Application/Services/TimerTaskss/TrackPushTask.cs @@ -13,15 +13,6 @@ namespace SJZY.Track.Application.Services.TimerTaskss; [ApiExplorerSettings(GroupName = "定时任务")] public class TrackPushTask : ApplicationService { - private readonly IOptionsSnapshot _options; - private readonly IOrderRepository _orderRepository; - - public TrackPushTask(IOrderRepository orderRepository, IOptionsSnapshot options) - { - _orderRepository = orderRepository; - _options = options; - } - /// /// 定时任务推送预警（》12小时没有推送的） /// @@ -30,50 +21,66 @@ public class TrackPushTask : ApplicationService [HttpPost("/TimerTask/PushWarn")] public async Task PushWarnForTaskAsync() { - var configs = _options.Value; + using var serviceScope = LazyServiceProvider.CreateScope(); + + var configs = serviceScope.ServiceProvider.GetRequiredService>() + .Value; if (configs == null || configs.Keys.Count == 0) { throw new BusinessException("未配置子系统"); } - var subSystemIds = await _orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); + var orderRepository = serviceScope.ServiceProvider.GetRequiredService(); + var subSystemIds = await orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); foreach (var item in subSystemIds) { - var subSystemId = item.id; - var redisKey = $"{GlobalConst.TrackPushWaringKeyPrefix}:{subSystemId}"; - if (await RedisHelper.Client.ExistsAsync(redisKey)) - { - continue; - } - - await RedisHelper.Client.SetAsync(redisKey, 1, TimeSpan.FromHours(1)); - - _ = Task.Run(async () => - { - try - { - int total = 0, pageSize = 1000, waitTime = 1000; - int rows; - var pageindex = 0; - - do - { - rows = await _orderRepository.PushWarnOrdersAsync(subSystemId, pageSize, pageindex); - pageindex++; - total += rows; - await Task.Delay(waitTime); - } while (rows == pageSize); - - LogHelper.Info($"本次预警扫描了：{total}个订单"); - } - catch (Exception e) - { - await RedisHelper.Client.DelAsync(redisKey); - LogHelper.Error(e.ToString()); - } - }); + await PushWarningBySubSystemAsync(serviceScope, item.id); } + await Task.Delay(1000); // life-time possible dispose + return "预警中"; } + + private async Task PushWarningBySubSystemAsync(IServiceScope serviceScope, int subSystemId) + { + var redisKey = $"{GlobalConst.TrackPushWaringKeyPrefix}:{subSystemId}"; + if (await RedisHelper.Client.ExistsAsync(redisKey)) + { + return; + } + + await RedisHelper.Client.SetAsync(redisKey, 1, TimeSpan.FromHours(1)); + + _ = Task.Run(async () => + { + try + { + var orderRepository = serviceScope.ServiceProvider.GetRequiredService(); + await DoPushWarningAsync(orderRepository, subSystemId); + } + catch (Exception e) + { + await RedisHelper.Client.DelAsync(redisKey); + LogHelper.Error(e.ToString()); + } + }); + } + + private async Task DoPushWarningAsync(IOrderRepository orderRepository, int subSystemId) + { + int total = 0, pageSize = 1000, waitTime = 1000; + int rows; + var pageindex = 0; + + do + { + rows = await orderRepository.PushWarnOrdersAsync(subSystemId, pageSize, pageindex); + pageindex++; + total += rows; + await Task.Delay(waitTime); + } while (rows == pageSize); + + LogHelper.Info($"本次预警扫描了：{total}个订单"); + } } diff --git a/src/SJZY.Track.Application/Services/TimerTaskss/TrackWarningTask.cs b/src/SJZY.Track.Application/Services/TimerTaskss/TrackWarningTask.cs index dbbbfcad..341db4fe 100644 --- a/src/SJZY.Track.Application/Services/TimerTaskss/TrackWarningTask.cs +++ b/src/SJZY.Track.Application/Services/TimerTaskss/TrackWarningTask.cs @@ -14,18 +14,6 @@ namespace SJZY.Track.Application.Services.TimerTaskss; [ApiExplorerSettings(GroupName = "定时任务")] public class TrackWarningTask : ApplicationService { - private readonly IForewarningNotifyService _forewarningNotifyService; - private readonly IOptionsSnapshot _options; - private readonly IOrderRepository _orderRepository; - - public TrackWarningTask(IForewarningNotifyService forewarningNotifyService, IOrderRepository orderRepository, - IOptionsSnapshot options) - { - _forewarningNotifyService = forewarningNotifyService; - _orderRepository = orderRepository; - _options = options; - } - /// /// 轨迹运输时效预警通知 /// @@ -34,41 +22,54 @@ public class TrackWarningTask : ApplicationService [HttpPost("/TimerTask/SendWarning")] public async Task SendWarningForTaskAsync() { - var configs = _options.Value; + using var serviceScope = LazyServiceProvider.CreateScope(); + + var configs = serviceScope.ServiceProvider.GetRequiredService>() + .Value; if (configs == null || configs.Keys.Count == 0) { throw new BusinessException("未配置子系统"); } - var subSystemIds = await _orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); + var orderRepository = serviceScope.ServiceProvider.GetRequiredService(); + var subSystemIds = await orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); foreach (var item in subSystemIds) { - var subSystemId = item.id; - // 如果已经发送过通知，不再发送 - var redisKey = $"{GlobalConst.ForewarningNotifyCheckKey}:{subSystemId}"; - if (await RedisHelper.Client.ExistsAsync(redisKey)) - { - continue; - } - - await RedisHelper.Client.SetExAsync(redisKey, configs[item.code].WarningNotifyInterval * 60, 1); - - // todo：改mq分流？ - _ = Task.Run(async () => - { - try - { - var cnt = await _forewarningNotifyService.SendNotificationAsync(subSystemId); - LogHelper.Info($"预警通知发送成功，新增预警记录{cnt}条"); - } - catch (Exception e) - { - await RedisHelper.Client.DelAsync(redisKey); - LogHelper.Error(e.ToString()); - } - }); + var ttl = configs[item.code].WarningNotifyInterval * 60; + await TrackWarningBySubSystemAsync(serviceScope, item.id, ttl); } + await Task.Delay(1000); // life-time possible dispose + return true; } + + private async Task TrackWarningBySubSystemAsync(IServiceScope serviceScope, int subSystemId, int ttl) + { + // 如果已经发送过通知，不再发送 + var redisKey = $"{GlobalConst.ForewarningNotifyCheckKey}:{subSystemId}"; + if (await RedisHelper.Client.ExistsAsync(redisKey)) + { + return; + } + + await RedisHelper.Client.SetExAsync(redisKey, ttl, 1); + + // todo：改mq分流？ + _ = Task.Run(async () => + { + try + { + var forewarningNotifyService = + serviceScope.ServiceProvider.GetRequiredService(); + var cnt = await forewarningNotifyService.SendNotificationAsync(subSystemId); + LogHelper.Info($"预警通知发送成功，新增预警记录{cnt}条"); + } + catch (Exception e) + { + await RedisHelper.Client.DelAsync(redisKey); + LogHelper.Error(e.ToString()); + } + }); + } }