如何在基于 C# 的 Windows 服务中处理以不同时间间隔并行运行的多个任务？

Posted 2023-04-12

【中文标题】如何在基于 C# 的 Windows 服务中处理以不同时间间隔并行运行的多个任务？

【英文标题】：How to handle multiple tasks running in parallel at different intervals inside a C# based Windows service?

【发布时间】：2021-11-07 02:00:50

【问题描述】：

我已经有一些在 Windows 中使用线程的经验，但大部分经验来自在 C/C++ 应用程序中使用 Win32 API 函数。然而，当谈到 .NET 应用程序时，我经常不确定如何正确处理多线程。有线程、任务、TPL 和其他各种我可以用于多线程的东西，但我不知道何时使用这些选项中的哪一个。 我目前正在开发基于 C# 的 Windows 服务，该服务需要定期验证来自不同数据源的不同数据组。实现验证本身对我来说并不是一个真正的问题，但我不确定如何处理同时运行的所有验证。 我需要一个解决方案，让我可以做以下所有事情：

以不同的（预定义的）间隔运行验证。 从一个地方控制所有不同的验证，以便我可以在必要时暂停和/或停止它们，例如当用户停止或重新启动服务时。 尽可能高效地使用系统资源以避免性能问题。

到目前为止，我只有一个类似的项目，之前我只是使用 Thread 对象结合 ManualResetEvent 和带有超时的 Thread.Join 调用来通知线程何时停止服务。这些线程中定期执行某些操作的逻辑如下所示：

while (!shutdownEvent.WaitOne(0))

    if (DateTime.Now > nextExecutionTime)
    
        // Do something
        nextExecutionTime = nextExecutionTime.AddMinutes(interval);
    

    Thread.Sleep(1000);

虽然这确实按预期工作，但我经常听说像这样直接使用线程被认为是“老派”甚至是一种不好的做法。我还认为这个解决方案不能非常有效地使用线程，因为它们大部分时间都在睡觉。我怎样才能以更现代和更有效的方式实现这样的目标？ 如果这个问题太模糊或基于意见，请告诉我，我会尽力使其尽可能具体。

【问题讨论】：

简化：使用Task 代替Thread。首选async await 语法。

相关：***.com/questions/4130194/…

不确定这个库是否有帮助 hangfire.io

【参考方案1】：

问题感觉有点宽泛，但我们可以使用提供的代码并尝试改进它。

确实，现有代码的问题在于，在大多数情况下，它使线程处于阻塞状态，而没有做任何有用的事情（休眠）。此外，线程每秒唤醒一次只是为了检查间隔，并且在大多数情况下再次进入睡眠状态，因为它还不是验证时间。为什么这样做？因为如果您要睡更长的时间 - 当您发出 shutdownEvent 信号然后加入线程时，您可能会阻塞很长时间。 Thread.Sleep 不提供根据请求中断的方式。

为了解决这两个问题，我们可以使用：

CancellationTokenSource+CancellationToken形式的合作取消机制。

Task.Delay 而不是Thread.Sleep。

例如：

async Task ValidationLoop(CancellationToken ct) 
    while (!ct.IsCancellationRequested) 
        try 
            var now = DateTime.Now;
            if (now >= _nextExecutionTime) 
                // do something
                _nextExecutionTime = _nextExecutionTime.AddMinutes(1);
            

            var waitFor = _nextExecutionTime - now;
            if (waitFor.Ticks > 0) 
                await Task.Delay(waitFor, ct);
            
        
        catch (OperationCanceledException) 
            // expected, just exit
            // otherwise, let it go and handle cancelled task 
            // at the caller of this method (returned task will be cancelled).
            return;
        
        catch (Exception) 
            // either have global exception handler here
            // or expect the task returned by this method to fail
            // and handle this condition at the caller
        
    

