Sundial 源码梳理 - v2.5.6

代码目录一览

通过入口点说明

实现IServiceCollection，并返回IServiceCollection（Extensions/ScheduleServiceCollectionExtensions.css 53行）

初始化作业调度构建器，并将构建器创建为服务

// 实例化ScheduleOptionsBuilder

scheduleOptionsBuilder ??= new ScheduleOptionsBuilder();

// 注册内部服务

services.AddInternalService(scheduleOptionsBuilder);

// AddInternalService方法中（ScheduleServiceCollectionExtensions.css 89行）

scheduleOptionsBuilder.Build(services);

/**

Build方法中添加了作业监视器，作业执行其和作业调度持久化和作业集群的依赖注入

通过AddSingleton(S, T)将接口和类注入到程序中，后期可通过services.BuildServiceProvider().GetService<T>()来获得服务对象

**/

// 注册作业调度器日志服务

services.AddSingleton<IScheduleLogger>(serviceProvider =>

{

    //ActivatorUtilities.CreateInstance 主要实现带参的IOC注入,默认启用日志

	var scheduleLogger = ActivatorUtilities.CreateInstance<ScheduleLogger>(serviceProvider

		, scheduleOptionsBuilder.LogEnabled);

	return scheduleLogger;

});

// 注册作业计划工厂服务

services.AddSingleton<ISchedulerFactory>(serviceProvider =>

{

	//默认不使用 UTC 时间

	var schedulerFactory = ActivatorUtilities.CreateInstance<SchedulerFactory>(serviceProvider

		, schedulerBuilders

		, scheduleOptionsBuilder.UseUtcTimestamp);

	return schedulerFactory;

});

// 注册作业调度器后台主机服务

// 1.使用AddHostedService，服务会自动调用Worker.StartAsync方法

// 2.返回的方法需集成IHostedService或者BackgroundService

// 3.核心代码，主要用于周期的调度任务

// 4.核心方法

// ExecuteAsync：开始执行

// BackgroundProcessing： ExecuteAsync中的死循环

// CheckIsBlocked ： 串行还是并行的判断

// Dispose方法 销毁

// 具体源代码分析参见下一章节(注册作业调度器后台主机服务 ScheduleHostedService.cs分析)

services.AddHostedService(serviceProvider =>

{

    // 创建作业调度器后台主机对象

    var scheduleHostedService = ActivatorUtilities.CreateInstance<ScheduleHostedService>(

    serviceProvider

   		, scheduleOptionsBuilder.UseUtcTimestamp

    	, scheduleOptionsBuilder.ClusterId);

    // 订阅未察觉任务异常事件

    var unobservedTaskExceptionHandler = scheduleOptionsBuilder.UnobservedTaskExceptionHandler;

    if (unobservedTaskExceptionHandler != default)

    {

    scheduleHostedService.UnobservedTaskException += unobservedTaskExceptionHandler;

    }

    return scheduleHostedService;

});

注册作业调度器后台主机服务 ScheduleHostedService.cs分析

重载了StartAsync（服务启动），ExecuteAsync（执行任务），Dispose（销毁）

StartAsync

作业集群启动通知

// 实现IJobClusterServer接口

// 出处：https://furion.baiqian.ltd/docs/job/#26113-%E4%BD%9C%E4%B8%9A%E9%9B%86%E7%BE%A4%E6%8E%A7%E5%88%B6

public void Start(JobClusterContext context){

	// 根据clusterId 判断，不存在新增，并将状态设置为ClusterStatus.Waiting

}

