.NET 轻量级、高效任务调度器:ScheduleTask

前言

至于任务调度这个基础功能,重要性不言而喻,大多数业务系统都会用到,世面上有很多成熟的三方库比如Quartz,Hangfire,Coravel

这里我们不讨论三方的库如何使用 而是从0开始自己制作一个简易的任务调度,如果只是到分钟级别的粒度基本够用。

正文

技术栈用到了:BackgroundServiceNCrontab

第一步我们定义一个简单的任务约定,不干别的就是一个执行方法:

public interface IScheduleTask
{
    Task ExecuteAsync();
}
public abstract class ScheduleTask : IScheduleTask
{
    public virtual Task ExecuteAsync()
    {
        return Task.CompletedTask;
    }
}

第二步定义特性标注任务执行周期等信的metadata

[AttributeUsage(AttributeTargets.Class, AllowMultiple = true, Inherited = false)]
public class ScheduleTaskAttribute(string cron) : Attribute
{
    /// <summary>
    /// 支持的cron表达式格式 * * * * *:https://en.wikipedia.org/wiki/Cron
    /// 最小单位为分钟
    /// </summary>
    public string Cron { get; set; } = cron;
    public string? Description { get; set; }
    /// <summary>
    /// 是否异步执行.默认false会阻塞接下来的同类任务
    /// </summary>
    public bool IsAsync { get; set; } = false;
    /// <summary>
    /// 是否初始化即启动,默认false
    /// </summary>
    public bool IsStartOnInit { get; set; } = false;
}

第三步我们定义一个调度器约定,不干别的就是判断当前的任务是否可以执行:

public interface IScheduler
{
    /// <summary>
    /// 判断当前的任务是否可以执行
    /// </summary>
    bool CanRun(ScheduleTaskAttribute scheduleMetadata, DateTime referenceTime);
}

好了,基础步骤就完成了,如果我们需要实现配置级别的任务调度或者动态的任务调度 那我们再抽象一个Store:

public class ScheduleTaskMetadata(Type scheduleTaskType, string cron)
{
    public Type ScheduleTaskType { get; set; } = scheduleTaskType;
    public string Cron { get; set; } = cron;
    public string? Description { get; set; }
    public bool IsAsync { get; set; } = false;
    public bool IsStartOnInit { get; set; } = false;
}
public interface IScheduleMetadataStore
{
    /// <summary>
    /// 获取所有ScheduleTaskMetadata
    /// </summary>
    Task<IEnumerable<ScheduleTaskMetadata>> GetAllAsync();
}

实现一个Configuration级别的Store

internal class ConfigurationScheduleMetadataStore(IConfiguration configuration) : IScheduleMetadataStore
{
    const string Key = "BiwenQuickApi:Schedules";

    public Task<IEnumerable<ScheduleTaskMetadata>> GetAllAsync()
    {
        var options = configuration.GetSection(Key).GetChildren();

        if (options?.Any() is true)
        {
            var metadatas = options.Select(x =>
            {
                var type = Type.GetType(x[nameof(ConfigurationScheduleOption.ScheduleType)]!);
                if (type is null)
                    throw new ArgumentException($"Type {x[nameof(ConfigurationScheduleOption.ScheduleType)]} not found!");

                return new ScheduleTaskMetadata(type, x[nameof(ConfigurationScheduleOption.Cron)]!)
                {
                    Description = x[nameof(ConfigurationScheduleOption.Description)],
                    IsAsync = string.IsNullOrEmpty(x[nameof(ConfigurationScheduleOption.IsAsync)]) ? false : bool.Parse(x[nameof(ConfigurationScheduleOption.IsAsync)]!),
                    IsStartOnInit = string.IsNullOrEmpty(x[nameof(ConfigurationScheduleOption.IsStartOnInit)]) ? false : bool.Parse(x[nameof(ConfigurationScheduleOption.IsStartOnInit)]!),
                };
            });
            return Task.FromResult(metadatas);
        }
        return Task.FromResult(Enumerable.Empty<ScheduleTaskMetadata>());
    }
}

然后,我们可能需要多任务调度的事件做一些操作或者日志存储。

比如失败了该干嘛,完成了回调其他后续业务等。

