fix:life-time possible dispose

This commit is contained in:
xiaolipro 2024-09-23 16:31:16 +08:00
parent d3aa525f9b
commit a07a95f19f
3 changed files with 143 additions and 131 deletions

View File

@ -9,6 +9,7 @@ using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Track2InterfaceRelay; using SJZY.Track.Application.Track2InterfaceRelay;
using SJZY.Track.Domain.Dtos.TrackPull; using SJZY.Track.Domain.Dtos.TrackPull;
using SJZY.Track.Domain.Queries; using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying; using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.Services.TimerTaskss; namespace SJZY.Track.Application.Services.TimerTaskss;
@ -17,20 +18,8 @@ namespace SJZY.Track.Application.Services.TimerTaskss;
/// 轨迹拉取定时任务 /// 轨迹拉取定时任务
/// </summary> /// </summary>
[ApiExplorerSettings(GroupName = "定时任务")] [ApiExplorerSettings(GroupName = "定时任务")]
public class TrackPullTask : ApplicationService public class TrackPullTask : ApplicationService, ISingletonDependency
{ {
private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
private readonly IProvideQueries _provideQueries;
private readonly ITrackPullCacheService _trackPullCacheService;
public TrackPullTask(IProvideQueries provideQueries, ITrackPullCacheService trackPullCacheService,
IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy)
{
_provideQueries = provideQueries;
_trackPullCacheService = trackPullCacheService;
_httpClientProxy = httpClientProxy;
}
/// <summary> /// <summary>
/// 定时任务刷新轨迹 /// 定时任务刷新轨迹
/// </summary> /// </summary>
@ -39,19 +28,25 @@ public class TrackPullTask : ApplicationService
[HttpPost("/TimerTask/Refresh")] [HttpPost("/TimerTask/Refresh")]
public async Task<bool> RefreshForTaskAsync() public async Task<bool> RefreshForTaskAsync()
{ {
var configs = await _provideQueries.GetAllPullConfigsAsync(); using var serviceScope = LazyServiceProvider.CreateScope();
var provideQueries = serviceScope.ServiceProvider.GetRequiredService<IProvideQueries>();
var configs = await provideQueries.GetAllPullConfigsAsync();
foreach (var config in configs) foreach (var config in configs)
{ {
await RefreshForConfigAsync(config); await RefreshForConfigAsync(serviceScope, config);
} }
await Task.Delay(1000); // life-time possible dispose
return true; 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<ITrackPullCacheService>();
var redisKey = trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderCode);
if (await RedisHelper.Client.ExistsAsync(redisKey)) if (await RedisHelper.Client.ExistsAsync(redisKey))
{ {
return; return;
@ -62,13 +57,29 @@ public class TrackPullTask : ApplicationService
_ = Task.Run(async () => _ = Task.Run(async () =>
{ {
try try
{
var provideQueries = serviceScope.ServiceProvider.GetRequiredService<IProvideQueries>();
var httpClientProxy = serviceScope.ServiceProvider
.GetRequiredService<IHttpClientProxy<IInterfaceForwardServiceV2>>();
await DoRefreshAsync(config, provideQueries, httpClientProxy);
}
catch (Exception e)
{
await RedisHelper.Client.DelAsync(redisKey);
LogHelper.Error(e.ToString());
}
});
}
private static async Task DoRefreshAsync(TrackPullConfigVo config, IProvideQueries provideQueries,
IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy)
{ {
using (Track2InterfaceRelaySubSystemContext.Use( using (Track2InterfaceRelaySubSystemContext.Use(
new Track2InterfaceRelaySubSystemContext(config.SubSystemCode))) new Track2InterfaceRelaySubSystemContext(config.SubSystemCode)))
{ {
var page = 1; var page = 1;
var orders = var orders =
await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode, page); await provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode, page);
while (orders.Any()) while (orders.Any())
{ {
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
@ -81,7 +92,7 @@ public class TrackPullTask : ApplicationService
})); }));
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderCode, var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderCode,
config.SubSystemId, group.Key)); config.SubSystemId, group.Key));
var res = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( var res = await httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach); GlobalConst.TrackRefreshSubscribe, attach);
@ -91,16 +102,9 @@ public class TrackPullTask : ApplicationService
} }
} }
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode, orders = await provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderCode,
++page); ++page);
} }
} }
} }
catch (Exception e)
{
await RedisHelper.Client.DelAsync(redisKey);
LogHelper.Error(e.ToString());
}
});
}
} }

View File

