前言
至於任務排程這個基礎功能，重要性不言而喻，大多數業務系統都會用到,世面上有很多成熟的三方庫比如Quartz，Hangfire，Coravel
這裏我們不討論三方的庫如何使用 而是從0開始自己制作一個簡易的任務排程,如果只是到分鐘級別的粒度基本夠用。
正文
技術棧用到了:
BackgroundService
和
NCrontab
庫
第一步我們定義一個簡單的任務約定,不幹別的就是一個執行方法:
publicinterfaceIScheduleTask
{
Task ExecuteAsync();
}
publicabstract classScheduleTask : IScheduleTask
{
publicvirtual 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>
publicstring Cron { get; set; } = cron;
publicstring? Description { get; set; }
///<summary>
/// 是否異步執行.預設false會阻塞接下來的同類任務
///</summary>
publicbool IsAsync { get; set; } = false;
///<summary>
/// 是否初始化即啟動,預設false
///</summary>
publicbool IsStartOnInit { get; set; } = false;
}
第三步我們定義一個排程器約定,不幹別的就是判斷當前的任務是否可以執行:
publicinterfaceIScheduler
{
///<summary>
/// 判斷當前的任務是否可以執行
///</summary>
boolCanRun(ScheduleTaskAttribute scheduleMetadata, DateTime referenceTime);
}
好了,基礎步驟就完成了,如果我們需要實作配置級別的任務排程或者動態的任務排程 那我們再抽象一個Store:
public class ScheduleTaskMetadata(Type scheduleTaskType, string cron)
{
public Type ScheduleTaskType { get; set; } = scheduleTaskType;
publicstring Cron { get; set; } = cron;
publicstring? Description { get; set; }
publicbool IsAsync { get; set; } = false;
publicbool IsStartOnInit { get; set; } = false;
}
publicinterfaceIScheduleMetadataStore
{
///<summary>
/// 獲取所有ScheduleTaskMetadata
///</summary>
Task<IEnumerable<ScheduleTaskMetadata>> GetAllAsync();
}
實作一個Configuration級別的Store
internal class ConfigurationScheduleMetadataStore(IConfiguration configuration) : IScheduleMetadataStore
{
conststring Key = "BiwenQuickApi:Schedules";
public Task<IEnumerable<ScheduleTaskMetadata>> GetAllAsync()
{
var options = configuration.Getp(Key).GetChildren();
if (options?.Any() istrue)
{
var metadatas = options.Select(x =>
{
var type = Type.GetType(x[nameof(ConfigurationScheduleOption.ScheduleType)]!);
if (type isnull)
thrownew ArgumentException($"Type {x[nameof(ConfigurationScheduleOption.ScheduleType)]} not found!");
returnnew 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、首先定義一個事件約定的空介面
publicinterfaceIEvent{}
2、然後定義事件訂閱者介面
publicinterfaceIEventSubscriber<T> whereT : IEvent
{
Task HandleAsync(T @event, CancellationToken ct);
///<summary>
/// 執行排序
///</summary>
int Order { get; }
///<summary>
/// 如果發生錯誤是否丟擲異常,將阻塞後續Handler
///</summary>
bool ThrowIfError { get; }
}
publicabstract classEventSubscriber<T> : IEventSubscriber<T> whereT : IEvent
{
publicabstract Task HandleAsync(T @event, CancellationToken ct);
publicvirtualint Order => 0;
///<summary>
/// 預設不丟擲異常
///</summary>
publicvirtualbool ThrowIfError => false;
}
3、接著就是釋出者
internal class Publisher(IServiceProvider serviceProvider)
{
publicasync Task PublishAsync<T>(T @event, CancellationToken ct) where T : IEvent
{
var handlers = serviceProvider.GetServices<IEventSubscriber<T>>();
if (handlers isnull) return;
foreach (var handler in handlers.OrderBy(x => x.Order))
{
try
{
await handler.HandleAsync(@event, ct);
}
catch
{
if (handler.ThrowIfError)
{
throw;
}
//todo:
}
}
}
}
4、到此釋出訂閱的基本程式碼也就寫完了.接下來就是註冊釋出者和所有的訂閱者了
publicabstract 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>
publicsealed class TaskSuccessedEvent(IScheduleTask scheduleTask, DateTime eventTime, DateTime endTime) : ScheduleTaskEvent(scheduleTask, eventTime)
{
///<summary>
/// 執行結束的時間
///</summary>
public DateTime EndTime { get; set; } = endTime;
}
///<summary>
/// 執行開始
///</summary>
publicsealed class TaskStartedEvent(IScheduleTask scheduleTask, DateTime eventTime) : ScheduleTaskEvent(scheduleTask, eventTime);
///<summary>
/// 執行失敗
///</summary>
publicsealed class TaskFailedEvent(IScheduleTask scheduleTask, DateTime eventTime, Exception exception) : ScheduleTaskEvent(scheduleTask, eventTime)
{
///<summary>
/// 異常資訊
///</summary>
public Exception Exception { get; privateset; } = exception;
}
接下來我們再實作基於
NCrontab
的簡易排程器,這個排程器主要是解析
Cron
運算式判斷傳入時間是否可以執行ScheduleTask,具體的程式碼:
internal classSampleNCrontabScheduler : IScheduler
{
///<summary>
/// 暫存上次執行時間
///</summary>
privatestatic ConcurrentDictionary<ScheduleTaskAttribute, DateTime> LastRunTimes = new();
publicboolCanRun(ScheduleTaskAttribute scheduleMetadata, DateTime referenceTime)
{
var now = DateTime.Now;
var haveExcuteTime = LastRunTimes.TryGetValue(scheduleMetadata, outvar time);
if (!haveExcuteTime)
{
var nextStartTime = CrontabSchedule.Parse(scheduleMetadata.Cron).GetNextOccurrence(referenceTime);
LastRunTimes.TryAdd(scheduleMetadata, nextStartTime);
//如果不是初始化啟動,則不執行
if (!scheduleMetadata.IsStartOnInit)
returnfalse;
}
if (now >= time)
{
var nextStartTime = CrontabSchedule.Parse(scheduleMetadata.Cron).GetNextOccurrence(referenceTime);
//更新下次執行時間
LastRunTimes.TryUpdate(scheduleMetadata, nextStartTime, time);
returntrue;
}
returnfalse;
}
}
然後就是核心的
BackgroundService
了，這裏我用的IdleTime心跳來實作，粒度分鐘，當然內部也可以封裝
Timer
等實作更復雜精度更高的排程,這裏就不展開講了。
程式碼如下:
internal classScheduleBackgroundService : BackgroundService
{
privatestaticreadonly TimeSpan _pollingTime
DEBUG
//輪詢20s 測試環境下，方便測試。
= TimeSpan.FromSeconds(20);
if
!DEBUG
//輪詢60s 正式環境下，考慮效能輪詢時間延長到60s
= TimeSpan.FromSeconds(60);
if
//心跳10s.
privatestaticreadonly TimeSpan _minIdleTime = TimeSpan.FromSeconds(10);
privatereadonly ILogger<ScheduleBackgroundService> _logger;
privatereadonly IServiceProvider _serviceProvider;
publicScheduleBackgroundService(ILogger<ScheduleBackgroundService> logger, IServiceProvider serviceProvider)
{
_logger = logger;
_serviceProvider = serviceProvider;
}
protectedoverrideasync 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);
}
}
privateasync Task RunAsync(CancellationToken stoppingToken)
{
usingvar scope = _serviceProvider.CreateScope();
var tasks = scope.ServiceProvider.GetServices<IScheduleTask>();
if (tasks isnull || !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 isnull || !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 isnull)
{
return;
}
await DoTaskAsync(task, attr);
}
});
}
privatestaticasync Task WaitAsync(Task pollingDelay, CancellationToken stoppingToken)
{
try
{
await Task.Delay(_minIdleTime, stoppingToken);
await pollingDelay;
}
catch (OperationCanceledException)
{
}
}
}
最後收尾階段我們老規矩擴充套件一下
IServiceCollection
:
internalstatic 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>
publicstatic IServiceCollection AddScheduler<T>(this IServiceCollection services) where T : class, IScheduler
{
services.AddSingleton<IScheduler, T>();
return services;
}
///<summary>
/// 註冊ScheduleMetadataStore
///</summary>
publicstatic 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
{
publicasync 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 classDemoStore : 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
{
publicoverride Task OnAbort()
{
logger.LogWarning($"[{DateTime.Now}]任務被打斷.因為有一個相同的任務正在執行!");
return Task.CompletedTask;
}
publicoverrideasync 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
轉自：萬雅虎
連結：cnblogs.com/vipwan/p/18194062/biwen-quickapi-scheduletask
- EOF -
推薦閱讀 點選標題可跳轉
看完本文有收獲？請轉發分享給更多人
推薦關註「DotNet」，提升.Net技能
點贊和在看就是最大的支持❤️