我们再来定义一下具体的事件IEvent,具体可以参考文章: https://www.cnblogs.com/vipwan/p/18184088

事件IEvent代码

1、首先定义一个事件约定的空接口

public interface IEvent{}

2、然后定义事件订阅者接口

public interface IEventSubscriber<T> where T : IEvent
{
    Task HandleAsync(T @event, CancellationToken ct);
    /// <summary>
    /// 执行排序
    /// </summary>
    int Order { get; }

    /// <summary>
    /// 如果发生错误是否抛出异常,将阻塞后续Handler
    /// </summary>
    bool ThrowIfError { get; }
}
public abstract class EventSubscriber<T> : IEventSubscriber<T> where T : IEvent
{
    public abstract Task HandleAsync(T @event, CancellationToken ct);
    public virtual int Order => 0;
    /// <summary>
    /// 默认不抛出异常
    /// </summary>
    public virtual bool ThrowIfError => false;
}

3、接着就是发布者

internal class Publisher(IServiceProvider serviceProvider)
{
 public async Task PublishAsync<T>(T @event, CancellationToken ct) where T : IEvent
 {
  var handlers = serviceProvider.GetServices<IEventSubscriber<T>>();
  if (handlers is null) return;
  foreach (var handler in handlers.OrderBy(x => x.Order))
  {
   try
   {
    await handler.HandleAsync(@event, ct);
   }
   catch
   {
    if (handler.ThrowIfError)
    {
     throw;
    }
    //todo:
   }
  }
 }
}

4、到此发布订阅的基本代码也就写完了.接下来就是注册发布者和所有的订阅者了

public abstract class ScheduleTaskEvent(IScheduleTask scheduleTask, DateTime eventTime) : IEvent
{
    /// <summary>
    /// 任务
    /// </summary>
    public IScheduleTask ScheduleTask { get; set; } = scheduleTask;
    /// <summary>
    /// 触发时间
    /// </summary>
    public DateTime EventTime { get; set; } = eventTime;
}
/// <summary>
/// 执行完成
/// </summary>
public sealed class TaskSuccessedEvent(IScheduleTask scheduleTask, DateTime eventTime, DateTime endTime) : ScheduleTaskEvent(scheduleTask, eventTime)
{
    /// <summary>
    /// 执行结束的时间
    /// </summary>
    public DateTime EndTime { get; set; } = endTime;
}
/// <summary>
/// 执行开始
/// </summary>
public sealed class TaskStartedEvent(IScheduleTask scheduleTask, DateTime eventTime) : ScheduleTaskEvent(scheduleTask, eventTime);
/// <summary>
/// 执行失败
/// </summary>
public sealed class TaskFailedEvent(IScheduleTask scheduleTask, DateTime eventTime, Exception exception) : ScheduleTaskEvent(scheduleTask, eventTime)
{
    /// <summary>
    /// 异常信息
    /// </summary>
    public Exception Exception { get; private set; } = exception;
}

接下来我们再实现基于NCrontab的简易调度器,这个调度器主要是解析Cron表达式判断传入时间是否可以执行ScheduleTask,具体的代码:

internal class SampleNCrontabScheduler : IScheduler
{
    /// <summary>
    /// 暂存上次执行时间
    /// </summary>
    private static ConcurrentDictionary<ScheduleTaskAttribute, DateTime> LastRunTimes = new();

    public bool CanRun(ScheduleTaskAttribute scheduleMetadata, DateTime referenceTime)
    {
        var now = DateTime.Now;
        var haveExcuteTime = LastRunTimes.TryGetValue(scheduleMetadata, out var time);
        if (!haveExcuteTime)
        {
            var nextStartTime = CrontabSchedule.Parse(scheduleMetadata.Cron).GetNextOccurrence(referenceTime);
            LastRunTimes.TryAdd(scheduleMetadata, nextStartTime);

            //如果不是初始化启动,则不执行
            if (!scheduleMetadata.IsStartOnInit)
                return false;
        }
        if (now >= time)
        {
            var nextStartTime = CrontabSchedule.Parse(scheduleMetadata.Cron).GetNextOccurrence(referenceTime);
            //更新下次执行时间
            LastRunTimes.TryUpdate(scheduleMetadata, nextStartTime, time);
            return true;
        }
        return false;
    }
}

