Longbow.Tasks

概述

大体分为了Scheduler（调度任务），Storage（持久化），Trigger（触发器），Task（任务）和逻辑模块，大体流程为通过逻辑代码进行实例化相关类，根据ITask来实现独立的业务流程（也可以使用自带业务类），通过配置触发器来设定触发时间，使用持久化来防止程序崩溃导致任务消失，通过调度任务执行任务。

Scheduler

包含了任务调度器的枚举（准备，运行，禁用），调度器类的接口，内部默认实现类和调度执行的实体类

IScheduler（任务调度类接口）
- 包含了调度器状态，下次与您时间，上次执行时间，上次执行结果，调度器创建时间，相关触发器，相关的任务信息
DefaultScheduler（内部默认任务调度类）
- Triggers（相关触发器）：SchedulerProcess
- NextRuntime（下次运行时间）：Triggers中可用的最小的运行时间
- LastRuntime（最近运行时间）：Triggers中最后一次运行时间的最大值
```
SchedulerProcess?.Triggers.Select(t => t.Trigger.LastRuntime).Max()
```
- CreatedTime (创建时间) ：DateTimeOffset.Now
Start和Stop方法--开始/停止任务调用
- 该方法是在将调度处理器设置为Running时调用

SchedulerProcess（调度执行实体类）

包含了日志委托，任务调度器，调度任务和所有触发器实例

Start ：调度开始

public void Start(ITask task, ITrigger trigger){

    Task.Run(() =>

    {

        // 获得任务或 where T：ITask,new()的泛型

        // DefaultTaskMetaData? TaskContext = new DefaultTaskMetaData(task);

        DefaultTaskMetaData? TaskContext = new DefaultTaskMetaData(new T());

        // 将任务赋值给调度器

        _sche.Task = TaskContext.Task;

        _initToken.Cancel();

        // Stop 调用

        if (_cancellationTokenSource?.IsCancellationRequested ?? false) return;

    });

    // 设置超时取消

    // 调用Task任务

    // 声明触发器，并注册回调函数开始触发器执行

    InternalStart(trigger);

}

// 执行Task并开始触发器

private void InternalStart(ITrigger trigger)

{

	//声明一个返回值为Task的，参数为CancellationToken的Func

    var dowork = new Func<CancellationToken, Task>(async token =>

    {

    	//没有异常信息

        _sche.Exception = null;

        // 设置任务超时取消令牌

        var taskCancelTokenSource = new CancellationTokenSource(trigger.Timeout);

        try

        {

            // 保证 ITask 的 new() 方法被执行完毕，泛型中有声明

            _initToken.Token.WaitHandle.WaitOne();

            var taskToken = CancellationTokenSource.CreateLinkedTokenSource(token, taskCancelTokenSource.Token);

            if (!taskToken.IsCancellationRequested && TaskContext != null)

            {

            	//执行异步任务

                await TaskContext.Execute(taskToken.Token).ConfigureAwait(false);

                //异步任务执行完毕，设置为成功

            	trigger.LastResult = TriggerResult.Success;

            }

        }

        catch (TaskCanceledException) { }

        catch (Exception ex)

        {

        	//有异常，保存异常

            _sche.Exception = ex;

            ex.Log();

        }

        // 设置 Trigger 状态

        // 取消

        if (token.IsCancellationRequested) trigger.LastResult = TriggerResult.Cancelled;

        // 超时

        if (taskCancelTokenSource.IsCancellationRequested) trigger.LastResult = TriggerResult.Timeout;

        // 错误

        if (_sche.Exception != null) trigger.LastResult = TriggerResult.Error;

    });

    // 声明触发器的维护类

    var triggerProcess = new TriggerProcess(Scheduler.Name, LoggerAction, trigger, Storage, dowork);

    // 保存到触发器数组中

    Triggers.Add(triggerProcess);

    // 注册触发器状态改变回调方法

    trigger.EnabeldChanged = enabled =>

    {

    	// 当触发器状态改变后进行操作

        LoggerAction($"{nameof(TriggerProcess)} Trigger({trigger.GetType().Name}) Enabled({enabled})");

        if (!enabled)

        {

        	// 触发器设置为禁用，则持久化的任务设置为停止

            triggerProcess.Stop();

            return;

        }

        // 状态为运行时，则继续运行触发器的维护类（让触发器开始工作）

        if (Status == SchedulerStatus.Running) triggerProcess.Start(_cancellationTokenSource?.Token);

    };

    // 当当前状态为就绪

    if (Status == SchedulerStatus.Ready)

    {

    	// 设置为运行

        Status = SchedulerStatus.Running;

        _cancellationTokenSource = new CancellationTokenSource();

        // 运行触发器的维护类

        triggerProcess.Start(_cancellationTokenSource.Token);

    }

}

Storage

持久化相关类，包含了持久化到文件，序列化触发器，加密等。主要流程为将已初始化的触发器经过加密设置（密钥和向量值）成序列化值，然后将序列化的信息保存到本地文件中的操作

FileStorageOptions

持久化配置信息：是否物理文件持久化（默认为true）；保存的文件目录；是否加密（默认true）；密钥；向量值

IStorage 持久化接口

/// 从持久化载体加载 ITrigger 触发器

Task LoadAsync();

/// 将 ITrigger 触发器保存到序列化载体中

/// <param name="schedulerName">任务调度器名称</param>

