diff --git a/src/TrackCenter.Api/appsettings.Development.json b/src/TrackCenter.Api/appsettings.Development.json index a6e86ace..01257b2b 100644 --- a/src/TrackCenter.Api/appsettings.Development.json +++ b/src/TrackCenter.Api/appsettings.Development.json @@ -4,5 +4,8 @@ "Default": "Debug", "Microsoft.AspNetCore": "Warning" } + }, + "FeiShuNotice": { + "EnableFeiShuExceptionSubscribe": false } } diff --git a/src/TrackCenter.Application/HostServices/TrackTaskComponentHostedService.cs b/src/TrackCenter.Application/HostServices/TrackTaskComponentHostedService.cs index bd164bbb..51399afe 100644 --- a/src/TrackCenter.Application/HostServices/TrackTaskComponentHostedService.cs +++ b/src/TrackCenter.Application/HostServices/TrackTaskComponentHostedService.cs @@ -13,11 +13,16 @@ public class TrackTaskComponentHostedService : IHostedService private readonly IOptionsMonitor _optionsMonitor; private TrackAppOptions? _oldOptions; private readonly ILogger _logger; + private readonly IRecurringJobManager _recurringJobManager; - public TrackTaskComponentHostedService(IOptionsMonitor optionsMonitor, ILogger logger) + public TrackTaskComponentHostedService( + IOptionsMonitor optionsMonitor, + ILogger logger, + IRecurringJobManager recurringJobManager) { _optionsMonitor = optionsMonitor; _logger = logger; + _recurringJobManager = recurringJobManager; var onChanged = DebounceHelper.WrapMethod(TimeSpan.FromSeconds(1), () => { ReloadConfig(optionsMonitor); }); optionsMonitor.OnChange(_ => onChanged.Debounce()); } @@ -43,7 +48,7 @@ public class TrackTaskComponentHostedService : IHostedService if ((_oldOptions?.TrackPullTaskMinuteInterval ?? 0) != newVal.TrackPullTaskMinuteInterval) { var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.TrackPullTaskLevel]; - RecurringJob.AddOrUpdate("Execute_TrackPullTask", + _recurringJobManager.AddOrUpdate("Execute_TrackPullTask", q, x => x.RefreshForTaskAsync(), $"*/{newVal.TrackPullTaskMinuteInterval} * * * *"); diff --git a/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs b/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs index 621f1f5e..6c49d638 100644 --- a/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs +++ b/src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs @@ -17,11 +17,16 @@ public class WarningTaskComponentHostedService : IHostedService private readonly IOptionsMonitor _optionsMonitor; private TrackWarningTaskOptions? _oldOptions; private readonly ILogger _logger; + private readonly IRecurringJobManager _recurringJobManager; - public WarningTaskComponentHostedService(IOptionsMonitor optionsMonitor, ILogger logger) + public WarningTaskComponentHostedService( + IOptionsMonitor optionsMonitor, + ILogger logger, + IRecurringJobManager recurringJobManager) { _optionsMonitor = optionsMonitor; _logger = logger; + _recurringJobManager = recurringJobManager; // 防抖配置变更包装 var onChanged = DebounceHelper.WrapMethod(TimeSpan.FromSeconds(1), () => { ReloadConfig(optionsMonitor); }); optionsMonitor.OnChange(_ => onChanged.Debounce()); @@ -50,7 +55,7 @@ public class WarningTaskComponentHostedService : IHostedService { var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanWarningNotifyTaskLevel]; // 预警通知任务扫描 - RecurringJob.AddOrUpdate("Scan_WarningNotifyTask", + _recurringJobManager.AddOrUpdate("Scan_WarningNotifyTask", q, x => x.ScanNotifyTaskAndSchedulingAsync(), $"0 */{newVal.ScanWarningNotifyTaskHourInterval} * * *"); @@ -60,7 +65,7 @@ public class WarningTaskComponentHostedService : IHostedService { var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanReturnWarningRetryPushTaskLevel]; // 回传预警通知重试推送任务扫描 - RecurringJob.AddOrUpdate("Scan_ReturnWarningRetryPushTask", + _recurringJobManager.AddOrUpdate("Scan_ReturnWarningRetryPushTask", q, x => x.ScanReturnRetryPushTaskAndSchedulingAsync(), $"*/{newVal.ScanReturnWarningRetryPushTaskMinuteInterval} * * * *"); @@ -70,7 +75,7 @@ public class WarningTaskComponentHostedService : IHostedService { var q = GlobalConst.TaskQueueLevelDict[_optionsMonitor.CurrentValue.ScanTransportWarningDetailsGenerateTaskLevel]; // 运输预警明细生成任务扫描 - RecurringJob.AddOrUpdate("Scan_TransportWarningDetailsGenerateTask", + _recurringJobManager.AddOrUpdate("Scan_TransportWarningDetailsGenerateTask", q, x => x.ScanTransportDetailsGenerateTaskAndSchedulingAsync(), $"*/{newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval} * * * *"); diff --git a/src/TrackCenter.Application/Options/TrackAppOptions.cs b/src/TrackCenter.Application/Options/TrackAppOptions.cs index 4d292d86..f6fd06a5 100644 --- a/src/TrackCenter.Application/Options/TrackAppOptions.cs +++ b/src/TrackCenter.Application/Options/TrackAppOptions.cs @@ -31,4 +31,9 @@ public class TrackAppOptions /// 拉取优先级 /// public int TrackPullTaskLevel { get; set; } + + /// + /// Hangfire 工作线程数 + /// + public int HangfireWorkerCount { get; set; } } diff --git a/src/TrackCenter.Application/Queries/ProvideQueries.cs b/src/TrackCenter.Application/Queries/ProvideQueries.cs index d369b379..df3fa14a 100644 --- a/src/TrackCenter.Application/Queries/ProvideQueries.cs +++ b/src/TrackCenter.Application/Queries/ProvideQueries.cs @@ -3,6 +3,7 @@ using SqlSugar; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackContentAggregate; using TrackCenter.Domain.Aggregates.TrackFlagAggregate; +using TrackCenter.Domain.Aggregates.TrackPullAggregate; using TrackCenter.Domain.Aggregates.TrackStatusAggregate; using TrackCenter.Domain.Dtos.Order; using TrackCenter.Domain.Dtos.TrackContents; @@ -51,14 +52,17 @@ public class ProvideQueries : SqlSugarRootlessRepository, { return await Context.Queryable() .InnerJoin((a, b) => a.Id == b.OrderId) - .LeftJoin((a, b, c) => a.AppKey == c.AppKey) + .LeftJoin((a, b, c) => + a.AppKey == c.AppKey && a.ServiceProviderCode == c.ServiceProviderCode) + .LeftJoin((a, b, c, d) => c.Id == d.TrackPullConfigId && a.Customer == d.Customer) .WhereIF(!appKey.IsNullOrWhiteSpace(), a => a.AppKey == appKey) .Where((a, b) => !a.IsCompleted && orderNos.Contains(a.OrderNo) && b.StartTime <= DateTime.Now && b.EndTime >= DateTime.Now) - .Select((a, b, c) => new OrderWithServiceProviderVO + .Select((a, b, c, d) => new OrderWithServiceProviderVO { - //优先使用第三方的订单号,为空时才使用主单号 - OrderNo = string.IsNullOrWhiteSpace(b.OrderNo) ? a.OrderNo : b.OrderNo, + OrderNo = d.OrderNoType == TrackOrderNoType.客户单号 ? a.OrderNo : + d.OrderNoType == TrackOrderNoType.跟踪单号 ? a.TrackNo : + d.OrderNoType == TrackOrderNoType.其他单号 ? a.ThirdPartyOrderNumber : a.OrderNo, OrderType = a.OrderType, AppKey = a.AppKey, ServiceProviderCode = b.ServiceProviderCode, @@ -81,9 +85,9 @@ public class ProvideQueries : SqlSugarRootlessRepository, .WhereIF(config.Customer != GlobalConst.All, a => a.Customer == config.Customer) .Select((a, b) => new OrderWithServiceProviderVO { - //优先使用第三方的订单号,为空时才使用主单号 OrderNo = config.OrderNoType == TrackOrderNoType.客户单号 ? a.OrderNo : - config.OrderNoType == TrackOrderNoType.跟踪单号 ? a.TrackNo : a.ThirdPartyOrderNumber, + config.OrderNoType == TrackOrderNoType.跟踪单号 ? a.TrackNo : + config.OrderNoType == TrackOrderNoType.其他单号 ? a.ThirdPartyOrderNumber : a.OrderNo, OrderType = a.OrderType, AppKey = a.AppKey, ServiceProviderCode = b.ServiceProviderCode diff --git a/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs b/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs index 559d26c9..9a418b9d 100644 --- a/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs +++ b/src/TrackCenter.Application/Services/TimerTaskss/TrackPullTask.cs @@ -1,6 +1,10 @@ +using Hangfire; +using Hangfire.Common; +using Hangfire.States; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.AspNetCore.Authorization; using Microsoft.Extensions.Options; +using Fake.UnitOfWork; using TrackCenter.Application.Dtos.TrackPull; using TrackCenter.Application.Helpers; using TrackCenter.Application.Options; @@ -16,27 +20,25 @@ namespace TrackCenter.Application.Services.TimerTaskss; /// 轨迹拉取定时任务 /// [ApiExplorerSettings(GroupName = "定时任务")] -public class TrackPullTask : ApplicationServiceBase, ISingletonDependency +public class TrackPullTask( + IBackgroundJobClient backgroundJobClient, + ITrackPullConfigRepository trackPullConfigRepository, + IOptionsMonitor trackAppOptions, + TrackPullCacheService trackPullCacheService, + ProvideQueries provideQueries, + InterfaceForwardService interfaceForwardService, + IOptions interfaceOptions) + : ApplicationServiceBase { - private readonly IServiceScopeFactory _serviceScopeFactory; - - public TrackPullTask(IServiceScopeFactory serviceScopeFactory) - { - _serviceScopeFactory = serviceScopeFactory; - } - /// /// 定时任务刷新轨迹 /// /// - [AllowAnonymous] + [DisableConcurrentExecution(timeoutInSeconds: 1800)] + [UnitOfWork] [HttpPost("/TimerTask/Refresh")] - public async Task RefreshForTaskAsync() + public virtual async Task RefreshForTaskAsync() { - using var serviceScope = _serviceScopeFactory.CreateScope(); - - var trackPullConfigRepository = serviceScope.ServiceProvider.GetRequiredService(); - var configs = await trackPullConfigRepository.AsQueryable() .LeftJoin((a, b) => a.Id == b.TrackPullConfigId) .Select((a, b) => new TrackPullConfigVo @@ -51,18 +53,22 @@ public class TrackPullTask : ApplicationServiceBase, ISingletonDependency }) .ToListAsync(); + if (configs.Count == 0) return true; + + var queue = GlobalConst.TaskQueueLevelDict[trackAppOptions.CurrentValue.TrackPullTaskLevel]; foreach (var config in configs) { - await RefreshForConfigAsync(serviceScope, config); + backgroundJobClient.Create( + Job.FromExpression(x => x.RefreshForConfigAsync(config)), + new EnqueuedState(queue)); } - await Task.Delay(1000); // life-time possible dispose - return true; } - private async Task RefreshForConfigAsync(IServiceScope serviceScope, TrackPullConfigVo config) + + [UnitOfWork] + public virtual async Task RefreshForConfigAsync(TrackPullConfigVo config) { - var trackPullCacheService = LazyServiceProvider.GetRequiredService(); var redisKey = trackPullCacheService.TrackPullKey(config); if (await RedisHelper.Client.ExistsAsync(redisKey)) { @@ -71,26 +77,19 @@ public class TrackPullTask : ApplicationServiceBase, ISingletonDependency await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, 1); - _ = Task.Run(async () => + try { - try - { - var provideQueries = serviceScope.ServiceProvider.GetRequiredService(); - var interfaceForward = serviceScope.ServiceProvider - .GetRequiredService(); - var interfaceOptions = serviceScope.ServiceProvider - .GetRequiredService>(); - await DoRefreshAsync(config, provideQueries, interfaceForward, interfaceOptions.Value); - } - catch (Exception e) - { - await RedisHelper.Client.DelAsync(redisKey); - Log.Error(e.ToString()); - } - }); + await DoRefreshAsync(config, provideQueries, interfaceForwardService, interfaceOptions.Value); + } + catch (Exception e) + { + await RedisHelper.Client.DelAsync(redisKey); + Log.Error(e.ToString()); + throw; + } } - private static async Task DoRefreshAsync(TrackPullConfigVo config, ProvideQueries provideQueries, + protected async Task DoRefreshAsync(TrackPullConfigVo config, ProvideQueries provideQueries, InterfaceForwardService interfaceForward, InterfaceRelayParamsConfigOptions interfaceOptionsValue) { diff --git a/src/TrackCenter.Application/TrackCenterApplicationModule.cs b/src/TrackCenter.Application/TrackCenterApplicationModule.cs index 93a5175a..9d2be4a5 100644 --- a/src/TrackCenter.Application/TrackCenterApplicationModule.cs +++ b/src/TrackCenter.Application/TrackCenterApplicationModule.cs @@ -4,7 +4,6 @@ using DotNetCore.CAP; using DotNetCore.CAP.Internal; using DotNetCore.CAP.Messages; using Fake; -using Fake.AspNetCore; using Fake.EventBus.RabbitMQ; using Fake.Modularity; using Fake.ObjectMapping.AutoMapper; @@ -14,17 +13,13 @@ using Hangfire; using Hangfire.Redis.StackExchange; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Options; -using Microsoft.IdentityModel.Logging; using RabbitMQ.Client; using Savorboard.CAP.InMemoryMessageQueue; using TrackCenter.Application.Kafka; using TrackCenter.Application.CAP; -using TrackCenter.Application.Hangfire; using TrackCenter.Application.HostServices; using TrackCenter.Application.Options; using TrackCenter.Application.Rabbit; -using TrackCenter.Application.Services.TimerTaskss; namespace TrackCenter.Application; @@ -188,6 +183,12 @@ public class TrackCenterApplicationModule : FakeModule o.UseRedisStorage(connectionString, options) .WithJobExpirationTimeout(TimeSpan.FromHours(1)); // 终止态任务只保留1小时 }); + var hangfireWorkerCount = configuration.GetValue("TrackApp:HangfireWorkerCount"); + if (hangfireWorkerCount <= 0) + { + hangfireWorkerCount = Math.Max(4, Environment.ProcessorCount); + } + context.Services.AddHangfireServer(o => { // 索引越小优先级越高 @@ -200,7 +201,7 @@ public class TrackCenterApplicationModule : FakeModule ]; o.ServerName = Dns.GetHostName(); o.ShutdownTimeout = TimeSpan.FromSeconds(10); - o.WorkerCount = 1; + o.WorkerCount = hangfireWorkerCount; }); // 注册轨迹业务定时任务服务