现在我们不再持有线程，因为await Task.Delay 不这样做。相反，在指定的时间间隔之后，它将在空闲线程池线程上执行后续代码（这比这个更复杂，但我们不会在这里详细介绍）。

我们也不需要无缘无故地每秒唤醒，因为Task.Delay 接受取消令牌作为参数。当该令牌发出信号时 - Task.Delay 将立即被异常中断，这是我们所期望的并从验证循环中中断。

要停止提供的循环，您需要使用CancellationTokenSource：

private readonly CancellationTokenSource _cts = new CancellationTokenSource();

然后您将其_cts.Token 令牌传递给提供的方法。然后，当您想发出令牌信号时，只需执行以下操作：

_cts.Cancel();

为了进一步改进资源管理 - 如果您的验证代码使用任何 IO 操作（从磁盘、网络、数据库访问等读取文件） - 使用上述操作的Async 版本。然后在执行 IO 时，您将不会保留任何不必要的线程阻塞等待。

现在您不再需要自己管理线程，而是根据需要执行的任务进行操作，让框架\操作系统为您管理线程。

【讨论】：

非常感谢您的详细回复！在退出流程之前等待所有创建的验证停止，我需要做些什么吗？我想在发出令牌信号后让所有验证有机会停止。在我的旧应用程序中，我通过一个超时的Thread.Join() 调用来执行此操作，然后通过调用Thread.Abort() 强制终止线程。

@Chris 是的，我的示例中的ValidationLoop 返回Task。您可以将该任务保存在某处（以您现在使用 Thread 的类似方式），然后在取消后调用 task.Wait(<timeout>)，类似于 Thread.Join。虽然没有类似于 Thread.Abort 的类比 - 任务的所有取消都是合作的（这意味着您尝试取消的代码应该意识到这种可能性并合作，就像我们使用 CancellationToken 所做的那样）。

@Chris 当您等待将CancellationToken 传递给它的东西时，会引发此异常。在这个例子中 - 它是由await Task.Delay(waitFor, ct) 抛出的。它不会像 Thread.Abort 那样随机抛出 - Task.Delay 中的代码处理取消并抛出此异常。这意味着取消不会强制终止验证 - 特定语句（并且这些语句通常需要很长时间）可能会引发异常，在此示例中仅由 Task.Delay 引发。

因此，如果验证正在进行并且您取消 - 它将继续直到到达 Task.Delay（或直到 while 循环检查），然后才会终止。至于验证崩溃 - 你应该只在验证本身周围放置 try catch，因为循环本身不会崩溃，因此不需要重新启动。或者您可以像示例中那样将 try catch 放入 while 循环中。

@Chris 顺便说一句，如果您的验证不使用任何 IO（数据库、文件访问等） - 您可以使用简单的计时器来实现类似的结果。只需将计时器作为回调处理程序中的第一条语句停止，然后作为最后一条语句（在验证后）再次启动计时器，以便它在_nextExecutionTime 再次触发。定时器当然也不持有线程，你可以在服务关闭时停止它们。

【参考方案2】：

您应该使用 Microsoft 的响应式框架（又名 Rx）- NuGet System.Reactive 并添加 using System.Reactive.Linq; - 然后您可以这样做：

Subject<bool> starter = new Subject<bool>();

IObservable<Unit> query =
    starter
        .StartWith(true)
        .Select(x => x
            ? Observable.Interval(TimeSpan.FromSeconds(5.0)).SelectMany(y => Observable.Start(() => Validation()))
            : Observable.Never<Unit>())
        .Switch();
        
IDisposable subscription = query.Subscribe();

每5.0 秒触发一次Validation() 方法。

当您需要暂停和恢复时，请执行以下操作：

starter.OnNext(false);
// Now paused

starter.OnNext(true);
// Now restarted.

如果您想停止这一切，请致电subscription.Dispose()。

【讨论】：

感谢您的回复！否决这个答案的人可以解释为什么吗？这个解决方案有什么问题或问题吗？或者它有什么不好的地方？