如何在 .NET 中自定义任务调度？

Question

我正在努力实现以下目标：

主线程应该创建一些任务（实际上，可能有很多线程，但现在让我们简化一些事情以避免并发和并行性带来额外的复杂性），这些任务以某种方式在以后安排和执行。可能是一些Timer 在固定的时间间隔内，可能稍后在同一个线程上阻塞这些任务时，可能是另一个专门为此任务指定的线程——调度的实现目前还没有很好的定义，我只是试图理解基本思想。 我希望我的代码看起来像这样：

'Somewhere in the main thread...
Dim MyTask = CreateTask(Of Integer)(Function()
    Console.WriteLine("Task has been called!")
    'Some activity...
    Return 42
End Sub)
'...
UseTheResult(MyTask.Result)

...其中CreateTask 是响应将委托安排为稍后执行的任务的子例程。

问题是，我不知道如何实现这个概念。起初，我尝试使用SynchronizationContext，但它似乎更多的是私有线程内存，与调度任务无关（尽管 Windows UI 似乎以某种方式使用它来实现这个目标 - 我不'还没有清楚地理解所有这些上下文）。然后我偶然发现了TaskScheduler，它与TaskFactory 相结合，似乎就是在做这件事。我试图实现这个线程，但我的代码没有按预期工作，这意味着我错过了一些关于TaskScheduler 工作方式的非常重要的东西，或者更糟糕的是——关于整个概念。我已经阅读了一些关于该主题的文章（最值得注意的是，TaskScheduler 上的 MSDN documentation 和代码示例），但仍然不明白这是如何工作的。

我的未按预期工作的代码，它应该安排然后执行任务之间的延迟为 100 毫秒（因此整个测试应该已经运行了大约 10 秒）：

Public Sub Main()
  Dim T As New TaskFactory(New TestScheduler)
  Dim tasks As New List(Of Task)
  For I = 1 To 100
    Dim J = I
    tasks.Add(T.StartNew(Sub() Console.WriteLine(J)))
  Next I
  Task.WaitAll(tasks.ToArray)

  Console.WriteLine("Test end")
  Console.ReadLine()
End Sub
Class TestScheduler : Inherits TaskScheduler
  Dim Tasks As New Collections.Concurrent.ConcurrentQueue(Of Task)
  'Dim Counter As New Threading.AutoResetEvent(False)
  Dim CTimer As New Threading.Timer(AddressOf TimerUp, Nothing, 0, 100)
  Protected Overrides Function GetScheduledTasks() As IEnumerable(Of Task)
    Return Tasks
  End Function

  Private Sub TimerUp(State As Object)
    Static AllTasks% = 0
    Dim T As task = Nothing
    If Tasks.TryDequeue(T) Then
      Console.WriteLine("Task... {0}", AllTasks) 'for debugging
      AllTasks += 1
      If Not MyBase.TryExecuteTask(T) Then
        QueueTask(T)
      End If
    End If
  End Sub

  Protected Overrides Sub QueueTask(task As Task)
    Tasks.Enqueue(task)
  End Sub

  Protected Overrides Function TryExecuteTaskInline(task As Task, taskWasPreviouslyQueued As Boolean) As Boolean
    If taskWasPreviouslyQueued Then Return False
    Return MyBase.TryExecuteTask(task)
  End Function
End Class
'Output:
'2
'1
'2
'... and that's it, no even "Test end" or any other output whatsoever.
'I have strong feeling that it throws an exception somewhere; however,
'when I try to debug it, it silently stops, without waiting for <Enter>,
'and when I run it without debugging - it produces the abovementioned output without
'any exceptions or messages.

我有一种强烈的感觉，我在这里误解了一些基本概念。

所以，总而言之，我的问题是：如何在 .NET 中自定义任务调度？

我应该使用什么 - SynchronizationContext、TaskScheduler、派生类或其他什么？如果是这样，我应该覆盖哪些确切的方法，它们的含义是什么？所有这些QueueTask、TryExecuteTaskInline 和TryExecuteTask 简直让我发疯……

P.S.：请原谅，如果您愿意，请纠正我在那里犯的任何语法或逻辑错误。英语不是我的母语，目前我的想法不是很清楚[睡着了]。谢谢！