然后就是核心的BackgroundService了,这里我用的IdleTime心跳来实现,粒度分钟,当然内部也可以封装Timer等实现更复杂精度更高的调度,这里就不展开讲了。

代码如下:

internal class ScheduleBackgroundService : BackgroundService
{
    private static readonly TimeSpan _pollingTime
DEBUG
      //轮询20s 测试环境下,方便测试。
      = TimeSpan.FromSeconds(20);
if
!DEBUG
     //轮询60s 正式环境下,考虑性能轮询时间延长到60s
     = TimeSpan.FromSeconds(60);
if
    //心跳10s.
    private static readonly TimeSpan _minIdleTime = TimeSpan.FromSeconds(10);
    private readonly ILogger<ScheduleBackgroundService> _logger;
    private readonly IServiceProvider _serviceProvider;
    public ScheduleBackgroundService(ILogger<ScheduleBackgroundService> logger, IServiceProvider serviceProvider)
    {
        _logger = logger;
        _serviceProvider = serviceProvider;
    }
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var pollingDelay = Task.Delay(_pollingTime, stoppingToken);
            try
            {
                await RunAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                //todo:
                _logger.LogError(ex.Message);
            }
            await WaitAsync(pollingDelay, stoppingToken);
        }
    }
    private async Task RunAsync(CancellationToken stoppingToken)
    {
        using var scope = _serviceProvider.CreateScope();
        var tasks = scope.ServiceProvider.GetServices<IScheduleTask>();
        if (tasks is null || !tasks.Any())
        {
            return;
        }
        //调度器
        var scheduler = scope.ServiceProvider.GetRequiredService<IScheduler>();
        async Task DoTaskAsync(IScheduleTask task, ScheduleTaskAttribute metadata)
        {
            if (scheduler.CanRun(metadata, DateTime.Now))
            {
                var eventTime = DateTime.Now;
                //通知启动
                _ = new TaskStartedEvent(task, eventTime).PublishAsync(default);
                try
                {
                    if (metadata.IsAsync)
                    {
                        //异步执行
                        _ = task.ExecuteAsync();
                    }
                    else
                    {
                        //同步执行
                        await task.ExecuteAsync();
                    }
                    //执行完成
                    _ = new TaskSuccessedEvent(task, eventTime, DateTime.Now).PublishAsync(default);
                }
                catch (Exception ex)
                {
                    _ = new TaskFailedEvent(task, DateTime.Now, ex).PublishAsync(default);
                }
            }
        };
        //注解中的task
        foreach (var task in tasks)
        {
            if (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            //标注的metadatas
            var metadatas = task.GetType().GetCustomAttributes<ScheduleTaskAttribute>();

            if (!metadatas.Any())
            {
                continue;
            }
            foreach (var metadata in metadatas)
            {
                await DoTaskAsync(task, metadata);
            }
        }
        //store中的scheduler
        var stores = _serviceProvider.GetServices<IScheduleMetadataStore>().ToArray();

        //并行执行,提高性能
        Parallel.ForEach(stores, async store =>
        {
            if (stoppingToken.IsCancellationRequested)
            {
                return;
            }
            var metadatas = await store.GetAllAsync();
            if (metadatas is null || !metadatas.Any())
            {
                return;
            }
            foreach (var metadata in metadatas)
            {
                var attr = new ScheduleTaskAttribute(metadata.Cron)
                {
                    Description = metadata.Description,
                    IsAsync = metadata.IsAsync,
                    IsStartOnInit = metadata.IsStartOnInit,
                };

                var task = scope.ServiceProvider.GetRequiredService(metadata.ScheduleTaskType) as IScheduleTask;
                if (task is null)
                {
                    return;
                }
                await DoTaskAsync(task, attr);
            }
        });
    }

    private static async Task WaitAsync(Task pollingDelay, CancellationToken stoppingToken)
    {
        try
        {
            await Task.Delay(_minIdleTime, stoppingToken);
            await pollingDelay;
        }
        catch (OperationCanceledException)
        {
        }
    }
}