/// <param name="trigger"></param>

bool Save(string schedulerName, ITrigger trigger);

/// 获得 上一次操作异常信息实例

Exception? Exception { get; }

/// 移除指定任务调度

/// <param name="schedulerNames">要移除调度名称集合</param>

bool Remove(IEnumerable<string> schedulerNames);

FileStorage 持久化到物理文件操作类

主要方法参数说明

 public virtual Task LoadAsync()

 {

     // 从文件加载

     Exception = null;

     // 物理持久化为true时

     if (Options.Enabled){

     	//获得配置中保存的路径，并获得所有*.bin的文件进行遍历

     	RetrieveSchedulers().AsParallel().ForAll(fileName =>{

     		// 如果存在

			if (File.Exists(fileName)){

                try{

                    lock (locker){

                    	//创建任务名称

                         var scheduleName = Path.GetFileNameWithoutExtension(fileName);

                         // 反序列化，将string转换为触发器

                         var trigger = JsonSerializeExtensions.Deserialize(fileName, Options);

                         //获得任务类

                         // 返回为null？？？很奇怪

                         var task = CreateTaskByScheduleName(scheduleName);

                         if (task != null){

                         TaskServicesManager.GetOrAdd(scheduleName, task, trigger);

                         }else{

                             // 返回为null？？？很奇怪

                             var callback = CreateCallbackByScheduleName(scheduleName);

                             if (callback != null){

                                TaskServicesManager.GetOrAdd(scheduleName, callback, trigger);

                         }

                     }

                 }

             }

             catch (Exception ex)

             {

             	Exception = ex;

                // load 失败删除文件防止一直 load 出错

                var target = $"{fileName}.err";

                if (File.Exists(target)) File.Delete(target);

                File.Move(fileName, $"{fileName}.err");

              }

          }

       });

    }

    return Task.CompletedTask;

}

FileStorageExtensions
- ITaskStorageBuilder的扩展类，用于注入物理文件持久化服务到容器

JsonSerializeExtensions 二进制序列化操作类

使用TripleDES加密方式

// 解密

private static string Decrypte(this string data, FileStorageOptions option)

{

    using var des = Create(option.Key, option.IV);

    // 创建解密者

    var decryptor = des.CreateDecryptor();

    // 将String转换为byte数组

    var buffer = Convert.FromBase64String(data);

    // 解密数组块

    var result = decryptor.TransformFinalBlock(buffer, 0, buffer.Length);

    return Encoding.UTF8.GetString(result);

}

// 加密

private static string Encrypte(this string data, FileStorageOptions option)

{

    using var des = Create(option.Key, option.IV);

    //创建加密者

    var encryptor = des.CreateEncryptor();

    //转换

    var buffer = Encoding.UTF8.GetBytes(data);

    // 加密

    var result = encryptor.TransformFinalBlock(buffer, 0, buffer.Length);

    return Convert.ToBase64String(result);

}

// 创建TripleDES

private static TripleDES Create(string key, string iv)

{

    var des = TripleDES.Create();

    //设置加密模式

    des.Mode = CipherMode.ECB;

    // 填充算法

    des.Padding = PaddingMode.PKCS7;

    // 向量值

    des.IV = Convert.FromBase64String(iv);

    // 加密值

    des.Key = Convert.FromBase64String(key);

    return des;

}

Trigger

ITrigger 触发器接口
Cron Cron操作类
DefaultTrigger 内部默认的触发器，只执行一次
CronTrigger Cron触发器
- 通过心跳来判断下次执行时间
- CronExpression.GetNextExecution() 获得下次执行时间
RecurringTrigger 重复执行的触发器
- 通过心跳来判断下次执行时间
- 等待一个间隔周期，进行触发

Task

ITask 任务类接口

逻辑模块

TaskServiceCollectionExtensions
- IServiceCollection注入方法
TaskServicesFactory
- Create
- StartAsync：调用物理持久化的加载项
- StopAsync：调用了TaskServicesManager.Shutdown(token)方法

TaskServicesManager

Init(初始化)

// 创建任务工厂

TaskServicesFactory.Create(options, storage ?? new NoneStorage());

// 加载物理持久化中的触发器

_ = Factory?.StartAsync();

GetOrAdd（新增）

internal static readonly ConcurrentDictionary<string, Lazy<SchedulerProcess>> _schedulerPool = new();

// 将任务与触发器添加到调度中 多线程安全

public static IScheduler GetOrAdd<T>(string schedulerName, ITrigger? trigger = null) where T : ITask, new()

{

	// 判断任务名称，如果不存在则直接使用任务名称

    if (string.IsNullOrEmpty(schedulerName))

    {

    	schedulerName = typeof(T).Name;

    }

	// 懒加载，因为内部方法中有并发任务，因此需要用懒加载

    return _schedulerPool.GetOrAdd(schedulerName, key => new Lazy<SchedulerProcess>(() =>

    {

    	// 创建调度器，并将创建的调度器与任务调度器关联

        var process = GetSchedulerProcess(key);

        // 绑定任务与触发器

        process.Start<T>(trigger ?? TriggerBuilder.Default.Build());

        return process;

    })).Value.Scheduler;

}

巴特西

Longbow.Tasks

Longbow.Tasks

概述

Scheduler

Storage

Trigger

Task

逻辑模块

最新文章

热门文章