// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。 // // 本项目主要遵循 MIT 许可证和 Apache 许可证(版本 2.0)进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。 // // 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动!任何基于本项目二次开发而产生的一切法律纠纷和责任,我们不承担任何责任! namespace Admin.NET.Core.Service; /// /// 作业持久化(数据库) /// public class DbJobPersistence : IJobPersistence { private readonly IServiceScopeFactory _serviceScopeFactory; public DbJobPersistence(IServiceScopeFactory serviceScopeFactory) { _serviceScopeFactory = serviceScopeFactory; } /// /// 作业调度服务启动时 /// /// /// /// public async Task> PreloadAsync(CancellationToken stoppingToken) { using var scope = _serviceScopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService().CopyNew(); var dynamicJobCompiler = scope.ServiceProvider.GetRequiredService(); // 获取所有定义的作业 var allJobs = App.EffectiveTypes.ScanToBuilders().ToList(); // 若数据库不存在任何作业,则直接返回 if (!await db.Queryable().AnyAsync(u => true, stoppingToken)) return allJobs; // 遍历所有定义的作业 foreach (var schedulerBuilder in allJobs) { // 获取作业信息构建器 var jobBuilder = schedulerBuilder.GetJobBuilder(); // 加载数据库数据 var dbDetail = await db.Queryable().FirstAsync(u => u.JobId == jobBuilder.JobId, stoppingToken); if (dbDetail == null) continue; // 同步数据库数据 jobBuilder.LoadFrom(dbDetail); // 获取作业的所有数据库的触发器 var dbTriggers = await db.Queryable().Where(u => u.JobId == jobBuilder.JobId).ToListAsync(stoppingToken); // 遍历所有作业触发器 foreach (var (_, triggerBuilder) in schedulerBuilder.GetEnumerable()) { // 加载数据库数据 var dbTrigger = dbTriggers.FirstOrDefault(u => u.JobId == jobBuilder.JobId && u.TriggerId == triggerBuilder.TriggerId); if (dbTrigger == null) continue; triggerBuilder.LoadFrom(dbTrigger).Updated(); // 标记更新 } // 遍历所有非编译时定义的触发器加入到作业中 foreach (var dbTrigger in dbTriggers) { if (schedulerBuilder.GetTriggerBuilder(dbTrigger.TriggerId)?.JobId == jobBuilder.JobId) continue; var triggerBuilder = TriggerBuilder.Create(dbTrigger.TriggerId).LoadFrom(dbTrigger); schedulerBuilder.AddTriggerBuilder(triggerBuilder); // 先添加 triggerBuilder.Updated(); // 再标记更新 } // 标记更新 schedulerBuilder.Updated(); } // 获取数据库所有通过脚本创建的作业 var allDbScriptJobs = await db.Queryable().Where(u => u.CreateType != JobCreateTypeEnum.BuiltIn).ToListAsync(stoppingToken); foreach (var dbDetail in allDbScriptJobs) { // 动态创建作业 Type jobType = dbDetail.CreateType switch { JobCreateTypeEnum.Script => dynamicJobCompiler.BuildJob(dbDetail.ScriptCode), JobCreateTypeEnum.Http => typeof(HttpJob), _ => throw new NotSupportedException(), }; // 动态构建的 jobType 的程序集名称为随机名称,需重新设置 dbDetail.AssemblyName = jobType.Assembly.FullName!.Split(',')[0]; var jobBuilder = JobBuilder.Create(jobType).LoadFrom(dbDetail); // 强行设置为不扫描 IJob 实现类 [Trigger] 特性触发器,否则 SchedulerBuilder.Create 会再次扫描,导致重复添加同名触发器 jobBuilder.SetIncludeAnnotations(false); // 获取作业的所有数据库的触发器加入到作业中 var dbTriggers = await db.Queryable().Where(u => u.JobId == jobBuilder.JobId).ToListAsync(); var triggerBuilders = dbTriggers.Select(u => TriggerBuilder.Create(u.TriggerId).LoadFrom(u).Updated()); var schedulerBuilder = SchedulerBuilder.Create(jobBuilder, triggerBuilders.ToArray()); // 标记更新 schedulerBuilder.Updated(); allJobs.Add(schedulerBuilder); } return allJobs; } /// /// 作业计划初始化通知 /// /// /// /// public Task OnLoadingAsync(SchedulerBuilder builder, CancellationToken stoppingToken) { return Task.FromResult(builder); } /// /// 作业计划Scheduler的JobDetail变化时 /// /// /// public async Task OnChangedAsync(PersistenceContext context) { using var scope = _serviceScopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService().CopyNew(); var jobDetail = context.JobDetail.Adapt(); switch (context.Behavior) { case PersistenceBehavior.Appended: await db.Insertable(jobDetail).ExecuteCommandAsync(); break; case PersistenceBehavior.Updated: await db.Updateable(jobDetail).WhereColumns(u => new { u.JobId }).IgnoreColumns(u => new { u.Id, u.CreateType, u.ScriptCode }).ExecuteCommandAsync(); break; case PersistenceBehavior.Removed: await db.Deleteable().Where(u => u.JobId == jobDetail.JobId).ExecuteCommandAsync(); break; } } /// /// 作业计划Scheduler的触发器Trigger变化时 /// /// /// public async Task OnTriggerChangedAsync(PersistenceTriggerContext context) { using var scope = _serviceScopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService().CopyNew(); var jobTrigger = context.Trigger.Adapt(); switch (context.Behavior) { case PersistenceBehavior.Appended: await db.Insertable(jobTrigger).ExecuteCommandAsync(); break; case PersistenceBehavior.Updated: await db.Updateable(jobTrigger).WhereColumns(u => new { u.TriggerId, u.JobId }).IgnoreColumns(u => new { u.Id }).ExecuteCommandAsync(); break; case PersistenceBehavior.Removed: await db.Deleteable().Where(u => u.TriggerId == jobTrigger.TriggerId && u.JobId == jobTrigger.JobId).ExecuteCommandAsync(); break; } } /// /// 作业触发器运行记录 /// /// /// public async Task OnExecutionRecordAsync(PersistenceExecutionRecordContext context) { using var scope = _serviceScopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService().CopyNew(); var jobTriggerRecord = context.Timeline.Adapt(); await db.Insertable(jobTriggerRecord).ExecuteCommandAsync(); await scope.ServiceProvider.GetRequiredService().ClearExpireJobTriggerRecord(jobTriggerRecord); } }