最后收尾阶段我们老规矩扩展一下IServiceCollection:

internal static IServiceCollection AddScheduleTask(this IServiceCollection services)
{
    foreach (var task in ScheduleTasks)
    {
        services.AddTransient(task);
        services.AddTransient(typeof(IScheduleTask), task);
    }
    //调度器
    services.AddScheduler<SampleNCrontabScheduler>();
    //配置文件Store:
ices.AddScheduleMetadataStore<ConfigurationScheduleMetadataStore>();
    //BackgroundService
   services.AddHostedService<ScheduleBackgroundService>();
    return services;
}
/// <summary>
/// 注册调度器AddScheduler
/// </summary>
public static IServiceCollection AddScheduler<T>(this IServiceCollection services) where T : class, IScheduler
{
    services.AddSingleton<IScheduler, T>();
    return services;
}

/// <summary>
/// 注册ScheduleMetadataStore
/// </summary>
public static IServiceCollection AddScheduleMetadataStore<T>(this IServiceCollection services) where T : class, IScheduleMetadataStore
{
    services.AddSingleton<IScheduleMetadataStore, T>();
    return services;
}

老规矩我们来测试一下:

//通过特性标注的方式执行:
[ScheduleTask(Constants.CronEveryMinute)] //每分钟一次
[ScheduleTask("0/3 * * * *")]//每3分钟执行一次
public class KeepAlive(ILogger<KeepAlive> logger) : IScheduleTask
{
    public async Task ExecuteAsync()
    {
        //执行5s
        await Task.Delay(TimeSpan.FromSeconds(5));
        logger.LogInformation("keep alive!");
    }
}
public class DemoConfigTask(ILogger<DemoConfigTask> logger) : IScheduleTask
{
    public Task ExecuteAsync()
    {
        logger.LogInformation("Demo Config Schedule Done!");
        return Task.CompletedTask;
    }
}

通过配置文件的方式配置Store:

{
  "BiwenQuickApi": {
    "Schedules": [
      {
        "ScheduleType": "Biwen.QuickApi.DemoWeb.Schedules.DemoConfigTask,Biwen.QuickApi.DemoWeb",
        "Cron": "0/5 * * * *",
        "Description": "Every 5 mins",
        "IsAsync": true,
        "IsStartOnInit": false
      },
      {
        "ScheduleType": "Biwen.QuickApi.DemoWeb.Schedules.DemoConfigTask,Biwen.QuickApi.DemoWeb",
        "Cron": "0/10 * * * *",
        "Description": "Every 10 mins",
        "IsAsync": false,
        "IsStartOnInit": true
      }
    ]
  }
}

我们还可以实现自己的Store,这里以放到内存为例,如果有兴趣 你可以可以自行开发一个面板管理:

public class DemoStore : IScheduleMetadataStore
{
    public Task<IEnumerable<ScheduleTaskMetadata>> GetAllAsync()
    {
        //模拟从数据库或配置文件中获取ScheduleTaskMetadata
        IEnumerable<ScheduleTaskMetadata> metadatas =
            [
                new ScheduleTaskMetadata(typeof(DemoTask),Constants.CronEveryNMinutes(2))
                {
                    Description="测试的Schedule"
                },
            ];
        return Task.FromResult(metadatas);
    }
}
//然后注册这个Store:
builder.Services.AddScheduleMetadataStore<DemoStore>();

所有的一切都大功告成,最后我们来跑一下Demo,成功了

图片

当然这里是自己的固定思维设计的一个简约版,还存在一些不足,欢迎板砖轻拍指正!

提供同一时间单一运行中的任务实现

/// <summary>
/// 模拟一个只能同时存在一个的任务.一分钟执行一次,但是耗时两分钟.
/// </summary>
/// <param name="logger"></param>
[ScheduleTask(Constants.CronEveryMinute, IsStartOnInit = true)]
public class OnlyOneTask(ILogger<OnlyOneTask> logger) : OnlyOneRunningScheduleTask
{
    public override Task OnAbort()
    {
        logger.LogWarning($"[{DateTime.Now}]任务被打断.因为有一个相同的任务正在执行!");
        return Task.CompletedTask;
    }

