diff --git a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs
index 432f238..278ccaf 100644
--- a/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs
+++ b/src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs
@@ -1,48 +1,37 @@
+using MediatR;
using Microsoft.Extensions.Options;
-using Workflow.Application.BuildFlow.Imports.Services;
-using Workflow.Application.Notifications;
-using Workflow.Domain.Enums;
-using Workflow.Infrastructure.Persistence;
+using Workflow.Application.BuildFlow.Imports.Commands;
+using Workflow.Application.Reliability;
namespace BuildFlow.Host.Services.BackgroundServices;
///
-/// 导入任务后台处理器:周期性扫描 Pending 任务,置为 Running 后调用 ImportJobProcessor。
-/// 参考 TimeoutSchedulerService 结构:BackgroundService 单例 + IServiceScopeFactory 创建独立 scope,
-/// 以便解析 scoped 依赖(WorkflowDbContext、IImportJobProcessor)。
+/// 周期触发导入任务原子领取命令的 Host 后台服务。
///
-public class ImportJobProcessorService : BackgroundService
+public sealed class ImportJobProcessorService(
+ IServiceScopeFactory scopeFactory,
+ IOptionsMonitor options,
+ ILogger logger) : BackgroundService
{
- private readonly IServiceScopeFactory _scopeFactory;
- private readonly ILogger _logger;
- private readonly IOptionsMonitor _options;
-
- public ImportJobProcessorService(
- IServiceScopeFactory scopeFactory,
- ILogger logger,
- IOptionsMonitor options)
- {
- _scopeFactory = scopeFactory;
- _logger = logger;
- _options = options;
- }
+ private readonly string _leaseOwner =
+ $"import:{Environment.MachineName}:{Environment.ProcessId}:{Guid.NewGuid():N}";
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
- // timer 间隔启动时定一次(方案 A:timer 参数不热重载,改间隔需重启服务)
- var opts = _options.CurrentValue;
- _logger.LogInformation("ImportJobProcessor 启动,轮询间隔 {Interval}s", opts.Scheduler.PollIntervalSeconds);
+ var importOptions = options.CurrentValue.Import;
+ logger.LogInformation(
+ "ImportJobProcessor 启动,Owner={LeaseOwner},轮询间隔 {Interval}s",
+ _leaseOwner,
+ importOptions.PollIntervalSeconds);
- // 启动时先等待一轮,避免与应用启动并发争抢资源
- await Task.Delay(opts.Scheduler.StartupDelaySeconds * 1000, stoppingToken);
-
- using var timer = new PeriodicTimer(TimeSpan.FromSeconds(opts.Scheduler.PollIntervalSeconds));
-
- // 整个 do-while 纳入 try/catch:timer.WaitForNextTickAsync 在 stoppingToken 取消时
- // 会抛 OperationCanceledException,该异常发生在 while 条件求值处(不在内层 try 范围),
- // 若不捕获会逃逸出 ExecuteAsync,触发 BackgroundServiceExceptionBehavior=StopHost 导致进程崩溃。
try
{
+ await Task.Delay(
+ TimeSpan.FromSeconds(importOptions.StartupDelaySeconds),
+ stoppingToken);
+ using var timer = new PeriodicTimer(
+ TimeSpan.FromSeconds(importOptions.PollIntervalSeconds));
+
do
{
try
@@ -55,45 +44,41 @@ public class ImportJobProcessorService : BackgroundService
}
catch (Exception ex)
{
- _logger.LogError(ex, "ImportJobProcessor 单轮处理异常");
+ logger.LogError(ex, "ImportJobProcessor 单轮处理异常");
}
}
while (await timer.WaitForNextTickAsync(stoppingToken));
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
- // 正常停机:WaitForNextTickAsync 被取消,优雅退出
+ // 正常停机。
}
- _logger.LogInformation("ImportJobProcessor 停止");
+ logger.LogInformation("ImportJobProcessor 停止,Owner={LeaseOwner}", _leaseOwner);
}
- /// 执行一轮扫描:拉取 Pending 任务 ID 列表,逐个调用 IImportJobProcessor.ProcessAsync。
- public async Task RunScanAsync(CancellationToken ct = default)
+ /// 执行一轮导入任务领取和处理。
+ public async Task RunScanAsync(CancellationToken ct = default)
{
- using var scope = _scopeFactory.CreateScope();
- var db = scope.ServiceProvider.GetRequiredService();
- var processor = scope.ServiceProvider.GetRequiredService();
+ using var scope = scopeFactory.CreateScope();
+ var mediator = scope.ServiceProvider.GetRequiredService();
+ var importOptions = options.CurrentValue.Import;
+ var processed = await mediator.Send(
+ new ProcessAvailableImportJobsCommand(
+ _leaseOwner,
+ importOptions.BatchSize,
+ TimeSpan.FromSeconds(importOptions.LeaseSeconds),
+ importOptions.MaxAttempts),
+ ct);
- var pendingIds = await db.BuildFlowImportJobs
- .AsNoTracking()
- .Where(j => j.Status == BuildFlowImportJobStatus.Pending)
- .OrderBy(j => j.CreatedAt)
- .Select(j => j.Id)
- .Take(20)
- .ToListAsync(ct);
-
- foreach (var jobId in pendingIds)
+ if (processed > 0)
{
- ct.ThrowIfCancellationRequested();
- try
- {
- await processor.ProcessAsync(jobId, ct);
- }
- catch (Exception ex)
- {
- _logger.LogError(ex, "导入任务 {JobId} 处理失败", jobId);
- }
+ logger.LogInformation(
+ "ImportJobProcessor 完成一轮处理,Owner={LeaseOwner},领取 {Count} 个任务",
+ _leaseOwner,
+ processed);
}
+
+ return processed;
}
}
diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
index f745e47..daffd74 100644
--- a/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
+++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/IImportJobProcessor.cs
@@ -9,8 +9,4 @@ public interface IImportJobProcessor
string leaseOwner,
TimeSpan leaseDuration,
CancellationToken ct);
-
- /// 旧 Host 扫描器的临时兼容入口,Host 改为 MediatR 后删除。
- Task ProcessAsync(Guid jobId, CancellationToken ct)
- => ProcessAsync(jobId, "legacy-import-host", TimeSpan.FromMinutes(5), ct);
}
diff --git a/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs b/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
index 7c5ecae..d6d25dd 100644
--- a/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
+++ b/tests/BuildFlow.Architecture.Tests/WorkflowJobArchitectureTests.cs
@@ -29,6 +29,21 @@ public class WorkflowJobArchitectureTests
Assert.DoesNotContain("ProcessEngine", source);
}
+ [Fact]
+ public void ImportJobProcessorServiceShouldUseMediatorWithoutWorkflowInfrastructure()
+ {
+ var source = ReadSource(
+ "src/BuildFlow.Host/Services/BackgroundServices/ImportJobProcessorService.cs");
+
+ Assert.Contains("IMediator", source);
+ Assert.Contains("mediator.Send", source);
+ Assert.Contains("ProcessAvailableImportJobsCommand", source);
+ Assert.DoesNotContain("WorkflowDbContext", source);
+ Assert.DoesNotContain("IImportJobProcessor", source);
+ Assert.DoesNotContain("BuildFlowImportJobStatus", source);
+ Assert.DoesNotContain(".Where(", source);
+ }
+
private static string ReadSource(string relativePath)
{
var path = Path.Combine(