feat: initialize Evergreen Health backend
This commit is contained in:
fuenmao
@@ -0,0 +1,188 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
/// <summary>
|
||||
/// 作业持久化（数据库）
|
||||
/// </summary>
|
||||
public class DbJobPersistence : IJobPersistence
|
||||
{
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
|
||||
public DbJobPersistence(IServiceScopeFactory serviceScopeFactory)
|
||||
{
|
||||
_serviceScopeFactory = serviceScopeFactory;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 作业调度服务启动时
|
||||
/// </summary>
|
||||
/// <param name="stoppingToken"></param>
|
||||
/// <returns></returns>
|
||||
/// <exception cref="NotSupportedException"></exception>
|
||||
public async Task<IEnumerable<SchedulerBuilder>> PreloadAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
|
||||
var dynamicJobCompiler = scope.ServiceProvider.GetRequiredService<DynamicJobCompiler>();
|
||||
|
||||
// 获取所有定义的作业
|
||||
var allJobs = App.EffectiveTypes.ScanToBuilders().ToList();
|
||||
// 若数据库不存在任何作业，则直接返回
|
||||
if (!await db.Queryable<SysJobDetail>().AnyAsync(u => true, stoppingToken)) return allJobs;
|
||||
|
||||
// 遍历所有定义的作业
|
||||
foreach (var schedulerBuilder in allJobs)
|
||||
{
|
||||
// 获取作业信息构建器
|
||||
var jobBuilder = schedulerBuilder.GetJobBuilder();
|
||||
|
||||
// 加载数据库数据
|
||||
var dbDetail = await db.Queryable<SysJobDetail>().FirstAsync(u => u.JobId == jobBuilder.JobId, stoppingToken);
|
||||
if (dbDetail == null) continue;
|
||||
|
||||
// 同步数据库数据
|
||||
jobBuilder.LoadFrom(dbDetail);
|
||||
|
||||
// 获取作业的所有数据库的触发器
|
||||
var dbTriggers = await db.Queryable<SysJobTrigger>().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<SysJobDetail>().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<SysJobTrigger>().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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 作业计划初始化通知
|
||||
/// </summary>
|
||||
/// <param name="builder"></param>
|
||||
/// <param name="stoppingToken"></param>
|
||||
/// <returns></returns>
|
||||
public Task<SchedulerBuilder> OnLoadingAsync(SchedulerBuilder builder, CancellationToken stoppingToken)
|
||||
{
|
||||
return Task.FromResult(builder);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 作业计划Scheduler的JobDetail变化时
|
||||
/// </summary>
|
||||
/// <param name="context"></param>
|
||||
/// <returns></returns>
|
||||
public async Task OnChangedAsync(PersistenceContext context)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
|
||||
|
||||
var jobDetail = context.JobDetail.Adapt<SysJobDetail>();
|
||||
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<SysJobDetail>().Where(u => u.JobId == jobDetail.JobId).ExecuteCommandAsync();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 作业计划Scheduler的触发器Trigger变化时
|
||||
/// </summary>
|
||||
/// <param name="context"></param>
|
||||
/// <returns></returns>
|
||||
public async Task OnTriggerChangedAsync(PersistenceTriggerContext context)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
|
||||
|
||||
var jobTrigger = context.Trigger.Adapt<SysJobTrigger>();
|
||||
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<SysJobTrigger>().Where(u => u.TriggerId == jobTrigger.TriggerId && u.JobId == jobTrigger.JobId).ExecuteCommandAsync();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 作业触发器运行记录
|
||||
/// </summary>
|
||||
/// <param name="context"></param>
|
||||
/// <returns></returns>
|
||||
public async Task OnExecutionRecordAsync(PersistenceExecutionRecordContext context)
|
||||
{
|
||||
using var scope = _serviceScopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>().CopyNew();
|
||||
|
||||
var jobTriggerRecord = context.Timeline.Adapt<SysJobTriggerRecord>();
|
||||
await db.Insertable(jobTriggerRecord).ExecuteCommandAsync();
|
||||
|
||||
await scope.ServiceProvider.GetRequiredService<SysJobService>().ClearExpireJobTriggerRecord(jobTriggerRecord);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
public class JobDetailInput
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
public string JobId { get; set; }
|
||||
}
|
||||
|
||||
public class PageJobDetailInput : BasePageInput
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
public string JobId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 组名称
|
||||
/// </summary>
|
||||
public string GroupName { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 描述信息
|
||||
/// </summary>
|
||||
public string Description { get; set; }
|
||||
}
|
||||
|
||||
public class AddJobDetailInput : SysJobDetail
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
[Required(ErrorMessage = "作业Id不能为空"), MinLength(2, ErrorMessage = "作业Id不能少于2个字符")]
|
||||
public override string JobId { get; set; }
|
||||
}
|
||||
|
||||
public class UpdateJobDetailInput : AddJobDetailInput
|
||||
{
|
||||
}
|
||||
|
||||
public class DeleteJobDetailInput : JobDetailInput
|
||||
{
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
public class JobDetailOutput
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业信息
|
||||
/// </summary>
|
||||
public SysJobDetail JobDetail { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 触发器集合
|
||||
/// </summary>
|
||||
public List<SysJobTrigger> JobTriggers { get; set; }
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
public class JobTriggerInput
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
public string JobId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 触发器Id
|
||||
/// </summary>
|
||||
public string TriggerId { get; set; }
|
||||
}
|
||||
|
||||
public class AddJobTriggerInput : SysJobTrigger
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
[Required(ErrorMessage = "作业Id不能为空"), MinLength(2, ErrorMessage = "作业Id不能少于2个字符")]
|
||||
public override string JobId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 触发器Id
|
||||
/// </summary>
|
||||
[Required(ErrorMessage = "触发器Id不能为空"), MinLength(2, ErrorMessage = "触发器Id不能少于2个字符")]
|
||||
public override string TriggerId { get; set; }
|
||||
}
|
||||
|
||||
public class UpdateJobTriggerInput : AddJobTriggerInput
|
||||
{
|
||||
}
|
||||
|
||||
public class DeleteJobTriggerInput : JobTriggerInput
|
||||
{
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
public class PageJobTriggerRecordInput : BasePageInput
|
||||
{
|
||||
/// <summary>
|
||||
/// 作业Id
|
||||
/// </summary>
|
||||
public string JobId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 触发器Id
|
||||
/// </summary>
|
||||
public string TriggerId { get; set; }
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
/// <summary>
|
||||
/// 作业集群控制
|
||||
/// </summary>
|
||||
public class JobClusterServer : IJobClusterServer
|
||||
{
|
||||
private static readonly SqlSugarRepository<SysJobCluster> _sysJobClusterRep = App.GetRequiredService<SqlSugarRepository<SysJobCluster>>();
|
||||
private readonly Random _random = new(DateTime.Now.Millisecond);
|
||||
|
||||
/// <summary>
|
||||
/// 当前作业调度器启动通知
|
||||
/// </summary>
|
||||
/// <param name="context">作业集群服务上下文</param>
|
||||
public async void Start(JobClusterContext context)
|
||||
{
|
||||
// 在作业集群表中，如果 clusterId 不存在，则新增一条（否则更新一条），并设置 status 为 ClusterStatus.Waiting
|
||||
if (await _sysJobClusterRep.IsAnyAsync(u => u.ClusterId == context.ClusterId))
|
||||
{
|
||||
await _sysJobClusterRep.AsUpdateable().SetColumns(u => u.Status == ClusterStatus.Waiting).Where(u => u.ClusterId == context.ClusterId).ExecuteCommandAsync();
|
||||
}
|
||||
else
|
||||
{
|
||||
await _sysJobClusterRep.AsInsertable(new SysJobCluster { ClusterId = context.ClusterId, Status = ClusterStatus.Waiting }).ExecuteCommandAsync();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 等待被唤醒
|
||||
/// </summary>
|
||||
/// <param name="context">作业集群服务上下文</param>
|
||||
/// <returns><see cref="Task"/></returns>
|
||||
public async Task WaitingForAsync(JobClusterContext context)
|
||||
{
|
||||
var clusterId = context.ClusterId;
|
||||
|
||||
while (true)
|
||||
{
|
||||
// 控制集群心跳频率（放在头部为了防止 IsAnyAsync continue 没sleep占用大量IO和CPU）
|
||||
await Task.Delay(3000 + _random.Next(500, 1000)); // 错开集群同时启动
|
||||
|
||||
try
|
||||
{
|
||||
ICache cache = App.GetRequiredService<ICacheProvider>().Cache;
|
||||
// 使用分布式锁
|
||||
using (cache.AcquireLock("lock:JobClusterServer:WaitingForAsync", 1000))
|
||||
{
|
||||
// 在这里查询数据库，根据以下两种情况处理
|
||||
// 1) 如果作业集群表已有 status 为 ClusterStatus.Working 则继续循环
|
||||
// 2) 如果作业集群表中还没有其他服务或只有自己，则插入一条集群服务或调用 await WorkNowAsync(clusterId); 之后 return;
|
||||
// 3) 如果作业集群表中没有 status 为 ClusterStatus.Working 的，调用 await WorkNowAsync(clusterId); 之后 return;
|
||||
if (await _sysJobClusterRep.IsAnyAsync(u => u.Status == ClusterStatus.Working)) continue;
|
||||
|
||||
await WorkNowAsync(clusterId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 当前作业调度器停止通知
|
||||
/// </summary>
|
||||
/// <param name="context">作业集群服务上下文</param>
|
||||
public async void Stop(JobClusterContext context)
|
||||
{
|
||||
// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed
|
||||
await _sysJobClusterRep.UpdateAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 当前作业调度器宕机
|
||||
/// </summary>
|
||||
/// <param name="context">作业集群服务上下文</param>
|
||||
public async void Crash(JobClusterContext context)
|
||||
{
|
||||
// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed
|
||||
await _sysJobClusterRep.UpdateAsync(u => new SysJobCluster { Status = ClusterStatus.Crashed }, u => u.ClusterId == context.ClusterId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 指示集群可以工作
|
||||
/// </summary>
|
||||
/// <param name="clusterId">集群 Id</param>
|
||||
/// <returns></returns>
|
||||
private static async Task WorkNowAsync(string clusterId)
|
||||
{
|
||||
// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Working
|
||||
await _sysJobClusterRep.UpdateAsync(u => new SysJobCluster { Status = ClusterStatus.Working }, u => u.ClusterId == clusterId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
/// <summary>
|
||||
/// 作业执行监视器
|
||||
/// </summary>
|
||||
public class JobMonitor : IJobMonitor
|
||||
{
|
||||
private readonly SysConfigService _sysConfigService;
|
||||
private readonly IEventPublisher _eventPublisher;
|
||||
private readonly ILogger<JobMonitor> _logger;
|
||||
|
||||
public JobMonitor(IServiceScopeFactory serviceScopeFactory, IEventPublisher eventPublisher, ILogger<JobMonitor> logger)
|
||||
{
|
||||
var serviceScope = serviceScopeFactory.CreateScope();
|
||||
_sysConfigService = serviceScope.ServiceProvider.GetRequiredService<SysConfigService>();
|
||||
_eventPublisher = eventPublisher;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public Task OnExecutingAsync(JobExecutingContext context, CancellationToken stoppingToken)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task OnExecutedAsync(JobExecutedContext context, CancellationToken stoppingToken)
|
||||
{
|
||||
if (context.Exception == null) return;
|
||||
|
||||
var exception = $"定时任务【{context.Trigger.Description}】错误：{context.Exception}";
|
||||
// 将作业异常信息记录到本地
|
||||
_logger.LogError(exception);
|
||||
|
||||
if (await _sysConfigService.GetConfigValue<bool>(ConfigConst.SysErrorMail))
|
||||
{
|
||||
// 将作业异常信息发送到邮件
|
||||
await _eventPublisher.PublishAsync(CommonConst.SendErrorMail, exception, stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,398 @@
|
||||
// Admin.NET 项目的版权、商标、专利和其他相关权利均受相应法律法规的保护。使用本项目应遵守相关法律法规和许可证的要求。
|
||||
//
|
||||
// 本项目主要遵循 MIT 许可证和 Apache 许可证（版本 2.0）进行分发和使用。许可证位于源代码树根目录中的 LICENSE-MIT 和 LICENSE-APACHE 文件。
|
||||
//
|
||||
// 不得利用本项目从事危害国家安全、扰乱社会秩序、侵犯他人合法权益等法律法规禁止的活动！任何基于本项目二次开发而产生的一切法律纠纷和责任，我们不承担任何责任！
|
||||
|
||||
namespace Admin.NET.Core.Service;
|
||||
|
||||
/// <summary>
|
||||
/// 系统作业任务服务 🧩
|
||||
/// </summary>
|
||||
[ApiDescriptionSettings(Order = 320, Description = "作业任务")]
|
||||
public class SysJobService : IDynamicApiController, ITransient
|
||||
{
|
||||
private readonly SqlSugarRepository<SysJobDetail> _sysJobDetailRep;
|
||||
private readonly SqlSugarRepository<SysJobTrigger> _sysJobTriggerRep;
|
||||
private readonly SqlSugarRepository<SysJobTriggerRecord> _sysJobTriggerRecordRep;
|
||||
private readonly SqlSugarRepository<SysJobCluster> _sysJobClusterRep;
|
||||
private readonly ISchedulerFactory _schedulerFactory;
|
||||
private readonly DynamicJobCompiler _dynamicJobCompiler;
|
||||
|
||||
public SysJobService(SqlSugarRepository<SysJobDetail> sysJobDetailRep,
|
||||
SqlSugarRepository<SysJobTrigger> sysJobTriggerRep,
|
||||
SqlSugarRepository<SysJobTriggerRecord> sysJobTriggerRecordRep,
|
||||
SqlSugarRepository<SysJobCluster> sysJobClusterRep,
|
||||
ISchedulerFactory schedulerFactory,
|
||||
DynamicJobCompiler dynamicJobCompiler)
|
||||
{
|
||||
_sysJobDetailRep = sysJobDetailRep;
|
||||
_sysJobTriggerRep = sysJobTriggerRep;
|
||||
_sysJobTriggerRecordRep = sysJobTriggerRecordRep;
|
||||
_sysJobClusterRep = sysJobClusterRep;
|
||||
_schedulerFactory = schedulerFactory;
|
||||
_dynamicJobCompiler = dynamicJobCompiler;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取作业分页列表 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("获取作业分页列表")]
|
||||
public async Task<SqlSugarPagedList<JobDetailOutput>> PageJobDetail(PageJobDetailInput input)
|
||||
{
|
||||
var jobDetails = await _sysJobDetailRep.AsQueryable()
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.JobId), u => u.JobId.Contains(input.JobId.Trim()))
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.GroupName), u => u.GroupName.Contains(input.GroupName.Trim()))
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.Description), u => u.Description.Contains(input.Description.Trim()))
|
||||
.Select(d => new JobDetailOutput
|
||||
{
|
||||
JobDetail = d,
|
||||
}).ToPagedListAsync(input.Page, input.PageSize);
|
||||
await _sysJobDetailRep.AsSugarClient().ThenMapperAsync(jobDetails.Items, async u =>
|
||||
{
|
||||
u.JobTriggers = await _sysJobTriggerRep.GetListAsync(t => t.JobId == u.JobDetail.JobId);
|
||||
});
|
||||
|
||||
// 提取中括号里面的参数值
|
||||
var rgx = new Regex(@"(?i)(?<=\[)(.*)(?=\])");
|
||||
foreach (var job in jobDetails.Items)
|
||||
{
|
||||
foreach (var jobTrigger in job.JobTriggers)
|
||||
{
|
||||
jobTrigger.Args = rgx.Match(jobTrigger.Args ?? "").Value;
|
||||
}
|
||||
}
|
||||
return jobDetails;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取作业组名称集合 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("获取作业组名称集合")]
|
||||
public async Task<List<string>> ListJobGroup()
|
||||
{
|
||||
return await _sysJobDetailRep.AsQueryable().Distinct().Select(e => e.GroupName).ToListAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 添加作业 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "AddJobDetail"), HttpPost]
|
||||
[DisplayName("添加作业")]
|
||||
public async Task AddJobDetail(AddJobDetailInput input)
|
||||
{
|
||||
var isExist = await _sysJobDetailRep.IsAnyAsync(u => u.JobId == input.JobId && u.Id != input.Id);
|
||||
if (isExist) throw Oops.Oh(ErrorCodeEnum.D1006);
|
||||
|
||||
// 动态创建作业
|
||||
Type jobType;
|
||||
switch (input.CreateType)
|
||||
{
|
||||
case JobCreateTypeEnum.Script when string.IsNullOrEmpty(input.ScriptCode):
|
||||
throw Oops.Oh(ErrorCodeEnum.D1701);
|
||||
case JobCreateTypeEnum.Script:
|
||||
{
|
||||
jobType = _dynamicJobCompiler.BuildJob(input.ScriptCode);
|
||||
|
||||
if (jobType.GetCustomAttributes(typeof(JobDetailAttribute)).FirstOrDefault() is not JobDetailAttribute jobDetailAttribute)
|
||||
throw Oops.Oh(ErrorCodeEnum.D1702);
|
||||
if (jobDetailAttribute.JobId != input.JobId)
|
||||
throw Oops.Oh(ErrorCodeEnum.D1703);
|
||||
break;
|
||||
}
|
||||
case JobCreateTypeEnum.Http:
|
||||
jobType = typeof(HttpJob);
|
||||
break;
|
||||
|
||||
default:
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
_schedulerFactory.AddJob(JobBuilder.Create(jobType).LoadFrom(input.Adapt<SysJobDetail>()).SetJobType(jobType));
|
||||
|
||||
// 延迟一下等待持久化写入，再执行其他字段的更新
|
||||
await Task.Delay(500);
|
||||
await _sysJobDetailRep.AsUpdateable()
|
||||
.SetColumns(u => new SysJobDetail { CreateType = input.CreateType, ScriptCode = input.ScriptCode })
|
||||
.Where(u => u.JobId == input.JobId).ExecuteCommandAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 更新作业 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "UpdateJobDetail"), HttpPost]
|
||||
[DisplayName("更新作业")]
|
||||
public async Task UpdateJobDetail(UpdateJobDetailInput input)
|
||||
{
|
||||
var isExist = await _sysJobDetailRep.IsAnyAsync(u => u.JobId == input.JobId && u.Id != input.Id);
|
||||
if (isExist) throw Oops.Oh(ErrorCodeEnum.D1006);
|
||||
|
||||
var sysJobDetail = await _sysJobDetailRep.GetFirstAsync(u => u.Id == input.Id);
|
||||
if (sysJobDetail.JobId != input.JobId) throw Oops.Oh(ErrorCodeEnum.D1704);
|
||||
|
||||
var scheduler = _schedulerFactory.GetJob(sysJobDetail.JobId);
|
||||
var oldScriptCode = sysJobDetail.ScriptCode; // 旧脚本代码
|
||||
input.Adapt(sysJobDetail);
|
||||
|
||||
if (input.CreateType == JobCreateTypeEnum.Script)
|
||||
{
|
||||
if (string.IsNullOrEmpty(input.ScriptCode)) throw Oops.Oh(ErrorCodeEnum.D1701);
|
||||
|
||||
if (input.ScriptCode != oldScriptCode)
|
||||
{
|
||||
// 动态创建作业
|
||||
var jobType = _dynamicJobCompiler.BuildJob(input.ScriptCode);
|
||||
|
||||
if (jobType.GetCustomAttributes(typeof(JobDetailAttribute)).FirstOrDefault() is not JobDetailAttribute jobDetailAttribute)
|
||||
throw Oops.Oh(ErrorCodeEnum.D1702);
|
||||
if (jobDetailAttribute.JobId != input.JobId) throw Oops.Oh(ErrorCodeEnum.D1703);
|
||||
|
||||
scheduler?.UpdateDetail(JobBuilder.Create(jobType).LoadFrom(sysJobDetail).SetJobType(jobType));
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
scheduler?.UpdateDetail(scheduler.GetJobBuilder().LoadFrom(sysJobDetail));
|
||||
}
|
||||
|
||||
// Tip: 假如这次更新有变更了 JobId，变更 JobId 后触发的持久化更新执行，会由于找不到 JobId 而更新不到数据
|
||||
// 延迟一下等待持久化写入，再执行其他字段的更新
|
||||
await Task.Delay(500);
|
||||
await _sysJobDetailRep.UpdateAsync(sysJobDetail);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 删除作业 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "DeleteJobDetail"), HttpPost]
|
||||
[DisplayName("删除作业")]
|
||||
public async Task DeleteJobDetail(DeleteJobDetailInput input)
|
||||
{
|
||||
_schedulerFactory.RemoveJob(input.JobId);
|
||||
|
||||
// 如果 _schedulerFactory 中不存在 JodId，则无法触发持久化，下面的代码确保作业和触发器能被删除
|
||||
await _sysJobDetailRep.DeleteAsync(u => u.JobId == input.JobId);
|
||||
await _sysJobTriggerRep.DeleteAsync(u => u.JobId == input.JobId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取触发器列表 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("获取触发器列表")]
|
||||
public async Task<List<SysJobTrigger>> GetJobTriggerList([FromQuery] JobDetailInput input)
|
||||
{
|
||||
return await _sysJobTriggerRep.AsQueryable()
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.JobId), u => u.JobId.Contains(input.JobId))
|
||||
.ToListAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 添加触发器 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "AddJobTrigger"), HttpPost]
|
||||
[DisplayName("添加触发器")]
|
||||
public async Task AddJobTrigger(AddJobTriggerInput input)
|
||||
{
|
||||
var isExist = await _sysJobTriggerRep.IsAnyAsync(u => u.TriggerId == input.TriggerId && u.Id != input.Id);
|
||||
if (isExist) throw Oops.Oh(ErrorCodeEnum.D1006);
|
||||
|
||||
var jobTrigger = input.Adapt<SysJobTrigger>();
|
||||
jobTrigger.Args = "[" + jobTrigger.Args + "]";
|
||||
|
||||
var scheduler = _schedulerFactory.GetJob(input.JobId);
|
||||
scheduler?.AddTrigger(Triggers.Create(input.AssemblyName, input.TriggerType).LoadFrom(jobTrigger));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 更新触发器 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "UpdateJobTrigger"), HttpPost]
|
||||
[DisplayName("更新触发器")]
|
||||
public async Task UpdateJobTrigger(UpdateJobTriggerInput input)
|
||||
{
|
||||
var isExist = await _sysJobTriggerRep.IsAnyAsync(u => u.TriggerId == input.TriggerId && u.Id != input.Id);
|
||||
if (isExist) throw Oops.Oh(ErrorCodeEnum.D1006);
|
||||
|
||||
var jobTrigger = input.Adapt<SysJobTrigger>();
|
||||
if (jobTrigger.EndTime.HasValue && jobTrigger.EndTime.Value.Year < 1901)
|
||||
{
|
||||
jobTrigger.EndTime = null;
|
||||
}
|
||||
if (jobTrigger.StartTime.HasValue && jobTrigger.StartTime.Value.Year < 1901)
|
||||
{
|
||||
jobTrigger.StartTime = null;
|
||||
}
|
||||
jobTrigger.Args = "[" + jobTrigger.Args + "]";
|
||||
|
||||
var scheduler = _schedulerFactory.GetJob(input.JobId);
|
||||
scheduler?.UpdateTrigger(Triggers.Create(input.AssemblyName, input.TriggerType).LoadFrom(jobTrigger));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 删除触发器 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "DeleteJobTrigger"), HttpPost]
|
||||
[DisplayName("删除触发器")]
|
||||
public async Task DeleteJobTrigger(DeleteJobTriggerInput input)
|
||||
{
|
||||
var scheduler = _schedulerFactory.GetJob(input.JobId);
|
||||
scheduler?.RemoveTrigger(input.TriggerId);
|
||||
|
||||
// 如果 _schedulerFactory 中不存在 JodId，则无法触发持久化，下行代码确保触发器能被删除
|
||||
await _sysJobTriggerRep.DeleteAsync(u => u.JobId == input.JobId && u.TriggerId == input.TriggerId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 暂停所有作业 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[DisplayName("暂停所有作业")]
|
||||
public void PauseAllJob()
|
||||
{
|
||||
_schedulerFactory.PauseAll();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 启动所有作业 ⏰
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[DisplayName("启动所有作业")]
|
||||
public void StartAllJob()
|
||||
{
|
||||
_schedulerFactory.StartAll();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 暂停作业 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("暂停作业")]
|
||||
public void PauseJob(JobDetailInput input)
|
||||
{
|
||||
_schedulerFactory.TryPauseJob(input.JobId, out _);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 启动作业 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("启动作业")]
|
||||
public void StartJob(JobDetailInput input)
|
||||
{
|
||||
_schedulerFactory.TryStartJob(input.JobId, out _);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 取消作业 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("取消作业")]
|
||||
public void CancelJob(JobDetailInput input)
|
||||
{
|
||||
_schedulerFactory.TryCancelJob(input.JobId, out _);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 执行作业 ⏰
|
||||
/// </summary>
|
||||
/// <param name="input"></param>
|
||||
[DisplayName("执行作业")]
|
||||
public void RunJob(JobDetailInput input)
|
||||
{
|
||||
if (_schedulerFactory.TryRunJob(input.JobId, out _) != ScheduleResult.Succeed) throw Oops.Oh(ErrorCodeEnum.D1705);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 暂停触发器 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("暂停触发器")]
|
||||
public void PauseTrigger(JobTriggerInput input)
|
||||
{
|
||||
var scheduler = _schedulerFactory.GetJob(input.JobId);
|
||||
scheduler?.PauseTrigger(input.TriggerId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 启动触发器 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("启动触发器")]
|
||||
public void StartTrigger(JobTriggerInput input)
|
||||
{
|
||||
var scheduler = _schedulerFactory.GetJob(input.JobId);
|
||||
scheduler?.StartTrigger(input.TriggerId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 强制唤醒作业调度器 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("强制唤醒作业调度器")]
|
||||
public void CancelSleep()
|
||||
{
|
||||
_schedulerFactory.CancelSleep();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 强制触发所有作业持久化 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("强制触发所有作业持久化")]
|
||||
public void PersistAll()
|
||||
{
|
||||
_schedulerFactory.PersistAll();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取集群列表 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("获取集群列表")]
|
||||
public async Task<List<SysJobCluster>> GetJobClusterList()
|
||||
{
|
||||
return await _sysJobClusterRep.GetListAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取作业触发器运行记录分页列表 ⏰
|
||||
/// </summary>
|
||||
[DisplayName("获取作业触发器运行记录分页列表")]
|
||||
public async Task<SqlSugarPagedList<SysJobTriggerRecord>> PageJobTriggerRecord(PageJobTriggerRecordInput input)
|
||||
{
|
||||
return await _sysJobTriggerRecordRep.AsQueryable()
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.JobId), u => u.JobId == input.JobId)
|
||||
.WhereIF(!string.IsNullOrWhiteSpace(input.TriggerId), u => u.TriggerId == input.TriggerId)
|
||||
.OrderByDescending(u => u.Id)
|
||||
.ToPagedListAsync(input.Page, input.PageSize);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 清空作业触发器运行记录 🔖
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[ApiDescriptionSettings(Name = "ClearJobTriggerRecord"), HttpPost]
|
||||
[DisplayName("清空作业触发器运行记录")]
|
||||
public void ClearJobTriggerRecord()
|
||||
{
|
||||
_sysJobTriggerRecordRep.AsSugarClient().DbMaintenance.TruncateTable<SysJobTriggerRecord>();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 清空不保留的作业触发器运行记录 🔖
|
||||
/// </summary>
|
||||
/// <returns></returns>
|
||||
[NonAction]
|
||||
[DisplayName("清空过期的作业触发器运行记录")]
|
||||
public async Task ClearExpireJobTriggerRecord(SysJobTriggerRecord input)
|
||||
{
|
||||
int keepRecords = 30;//保留记录条数
|
||||
// 使用CopyNew()创建新的数据库连接实例，避免连接冲突
|
||||
var db = _sysJobTriggerRecordRep.AsSugarClient().CopyNew();
|
||||
await db.Deleteable<SysJobTriggerRecord>().In(it => it.Id,
|
||||
db.Queryable<SysJobTriggerRecord>()
|
||||
.Skip(keepRecords)
|
||||
.OrderByDescending(it => it.LastRunTime)
|
||||
.Where(u => u.JobId == input.JobId && u.TriggerId == input.TriggerId)
|
||||
.Select(it => it.Id) //注意Select不要ToList(), ToList就2次查询了
|
||||
).ExecuteCommandAsync();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user