    public override async Task ExecuteAsync()
    {
        var now = DateTime.Now;
        //模拟一个耗时2分钟的任务
        await Task.Delay(TimeSpan.FromMinutes(2));
        logger.LogInformation($"[{now}] ~ {DateTime.Now} 执行一个耗时两分钟的任务!");
    }
}

源码地址

https://github.com/vipwan/Biwen.QuickApi

https://github.com/vipwan/Biwen.QuickApi/tree/master/Biwen.QuickApi/Scheduling

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/654263.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

centos7防火墙入站白名单配置

firewall-cmd --set-default-zonedropfirewall-cmd --get-active-zone记录下当前激活网卡firewall-cmd --permanent --change-interfaceens33 --zonedrop firewall-cmd --zonedrop --list-all 添加信任的源IP和开放端口 firewall-cmd --permanent --add-source192.168.254.1 -…

pikachu—exec“eval“

这是原画面 然后呢&#xff1f; 我们知道会传入到后台rce_eval.php来处理然后通过 eval()是啥? 在eval括号里面可以执行外来机器的命令 然后我们通过php的一个内置的命令 我们通过phpinfo()&#xff1b; 这是输入后的结果

华为机考入门python3--(26)牛客26-字符串排序

分类&#xff1a;字符串 知识点&#xff1a; 字符串是否仅由字母构成 my_str.isalpha() 字母列表按小写排序 letters.sort(keylambda x: x.lower()) 题目来自【牛客】 def custom_sort(input_str):letters []non_letters []for char in input_str:if char.isalpha…

打工人都在偷偷做的副业项目—steam搬砖

steam搬砖其实是一个非常老牌的项目了&#xff0c;之前只有玩游戏玩市场的人知道&#xff0c;其他普通人都不知道。 我们陪跑这个项目不是说这个项目不行了&#xff0c;再拿出来割韭菜&#xff0c;现在依然可以做&#xff0c;我们本身就是项目和培训一比一在做&#xff0c;这一…

R语言学习 - 箱线图一步法

箱线图 - 一步绘制 绘图时通常会碰到两个头疼的问题&#xff1a; 1、有时需要绘制很多的图&#xff0c;唯一的不同就是输入文件&#xff0c;其它都不需要修改。如果用R脚本&#xff0c;需要反复替换文件名&#xff0c;繁琐又容易出错。 (R也有命令行参数&#xff0c;不熟&…

关于高性能滤波器和普通型滤波器的区别说明

高性能滤波器和普通型滤波器在性能和滤波效果上存在显著差异。以三安培为代表分析高性能滤波器和普通型滤波器的区别&#xff1a; 从上图曲线可看出&#xff1a; 1.高性能滤波器和普通型滤波器的滤波范围不同。普通型滤波器有效滤波范围为 150KHz~30MHz&#xff0c;而高性能滤…

QColor官网文档简介

颜色可以用不同的模型来进行表示,常见的有RGB,HSV或者CMYK.QColor基于RGB值创建颜色,如果要转接到HSV或者CMYK需要使用转接函数toHsv() toCmyk() 返回期望格式的副本 get...和set...函数具体参见文档用isValid()检查RGB颜色是否合法颜色组成部分可以单独检索QRgb是一种无符号整…

Hololens 2 新建自定义按钮

官方链接地址 1、创建Cube 2、添加PressableButton脚本&#xff0c;并点击AddNearin… 3、把Cube拖入到MovingButtonVisuals变量中 4、点击NearInteractionTouchable组件&#xff08;这个组件是添加和上一个脚本绑定的&#xff0c;自动添加上来的&#xff09;上的Fix… 5、…

一文搞懂oracle事务提交以及脏数据落盘的原则

本文基于oracle 19c 做事务提交以及oracle脏数据落盘的相关解读 第一章 相关进程及组件介绍&#xff1a; 1.LGWR&#xff1a; 重做日志条目在系统全局区域 &#xff08;SGA&#xff09; 的重做日志缓冲区中生成。LGWR 按顺序将重做日志条目写入重做日志文件。如果数据库具有…

14.Redis之JAVASpring客户端

