using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; namespace Admin.NET.Plugin.AiDOP.Infrastructure; /// /// 数据中台全量刷新后台派发器。 /// 将评审 / 采购管线末尾的全量 ETL 从「同步 await」改为「后台异步执行」, /// 并按 key(如 "S1"、"S4")做「单飞 + 合并去抖」: /// · 该 key 空闲 → 立即后台执行一轮; /// · 正在执行 → 只置 pending,不排队叠加; /// · 一轮结束若 pending 为真 → 再补跑一轮(吸收运行期间累计的触发)。 /// 说明:进程内单飞;多实例部署时跨实例合并依赖各模块的定时刷新任务兜底。 /// public class MdpRefreshDispatcher : ISingleton { private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; private readonly object _gate = new(); private readonly Dictionary _states = new(); public MdpRefreshDispatcher(IServiceScopeFactory scopeFactory, ILogger logger) { _scopeFactory = scopeFactory; _logger = logger; } /// /// 提交一次后台刷新(立即返回,不等待执行完成)。 /// /// 去抖分组键,同一 key 同一时刻最多 1 个在跑。 /// 在独立 DI scope 内执行的刷新委托。 public void Enqueue(string key, Func job) { lock (_gate) { if (_states.TryGetValue(key, out var state) && state.Running) { state.Pending = true; // 合并:运行期间的新触发只置 pending return; } _states[key] = new RefreshState { Running = true, Pending = false }; } // fire-and-forget:独立后台任务,异常在内部吞掉并记日志 _ = RunLoopAsync(key, job); } private async Task RunLoopAsync(string key, Func job) { while (true) { try { using var scope = _scopeFactory.CreateScope(); await job(scope.ServiceProvider, CancellationToken.None); } catch (Exception ex) { _logger.LogError(ex, "MdpRefreshDispatcher[{Key}] 后台全量刷新失败", key); } lock (_gate) { var state = _states[key]; if (!state.Pending) { state.Running = false; return; } state.Pending = false; // 吸收运行期间累计的触发,再补跑一轮 } } } private sealed class RefreshState { public bool Running; public bool Pending; } }