public async Task WaitingForAsync(JobClusterContext context){

    var clusterId = context.ClusterId;

    while (true){

        try

        {

        // 在这里查询数据库，根据以下两种情况处理

        // 1) 如果作业集群表已有 status 为 ClusterStatus.Working 则继续循环

        // 2) 如果作业集群表中还没有其他服务或只有自己，则插入一条集群服务或调用 await WorkNowAsync(clusterId); 之后 return;

        // 3) 如果作业集群表中没有 status 为 ClusterStatus.Working 的，调用 await WorkNowAsync(clusterId); 之后 return;

        await WorkNowAsync(clusterId);

        return;

    }

    catch { }

    // 控制集群心跳频率

    await Task.Delay(3000);

}

/// <summary>

/// 当前作业调度器停止通知

/// </summary>

/// <param name="context">作业集群服务上下文</param>

public void Stop(JobClusterContext context)

{

	// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed

}

/// <summary>

/// 当前作业调度器宕机

/// </summary>

/// <param name="context">作业集群服务上下文</param>

public void Crash(JobClusterContext context)

{

	// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Crashed

}

/// <summary>

/// 指示集群可以工作

/// </summary>

/// <param name="clusterId">集群 Id</param>

/// <returns></returns>

private Task WorkNowAsync(string clusterId)

{

	// 在作业集群表中，更新 clusterId 的 status 为 ClusterStatus.Working

	// 模拟数据库更新操作（耗时）

	await Task.Delay(3000);

}

Dispose
- 调用作业集群宕机通知（Crash方法）

ExecuteAsync

调用作业集群的WaitingForAsync方法参见IJobClusterServer中的WaitingForAsync

作业调度工厂进行初始化（SchedulerFactory.Preload）

/// 作业计划工厂默认实现类（内部服务）

/// 参见Factories/SchedulerFactory.Internal.cs 文件

internal sealed partial class SchedulerFactory : ISchedulerFactory{

	// 作业调度初始化

	public void Preload(){

        // 输出作业调度度初始化日志

        _logger.LogInformation("Schedule hosted service is preloading...");

        // 标记是否初始化成功

        var preloadSucceed = true;

        try

        {

            // 装载初始作业计划

            var initialSchedulerBuilders = _schedulerBuilders.Concat(Persistence?.Preload() ?? Array.Empty<SchedulerBuilder>());

            // 如果作业调度器中包含作业计划构建器

            if (initialSchedulerBuilders.Any())

            {

                // 逐条遍历并新增到内存中

                foreach (var schedulerBuilder in initialSchedulerBuilders)

                {

                    _ = TrySaveJob(Persistence?.OnLoading(schedulerBuilder) ?? schedulerBuilder

                        , out _

                        , false);

                }

            }

        }

        catch (Exception ex)

        {

            preloadSucceed = false;

            _logger.LogError(ex, "Schedule hosted service preload failed, and a total of <0> schedulers are appended.");

        }

        // 标记当前方法初始化完成

        PreloadCompleted = true;

        // 释放引用内存并立即回收GC

        _schedulerBuilders.Clear();

        GC.Collect();

        // 输出作业调度器初始化日志

        if (preloadSucceed) _logger.LogWarning("Schedule hosted service preload completed, and a total of <{Count}> schedulers are appended.", _schedulers.Count);

	}

}

开始监听（调用BackgroundProcessing方法）

参见BackgroundProcessing
释放作业计划工厂
- _schedulerFactory.Dispose();

BackgroundProcessing

该方法为异步任务方法：async Task BackgroundProcessing(CancellationToken stoppingToken)
方法中获取的检查时间为UTC时间或默认显示时间
查找所有该时间要触发的作业
创建一个任务工厂（TaskFactory）
Parallel.ForEach的方式并行触发每个符和条件的作业任务
遍历触发器 Triggers
判断是否为串行执行：CheckIsBlocked，如果是且还未完成则跳出
设置作业触发器为运行态，检查记录信息并设置下一个触发的时间
将作业和触发器放到作业计划工厂的BlockingCollection中
使用Parallel.For来提高并发
创建线程taskFactory.StartNew
创建上下文JobExecutingContext
通过await Monitor.OnExecutingAsync()方法进行作业执行前的任务监听（自然也有结束后的监听）

开始执行IJobExecutor执行器（默认执行自带的，也可以自己实现）

// 默认策略

// 调用作业处理程序并配置出错执行重试

await Retry.InvokeAsync(async () =>

{

	await jobHandler.ExecuteAsync(jobExecutingContext, stoppingToken);

}

, trigger.NumRetries

, trigger.RetryTimeout

, retryAction: (total, times) =>

{

	// 输出重试日志

	_logger.LogWarning("Retrying {times}/{total} times for {jobExecutingContext}", times, total, jobExecutingContext);

});

// 执行器的重试策略

public class YourJobExecutor : IJobExecutor

{

    private readonly ILogger<YourJobExecutor> _logger;

    public YourJobExecutor(ILogger<YourJobExecutor> logger)

    {

        _logger = logger;

    }

    public async Task ExecuteAsync(JobExecutingContext context, IJob jobHandler, CancellationToken stoppingToken)

    {

        // 实现失败重试策略，如失败重试 3 次

        await Retry.InvokeAsync(async () =>

        {

            await jobHandler.ExecuteAsync(context, stoppingToken);

        }, 3, 1000

        // 每次重试输出日志

        , retryAction: (total, times) =>

        {

            _logger.LogWarning("Retrying {current}/{times} times for {context}", times, total, context);

        });

    }

}

在注册 Schedule 服务中注册 YourJobExecutor：

services.AddSchedule(options =>

{

      // 添加作业执行器

      options.AddExecutor<YourJobExecutor>();

});

检查作业的信息，如果都处于运行状态，则将触发器变为就绪状态

完成后做后续处理：作业完成，将作业信息的运行数据写入持久化中，写入执行日志

//运行数据写入持久化中

// 放入BlockingCollection<PersistenceContext>中

// 该队列使用了TaskCreationOptions.LongRunning 方法让线程长时间运行该任务

Task.Factory.StartNew(state => ((SchedulerFactory)state).ProcessQueue()

                , this, TaskCreationOptions.LongRunning)

private void ProcessQueue(){

	// GetConsumingEnumerable 遍历

    foreach (var context in _persistenceMessageQueue.GetConsumingEnumerable())

    {

        try

        {

            // 作业触发器更改通知

            if (context is PersistenceTriggerContext triggerContext)

            {

            Persistence.OnTriggerChanged(triggerContext);

            }

            // 作业信息更改通知

            else Persistence.OnChanged(context);

        }

        catch (Exception ex)

        {

            if (context is PersistenceTriggerContext triggerContext) _logger.LogError(ex, "Persistence of <{TriggerId}> trigger of <{JobId}> job failed.", triggerContext.TriggerId, triggerContext.JobId);

            else _logger.LogError(ex, "The JobDetail of <{JobId}> persist failed.", context.JobId);

        }

    }

}

// 作业持久化器 IJobPersistence

public class DbJobPersistence : IJobPersistence

{

    public IEnumerable<SchedulerBuilder> Preload()

    {

        // 作业调度服务启动时运行时初始化，可通过数据库加载，或者其他方式

        return Array.Empty<SchedulerBuilder>();

    }

    public SchedulerBuilder OnLoading(SchedulerBuilder builder)

    {

        // 如果是更新操作，则 return builder.Updated(); 将生成 UPDATE 语句

        // 如果是新增操作，则 return builder.Appended(); 将生成 INSERT 语句

        // 如果是删除操作，则 return builder.Removed(); 将生成 DELETE 语句

        // 如果无需标记操作，返回 builder 默认值即可

        return builder;

    }

    public void OnChanged(PersistenceContext context)

    {

        var sql = context.ConvertToSQL("job_detail");

        // 这里执行 sql

    }

    public void OnTriggerChanged(PersistenceTriggerContext context)

    {

        var sql = context.ConvertToSQL("job_trigger");

        // 这里执行 sql

    }

}

// 之后在 Startup.cs 中注册：

services.AddSchedule(options =>

{

      options.AddPersistence<DbJobPersistence>();

});

异常处理

巴特西

Sundial（一）

Sundial 源码梳理 - v2.5.6

代码目录一览

通过入口点说明

注册作业调度器后台主机服务 ScheduleHostedService.cs分析

未完待续

最新文章

热门文章