From 42b9316f609eefd54f54b4d353df30200aebce58 Mon Sep 17 00:00:00 2001 From: xiangning Date: Thu, 23 Jul 2026 19:01:28 +0800 Subject: [PATCH] refactor: dispatch import jobs through MediatR --- .../ImportJobProcessorService.cs | 101 ++++++++---------- .../Imports/Services/IImportJobProcessor.cs | 4 - .../WorkflowJobArchitectureTests.cs | 15 +++ 3 files changed, 58 insertions(+), 62 deletions(-) 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(