@ -13,15 +13,6 @@ namespace SJZY.Track.Application.Services.TimerTaskss;
[ApiExplorerSettings(GroupName = "定时任务")] [ApiExplorerSettings(GroupName = "定时任务")]
public class TrackPushTask : ApplicationService public class TrackPushTask : ApplicationService
{ {
private readonly IOptionsSnapshot<Track2InterfaceRelayOptions> _options;
private readonly IOrderRepository _orderRepository;
public TrackPushTask(IOrderRepository orderRepository, IOptionsSnapshot<Track2InterfaceRelayOptions> options)
{
_orderRepository = orderRepository;
_options = options;
}
/// <summary> /// <summary>
/// 定时任务推送预警(》12小时没有推送的) /// 定时任务推送预警(》12小时没有推送的)
/// </summary> /// </summary>
@ -30,20 +21,33 @@ public class TrackPushTask : ApplicationService
[HttpPost("/TimerTask/PushWarn")] [HttpPost("/TimerTask/PushWarn")]
public async Task<string> PushWarnForTaskAsync() public async Task<string> PushWarnForTaskAsync()
{ {
var configs = _options.Value; using var serviceScope = LazyServiceProvider.CreateScope();
var configs = serviceScope.ServiceProvider.GetRequiredService<IOptionsSnapshot<Track2InterfaceRelayOptions>>()
.Value;
if (configs == null || configs.Keys.Count == 0) if (configs == null || configs.Keys.Count == 0)
{ {
throw new BusinessException("未配置子系统"); throw new BusinessException("未配置子系统");
} }
var subSystemIds = await _orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); var orderRepository = serviceScope.ServiceProvider.GetRequiredService<IOrderRepository>();
var subSystemIds = await orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList());
foreach (var item in subSystemIds) foreach (var item in subSystemIds)
{ {
var subSystemId = item.id; 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}"; var redisKey = $"{GlobalConst.TrackPushWaringKeyPrefix}:{subSystemId}";
if (await RedisHelper.Client.ExistsAsync(redisKey)) if (await RedisHelper.Client.ExistsAsync(redisKey))
{ {
continue; return;
} }
await RedisHelper.Client.SetAsync(redisKey, 1, TimeSpan.FromHours(1)); await RedisHelper.Client.SetAsync(redisKey, 1, TimeSpan.FromHours(1));
@ -52,19 +56,8 @@ public class TrackPushTask : ApplicationService
{ {
try try
{ {
int total = 0, pageSize = 1000, waitTime = 1000; var orderRepository = serviceScope.ServiceProvider.GetRequiredService<IOrderRepository>();
int rows; await DoPushWarningAsync(orderRepository, subSystemId);
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) catch (Exception e)
{ {
@ -74,6 +67,20 @@ public class TrackPushTask : ApplicationService
}); });
} }
return "预警中"; 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}个订单");
} }
} }

View File

@ -14,18 +14,6 @@ namespace SJZY.Track.Application.Services.TimerTaskss;
[ApiExplorerSettings(GroupName = "定时任务")] [ApiExplorerSettings(GroupName = "定时任务")]
public class TrackWarningTask : ApplicationService public class TrackWarningTask : ApplicationService
{ {
private readonly IForewarningNotifyService _forewarningNotifyService;
private readonly IOptionsSnapshot<Track2InterfaceRelayOptions> _options;
private readonly IOrderRepository _orderRepository;
public TrackWarningTask(IForewarningNotifyService forewarningNotifyService, IOrderRepository orderRepository,
IOptionsSnapshot<Track2InterfaceRelayOptions> options)
{
_forewarningNotifyService = forewarningNotifyService;
_orderRepository = orderRepository;
_options = options;
}
/// <summary> /// <summary>
/// 轨迹运输时效预警通知 /// 轨迹运输时效预警通知
/// </summary> /// </summary>
@ -34,31 +22,47 @@ public class TrackWarningTask : ApplicationService
[HttpPost("/TimerTask/SendWarning")] [HttpPost("/TimerTask/SendWarning")]
public async Task<bool> SendWarningForTaskAsync() public async Task<bool> SendWarningForTaskAsync()
{ {
var configs = _options.Value; using var serviceScope = LazyServiceProvider.CreateScope();
var configs = serviceScope.ServiceProvider.GetRequiredService<IOptionsSnapshot<Track2InterfaceRelayOptions>>()
.Value;
if (configs == null || configs.Keys.Count == 0) if (configs == null || configs.Keys.Count == 0)
{ {
throw new BusinessException("未配置子系统"); throw new BusinessException("未配置子系统");
} }
var subSystemIds = await _orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); var orderRepository = serviceScope.ServiceProvider.GetRequiredService<IOrderRepository>();
var subSystemIds = await orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList());
foreach (var item in subSystemIds) foreach (var item in subSystemIds)
{ {
var subSystemId = item.id; 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}"; var redisKey = $"{GlobalConst.ForewarningNotifyCheckKey}:{subSystemId}";
if (await RedisHelper.Client.ExistsAsync(redisKey)) if (await RedisHelper.Client.ExistsAsync(redisKey))
{ {
continue; return;
} }
await RedisHelper.Client.SetExAsync(redisKey, configs[item.code].WarningNotifyInterval * 60, 1); await RedisHelper.Client.SetExAsync(redisKey, ttl, 1);
// todo:改mq分流? // todo:改mq分流?
_ = Task.Run(async () => _ = Task.Run(async () =>
{ {
try try
{ {
var cnt = await _forewarningNotifyService.SendNotificationAsync(subSystemId); var forewarningNotifyService =
serviceScope.ServiceProvider.GetRequiredService<IForewarningNotifyService>();
var cnt = await forewarningNotifyService.SendNotificationAsync(subSystemId);
LogHelper.Info($"预警通知发送成功,新增预警记录{cnt}条"); LogHelper.Info($"预警通知发送成功,新增预警记录{cnt}条");
} }
catch (Exception e) catch (Exception e)
@ -68,7 +72,4 @@ public class TrackWarningTask : ApplicationService
} }
}); });
} }
return true;
}
} }