1.引入依赖 此时就会引入操作 redis 的依赖了~~ 2.yml配置 spring:redis:host: 127.0.0.1port: 8888 3.准备 前面使用 jedis,是通过 Jedis 对象里的各种方法来操作 redis 的.此处Spring 中则是通过 StringRedisTemplate 来操作 redis .最原始提供的类是 RedisTemplateStrin…

OceanBase:列存储

目录 1、列存储的定义 1、默认创建列存表 3、指定创建列存表 4、指定创建列存行存冗余表 5、行、列存储查询测试 1、列存储的定义 行存储&#xff08;Row-based Storage&#xff09;&#xff1a;行存储是以行为单位进行组织和存储数据。在这一模式下&#xff0c;数据库将…

C++ Primer 第五版 第十三章 拷贝控制

当定义一个类时&#xff0c;我们显式地或隐式地指定在此类型的对象拷贝、移动、赋值和销毁时做什么。一个类通过定义五种特殊的成员函数来控制这些操作&#xff0c;包括&#xff1a;拷贝构造函数&#xff08;copy constructor&#xff09;、拷贝赋值运算符&#xff08;copy-ass…

web自动化的断言和日志封装

断言 UI自动化常见的断言条件包括&#xff1a; 通过当前页面的URL地址通过当前页面的标题通过当前页面的提示文本信息通过当前页面的某些元素变化/显示 一句话总结&#xff1a;通过肉眼观察页面的变化检查。 【用代码模仿人的识别页面】 一般断言写一条就够了&#xff0c;如…

Visual Studio 智能代码插件:CodeGeeX

前言 在软件开发领域&#xff0c;高效的编程助手一直是提升开发者效率和质量的关键。 随着人工智能技术的不断发展&#xff0c;智能编程助手逐渐成为开发者们不可或缺的工具。其中&#xff0c;CodeGeeX作为一款专为Visual Studio设计的免费智能编程助手&#xff0c;凭借其强大…

goimghdr,一个有趣的 Python 库!

更多Python学习内容&#xff1a;ipengtao.com 大家好&#xff0c;今天为大家分享一个有趣的 Python 库 - goimghdr。 Github地址&#xff1a;https://github.com/corona10/goimghdr 在图像处理和分析过程中&#xff0c;识别图像文件的类型是一个常见的需求。Python自带的imghdr…

AT32F415 使用embedded studio笔记

由于Embedded Studio还没有支持雅特力的支持包&#xff0c;因此通过MDK的工程无法直接导入到embedded studio中。需要自己新建一个工程。 新建的时候是可以选择雅特力的单片机的。 工程新建好后&#xff0c;默认会包含3个文件。如下图所示。其中两个汇编文件的作用是之前MDK的…

出现 Transaction rolled back because it has been marked as rollback-only 解决方法

目录 1. 问题所示2. 原理分析3. 解决方法1. 问题所示 用户反馈的Bug如下所示: Transaction rolled back because it has been marked as rollback-only截图如下: 浏览器终端同样显示: 2. 原理分析 错误表明,在事务的生命周期内,遇到了某个异常或条件,导致该事务被标记…

Vue3学习使用axios和qs进行POST请求和响应处理

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 一、前言1.准备工作2.发送POST请求3.处理响应数据4.总结 一、前言 在前端开发中&#xff0c;经常需要与后端进行数据交互&#xff0c;其中包括发送POST请求并处理响…

2024年中国金融行业网络安全市场全景图

网络安全一直是国家安全的核心组成部分&#xff0c;特别是在金融行业&#xff0c;金融机构拥有大量的敏感数据&#xff0c;包括个人信息、交易记录、财务报告等&#xff0c;这些数据的安全直接关系到消费者的利益和金融市场的稳定&#xff0c;因此金融行业在网络安全建设领域一…

【机器学习】利用机器学习优化陆军战术决策与战场态势感知

&#x1f512;文章目录&#xff1a; &#x1f4a5;1.引言 &#x1f6f4;2.机器学习在陆军战术决策中的应用 &#x1f6e3;️2.1数据收集与预处理 &#x1f304;2.2模型构建与训练&#xff1a; &#x1f305;2.3实时决策支持&#xff1a; &#x1f305;2.4代码实现 &…