feat: 轨迹拉取定时任务

src/TrackCenter.Api/appsettings.Development.json

@@ -4,5 +4,8 @@
       "Default": "Debug",
       "Microsoft.AspNetCore": "Warning"
     }
+  },
+  "FeiShuNotice": {
+    "EnableFeiShuExceptionSubscribe": false
   }
 }

src/TrackCenter.Application/HostServices/TrackTaskComponentHostedService.cs

@@ -13,11 +13,16 @@ public class TrackTaskComponentHostedService : IHostedService
     private readonly IOptionsMonitor<TrackAppOptions> _optionsMonitor;
     private TrackAppOptions? _oldOptions;
     private readonly ILogger<TrackTaskComponentHostedService> _logger;
+    private readonly IRecurringJobManager _recurringJobManager;
 
-    public TrackTaskComponentHostedService(IOptionsMonitor<TrackAppOptions> optionsMonitor, ILogger<TrackTaskComponentHostedService> logger)
+    public TrackTaskComponentHostedService(
+        IOptionsMonitor<TrackAppOptions> optionsMonitor,
+        ILogger<TrackTaskComponentHostedService> 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<TrackPullTask>("Execute_TrackPullTask",
+            _recurringJobManager.AddOrUpdate<TrackPullTask>("Execute_TrackPullTask",
                 q,
                 x => x.RefreshForTaskAsync(),
                 $"*/{newVal.TrackPullTaskMinuteInterval} * * * *");

src/TrackCenter.Application/HostServices/WarningTaskComponentHostedService.cs

@@ -17,11 +17,16 @@ public class WarningTaskComponentHostedService : IHostedService
     private readonly IOptionsMonitor<TrackWarningTaskOptions> _optionsMonitor;
     private TrackWarningTaskOptions? _oldOptions;
     private readonly ILogger<WarningTaskComponentHostedService> _logger;
+    private readonly IRecurringJobManager _recurringJobManager;
 
-    public WarningTaskComponentHostedService(IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor, ILogger<WarningTaskComponentHostedService> logger)
+    public WarningTaskComponentHostedService(
+        IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor,
+        ILogger<WarningTaskComponentHostedService> 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<IWarningTaskScanner>("Scan_WarningNotifyTask",
+            _recurringJobManager.AddOrUpdate<IWarningTaskScanner>("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<IWarningTaskScanner>("Scan_ReturnWarningRetryPushTask",
+            _recurringJobManager.AddOrUpdate<IWarningTaskScanner>("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<IWarningTaskScanner>("Scan_TransportWarningDetailsGenerateTask",
+            _recurringJobManager.AddOrUpdate<IWarningTaskScanner>("Scan_TransportWarningDetailsGenerateTask",
                 q,
                 x => x.ScanTransportDetailsGenerateTaskAndSchedulingAsync(),
                 $"*/{newVal.ScanTransportWarningDetailsGenerateTaskMinuteInterval} * * * *");

src/TrackCenter.Application/Options/TrackAppOptions.cs

@@ -31,4 +31,9 @@ public class TrackAppOptions
     /// 拉取优先级
     /// </summary>
     public int TrackPullTaskLevel { get; set; }
+
+    /// <summary>
+    /// Hangfire 工作线程数
+    /// </summary>
+    public int HangfireWorkerCount { get; set; }
 }

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<TrackCenterDbContext>,
     {
         return await Context.Queryable<OrderEntity>()
             .InnerJoin<OrderPullStatusEntity>((a, b) => a.Id == b.OrderId)
-            .LeftJoin<VAppEntity>((a, b, c) => a.AppKey == c.AppKey)
+            .LeftJoin<TrackPullConfigEntity>((a, b, c) =>
+                a.AppKey == c.AppKey && a.ServiceProviderCode == c.ServiceProviderCode)
+            .LeftJoin<TrackPullPriorityEntity>((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 = d.OrderNoType == TrackOrderNoType.客户单号 ? a.OrderNo :
-                OrderNo = string.IsNullOrWhiteSpace(b.OrderNo) ? a.OrderNo : b.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<TrackCenterDbContext>,
             .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

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;
 /// 轨迹拉取定时任务
 /// </summary>
 [ApiExplorerSettings(GroupName = "定时任务")]
-public class TrackPullTask : ApplicationServiceBase, ISingletonDependency
+public class TrackPullTask(
+    IBackgroundJobClient backgroundJobClient,
+    ITrackPullConfigRepository trackPullConfigRepository,
+    IOptionsMonitor<TrackAppOptions> trackAppOptions,
+    TrackPullCacheService trackPullCacheService,
+    ProvideQueries provideQueries,
+    InterfaceForwardService interfaceForwardService,
+    IOptions<InterfaceRelayParamsConfigOptions> interfaceOptions)
+    : ApplicationServiceBase
 {
-    private readonly IServiceScopeFactory _serviceScopeFactory;
-
-    public TrackPullTask(IServiceScopeFactory serviceScopeFactory)
-    {
-        _serviceScopeFactory = serviceScopeFactory;
-    }
-
     /// <summary>
     /// 定时任务刷新轨迹
     /// </summary>
     /// <returns></returns>
-    [AllowAnonymous]
+    [DisableConcurrentExecution(timeoutInSeconds: 1800)]
+    [UnitOfWork]
     [HttpPost("/TimerTask/Refresh")]
-    public async Task<bool> RefreshForTaskAsync()
+    public virtual async Task<bool> RefreshForTaskAsync()
     {
-        using var serviceScope = _serviceScopeFactory.CreateScope();
-
-        var trackPullConfigRepository = serviceScope.ServiceProvider.GetRequiredService<ITrackPullConfigRepository>();
-
         var configs = await trackPullConfigRepository.AsQueryable()
             .LeftJoin<TrackPullPriorityEntity>((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<TrackPullTask>(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<TrackPullCacheService>();
         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
         {
-                var provideQueries = serviceScope.ServiceProvider.GetRequiredService<ProvideQueries>();
+            await DoRefreshAsync(config, provideQueries, interfaceForwardService, interfaceOptions.Value);
-                var interfaceForward = serviceScope.ServiceProvider
-                    .GetRequiredService<InterfaceForwardService>();
-                var interfaceOptions = serviceScope.ServiceProvider
-                    .GetRequiredService<IOptions<InterfaceRelayParamsConfigOptions>>();
-                await DoRefreshAsync(config, provideQueries, interfaceForward, 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)
     {

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<int>("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;
         });
 
         // 注册轨迹业务定时任务服务