f43e35a4b1
配置中心： - BuildFlow.Host.csproj 加 Winton.Extensions.Configuration.Consul 3.0.1 - Program.cs 注册 AddConsul（Optional=true 容错，ReloadOnChange=true 热重载） - 配置优先级：appsettings.json → Secrets → Consul KV（最高，生产真相源） 热重载（IOptions<T> → IOptionsMonitor<T>）： - KkFileView/AiOptions/NotificationService：.Value → .CurrentValue，完全热重载 - 4 个 HostedService：业务参数热重载，timer 间隔启动时定一次（方案 A） - S3StorageService：IOptionsMonitor + OnChange 回调重建 AmazonS3Client（volatile + lock） - 测试全过（Rag 40 + Workflow Notification 21） compose 加 Consul__Host 环境变量（api 容器注入 Consul 地址） 后续：Consul 部署在 middleware，KV 全量灌入 buildflow/* 前缀
100 lines
3.9 KiB
C#
100 lines
3.9 KiB
C#
using Microsoft.Extensions.Options;
|
||
using Workflow.Application.BuildFlow.Imports.Services;
|
||
using Workflow.Application.Notifications;
|
||
using Workflow.Domain.Enums;
|
||
using Workflow.Infrastructure.Persistence;
|
||
|
||
namespace BuildFlow.Host.Services.BackgroundServices;
|
||
|
||
/// <summary>
|
||
/// 导入任务后台处理器：周期性扫描 Pending 任务，置为 Running 后调用 ImportJobProcessor。
|
||
/// 参考 TimeoutSchedulerService 结构：BackgroundService 单例 + IServiceScopeFactory 创建独立 scope，
|
||
/// 以便解析 scoped 依赖（WorkflowDbContext、IImportJobProcessor）。
|
||
/// </summary>
|
||
public class ImportJobProcessorService : BackgroundService
|
||
{
|
||
private readonly IServiceScopeFactory _scopeFactory;
|
||
private readonly ILogger<ImportJobProcessorService> _logger;
|
||
private readonly IOptionsMonitor<NotificationOptions> _options;
|
||
|
||
public ImportJobProcessorService(
|
||
IServiceScopeFactory scopeFactory,
|
||
ILogger<ImportJobProcessorService> logger,
|
||
IOptionsMonitor<NotificationOptions> options)
|
||
{
|
||
_scopeFactory = scopeFactory;
|
||
_logger = logger;
|
||
_options = options;
|
||
}
|
||
|
||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||
{
|
||
// timer 间隔启动时定一次（方案 A：timer 参数不热重载，改间隔需重启服务）
|
||
var opts = _options.CurrentValue;
|
||
_logger.LogInformation("ImportJobProcessor 启动，轮询间隔 {Interval}s", opts.Scheduler.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
|
||
{
|
||
do
|
||
{
|
||
try
|
||
{
|
||
await RunScanAsync(stoppingToken);
|
||
}
|
||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||
{
|
||
break;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.LogError(ex, "ImportJobProcessor 单轮处理异常");
|
||
}
|
||
}
|
||
while (await timer.WaitForNextTickAsync(stoppingToken));
|
||
}
|
||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||
{
|
||
// 正常停机：WaitForNextTickAsync 被取消，优雅退出
|
||
}
|
||
|
||
_logger.LogInformation("ImportJobProcessor 停止");
|
||
}
|
||
|
||
/// <summary>执行一轮扫描：拉取 Pending 任务 ID 列表，逐个调用 IImportJobProcessor.ProcessAsync。</summary>
|
||
public async Task RunScanAsync(CancellationToken ct = default)
|
||
{
|
||
using var scope = _scopeFactory.CreateScope();
|
||
var db = scope.ServiceProvider.GetRequiredService<WorkflowDbContext>();
|
||
var processor = scope.ServiceProvider.GetRequiredService<IImportJobProcessor>();
|
||
|
||
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)
|
||
{
|
||
ct.ThrowIfCancellationRequested();
|
||
try
|
||
{
|
||
await processor.ProcessAsync(jobId, ct);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.LogError(ex, "导入任务 {JobId} 处理失败", jobId);
|
||
}
|
||
}
|
||
}
|
||
}
|