编辑： 非常抱歉误导大家。代码运行非常好，我刚刚犯了一个非常粗暴的错误，忘记将我的项目标记为“启动” - 因此不小心在同一个解决方案中运行另一个测试项目......起初我打算结束这个问题，但后来我意识到最好还是回答它 - 它应该可以帮助任何有同样问题的人，最重要的是，允许一个更好的解决方案：到目前为止还剩下很多问题没有回答。为什么我们需要TaskScheduler.TryExecuteTask - 为什么要“尝试”？这个方法什么时候可以返回False？ TryExecuteTaskInline 方法何时运行，何时可以使用taskWasPreviouslyQueued = true 执行？我有一些假设，但同样，它们可能是错误的。

我是否需要将这些主题作为单独的问题提出，或者，因为它们仍然属于这个问题的标题，所以它们可以在那里？

Answer 1

因此，为了实现这一目标，可以使用 TaskScheduler 和 TaskFactory 的组合，这对我来说非常有效：

Public Sub Main()
  Dim T As New TaskFactory(New TestScheduler)
  Dim tasks As New List(Of Task)
  For I = 1 To 100
    Dim J = I
    tasks.Add(T.StartNew(Sub() Console.WriteLine(J)))
  Next I
  Task.WaitAll(tasks.ToArray)

  Console.WriteLine("Test end")
  Console.ReadLine()
End Sub

Public Function CreateTask(ActionToSchedule As Action) As Task
'This is exactly the needed function!
'A single static instances of both TaskFactory and custom TaskSheduler are used.
  Static Factory As New TaskFactory(New TestScheduler)
  Return Factory.StartNew(ActionToSchedule)
End Function

Class TestScheduler : Inherits TaskScheduler
  Dim Tasks As New Collections.Concurrent.ConcurrentQueue(Of Task)
  Dim CTimer As New Threading.Timer(AddressOf TimerUp, Nothing, 0, 10)
  Protected Overrides Function GetScheduledTasks() As IEnumerable(Of Task)
    Return Tasks
  End Function

  Private Sub TimerUp(State As Object)
    Static AllTasks% = 0
    Dim T As task = Nothing
    If Tasks.TryDequeue(T) Then
      Console.WriteLine("Task... {0}", AllTasks) 'for debugging purposes
      AllTasks += 1
      If Not MyBase.TryExecuteTask(T) Then
        QueueTask(T)
      End If
    End If
  End Sub

  Protected Overrides Sub QueueTask(task As Task)
    'Here you should schedule the given task for later execution, whatever that means.
    'In this case it is simply put in the queue to be executed later when the timer fires.
    Tasks.Enqueue(task)
  End Sub

  Protected Overrides Function TryExecuteTaskInline(task As Task, taskWasPreviouslyQueued As Boolean) As Boolean
    '> Attempts to execute the specified task on the current thread. 
    'As far as I understand - when the thread starts to wait for the task completion, either by trying to read
    '"task.Result" or executing "T.RunSynchronously()", this function will allow to execute the task synchronously
    'on the waiting thread instead of blocking, which may lead to a huge performance&speed gain.
    'If our task is not executing - waiting somewhere in the queue or is not even enqueued yet - we 
    'may dequeue it and run instantly to avoid the blocking.

    Return False
    'In this particular case, I've decided that the timer should run all the tasks, and, therefore, there must be done 
    'no inlining. Hovewer, the code below (obviously, it will not execute) should illustrate
    'this idea.
    If taskWasPreviouslyQueued Then
      'If the task is already enqueued, then we may try to remove it from
      'the queue and execute there instead.
      If Tasks.TryDequeue(task) Then
        Return MyBase.TryExecuteTask(task)
      Else
        'It has already been dequeued and maybe even completed already, so - no, no inlining. :)
        Return False
      End If
    Else
      'If it wasn't even enqueued - ha! No questions - we may run it now without any preparations.
      Return MyBase.TryExecuteTask(task)
    End If
  End Function
End Class

我仍然不确定是否应该发布一个被误导的问题的更正和完成作为答案，但我希望它比简单地关闭它对社区更有好处。