如何使用 C# 構(gòu)建自己的 task scheduler

譯文鏈接：https://www.infoworld.com/article/3063560/how-to-build-your-own-task-scheduler-in-csharp.html

TPL（任務(wù)并行庫）是 .NET Framework 最新版本中最有趣的新功能之一，在.NET Framework 4.0 中被引入。要使用TPL，您需要引用 System.Threading.Tasks 命名空間。

什么是 task schedulers？為什么需要它們？

如果我想問，Task 是如何被調(diào)度執(zhí)行的？其實有一個組件叫做 task scheduler，它專門負責(zé)調(diào)度 task。從本質(zhì)上來說，這是一個底層對象的抽象層，它可以將您的 task 通過排隊調(diào)度到線程上。

.NET Framework 為您提供了兩個 task schedulers。包括基于默認的 task scheduler 用于調(diào)度在 Thread Pool 之上，另一個 task scheduler 能夠調(diào)度在 指定對象的同步上下文上。請注意，TPL的默認 task scheduler 利用了.NET Framework 線程池。該線程池的表示類 ThreadPool 是在 System.Threading.Tasks命名空間下。

盡管默認的 task scheduler 在大多數(shù)情況下已足夠，但有時候你也許也想構(gòu)建自定義的 task scheduler 去完成個性化的功能，比如那些默認的 task scheduler 沒有提供的。此類功能可能包括 FIFO執(zhí)行，并發(fā)度等。

在C＃中擴展 TaskScheduler

要構(gòu)建自定義的 task scheduler，您需要創(chuàng)建一個類并繼承 System.Threading.Tasks.TaskScheduler 。因此，要構(gòu)建自定義的 task scheduler，您需要擴展 TaskScheduler抽象類并重寫以下方法。

• QueueTask 返回void并接受Task對象作為參數(shù)，當(dāng)一個 task 需要調(diào)度的時候就要調(diào)用這個方法

• GetScheduledTasks 返回所有已被調(diào)用的 task list， （準(zhǔn)確的說是IEnumerable）

• TryExecuteTaskInline 用于內(nèi)聯(lián)方式（即在當(dāng)前線程上）執(zhí)行task。在這種情況下，無需排隊即可執(zhí)行 task

下面的代碼段展示了如何繼承 TaskScheduler 類來實現(xiàn)我們自定義的 scheduler。

????public?class?CustomTaskScheduler?:?TaskScheduler,?IDisposable
????{
????}

正如在本文前面所討論的，您需要在 CustomTaskScheduler 中重寫下面3個方法：GetScheduledTasks，QueueTask和TryExecuteTaskInline。

public?sealed?class?CustomTaskScheduler?:?TaskScheduler,?IDisposable
??{
????????protected?override?IEnumerable?GetScheduledTasks()
????????{
????????????//TODO
????????}
????????protected?override?void?QueueTask(Task?task)
????????{
?????????????//TODO
????????}
????????protected?override?bool?TryExecuteTaskInline(Task?task,?bool?taskWasPreviouslyQueued)
????????{
????????????//TODO
????????}
????????public?void?Dispose()
????????{
????????????//TODO
????????}
??}

使用 BlockingCollection 來存儲 task對象集合

現(xiàn)在我們開始實現(xiàn)自定義的 task scheduler。以下代碼段顯示了如何利用 BlockingCollection 存儲 task對象的集合。

?public?sealed?class?CustomTaskScheduler?:?TaskScheduler,?IDisposable
?{
????????private?BlockingCollection?tasksCollection?=?new?BlockingCollection();
????????private?readonly?Thread?mainThread?=?null;
????????public?CustomTaskScheduler()
????????{
????????????mainThread?=?new?Thread(new?ThreadStart(Execute));
????????????if?(!mainThread.IsAlive)
????????????{
????????????????mainThread.Start();
????????????}
????????}
????????private?void?Execute()
????????{
????????????foreach?(var?task?in?tasksCollection.GetConsumingEnumerable())
????????????{
????????????????TryExecuteTask(task);
????????????}
????????}?
??????//Other?methods
??}

可以著重看看 CustomTaskScheduler 類的構(gòu)造函數(shù)，一個 thread 是如何被創(chuàng)建并且如何開始調(diào)度 Execute 方法的。

實現(xiàn) GetScheduledTasks，QueueTask 和 TryExecuteTaskInline

接下來，我們需要在 CustomTaskScheduler 中實現(xiàn)三個自定義方法。這三種方法包括 GetScheduledTasks，QueueTask 和 TryExecuteTaskInline。

GetScheduledTasks方法返回 IEnumerable 類型的 task 集合,這樣你就可以迭代這個 collection，就像上面 Execute 方法一樣。QueueTask方法接受Task對象作為方法參數(shù)，然后將其存儲在 task collection 中，TryExecuteTaskInline方法暫不實現(xiàn)， 我準(zhǔn)備將它留給讀者來實現(xiàn)。

????????protected?override?IEnumerable?GetScheduledTasks()
????????{
????????????return?tasksCollection.ToArray();
????????}
????????protected?override?void?QueueTask(Task?task)
????????{
????????????if?(task?!=?null)
????????????????tasksCollection.Add(task);
????????}
????????protected?override?bool?TryExecuteTaskInline(Task?task,?bool?taskWasPreviouslyQueued)
????????{
????????????return?false;
????????}

完整的 CustomTaskScheduler 代碼如下

下面是最終版本的 ?CustomTaskScheduler ?代碼清單：

????public?sealed?class?CustomTaskScheduler?:?TaskScheduler,?IDisposable
????{
????????private?BlockingCollection?tasksCollection?=?new?BlockingCollection();
????????private?readonly?Thread?mainThread?=?null;
????????public?CustomTaskScheduler()
????????{
????????????mainThread?=?new?Thread(new?ThreadStart(Execute));
????????????if?(!mainThread.IsAlive)
????????????{
????????????????mainThread.Start();
????????????}
????????}
????????private?void?Execute()
????????{
????????????foreach?(var?task?in?tasksCollection.GetConsumingEnumerable())
????????????{
????????????????TryExecuteTask(task);
????????????}
????????}
????????protected?override?IEnumerable?GetScheduledTasks()
????????{
????????????return?tasksCollection.ToArray();
????????}
????????protected?override?void?QueueTask(Task?task)
????????{
????????????if?(task?!=?null)
????????????????tasksCollection.Add(task);???????????
????????}
????????protected?override?bool?TryExecuteTaskInline(Task?task,?bool?taskWasPreviouslyQueued)
????????{
????????????return?false;
????????}
????????private?void?Dispose(bool?disposing)
????????{
????????????if?(!disposing)?return;
????????????tasksCollection.CompleteAdding();
????????????tasksCollection.Dispose();
????????}
????????public?void?Dispose()
????????{
????????????Dispose(true);
????????????GC.SuppressFinalize(this);
????????}
????}

要使用我們剛剛實現(xiàn)的 CustomTaskScheduler ，您可以使用以下代碼段：

CustomTaskScheduler?taskScheduler?=?new?CustomTaskScheduler();
Task.Factory.StartNew(()?=>?SomeMethod(),?CancellationToken.None,?TaskCreationOptions.None,?taskScheduler);