Windows 服务和特定数量的异步任务答案

【问题标题】：Windows service and specific number of async tasksWindows 服务和特定数量的异步任务
【发布时间】：2016-10-07 19:12:52
【问题描述】：

对于一个项目，我们需要编写一个 Windows 服务，该服务将处理一个作业队列并对不同的 API 进行必要的处理和调用，并且它将在独立的机器上运行。

上下文：此服务将每 x（例如：5）秒轮询 SQL 数据库，根据优先级和创建日期获取前 Y 个作业，然后开始处理它们。我们预计此 Windows 服务的容量很大，因此我们希望使其成为多线程或异步的。如果达到最大并行处理量，我们不希望它启动更多线程。但同时，如果我们有 7 个作业需要 30 秒，一个需要 5 分钟，我们不希望它等待 5 分钟的作业完成后再循环并开始另一批 8 个作业。

我们寻找的第一个选项是 BackgroundWorker。计时器的每次迭代都会检查每个 BackgroundWorker 的状态，如果可用，将指示它处理新作业。但是对于较新版本的 .Net 框架，异步方法已经过时了。

我们寻找的第二个选项是 Parallel.ForEach 和 awaitAll。但问题是，如果 8 个线程中有 7 个需要 1 分钟，而最后一个需要 6 分钟，我们不想等到最后一个线程完成后再开始 7 个新的作业进程。

最合适的方法是任务。

我的问题是：有没有办法跟踪多个任务的运行状态？并限制同时运行的任务数量？我是否应该在服务的 OnStart() 和计时器的每个 OnElapsed 中实例化我的所有任务，检查每个任务的状态，如果有的话，用新工作再次启动它？还是我对任务的工作方式完全错误？

允许的并行处理数量将在应用配置文件中定义，并在 Windows 服务的 OnStart 中初始化。

【问题讨论】：

您需要查看任务并行库和/或信号量。你想要的很可能，但你必须写繁重的工作。我会建议反对背景工作者。使用 TPL，您可以链接非常强大的延续任务。
我会为此查看 QuartzScheduler。我不得不解决一个类似的问题，但是因为我们启动了很多不同的外部程序，所以我最终为它创建了一个库。 Quartz 应该做你需要做的一切，并且可以使用 CRON 表达式进行配置，使其非常灵活。 quartz-scheduler.org/documentation/quartz-2.1.x/examples
类似msdn.microsoft.com/en-us/library/ee789351(v=vs.100).aspx 的东西？ This one 不一样，但可以帮助您编写自己的 TaskScheduler ....
确认@Botonomous 所说的，BackgroundWorker 将给你无尽的地狱做这件事。此外，切勿在 Windows 服务中使用 C# Thread。除非您有一些仔细的池逻辑来调节线程轮询，否则您将看到 CPU 使用率增加 30%。
@L.B 非常有趣。我以前没有见过自定义调度程序的使用，我真的很喜欢这个。

标签： c# windows-services async-await

【解决方案1】：

一种可能的方法是使用TPL Dataflow 库。您可以配置一个块（在本例中为ActionBlock）来处理处理并将项目异步发送到块中（如果您愿意，也可以同步）。您可以限制您希望允许的最大并行度，以及设置限制缓冲区中项目数量的容量。

通过使用SendAsync，您可以让您的进程异步等待，直到缓冲区中有空间容纳新项目。使用 TPL 数据流库这一切都相当简单：

var processBlock = new ActionBlock<DataItem>(
    item => ProcessItemAsync(item),
    new ExecutionDataflowBlockOptions
    {
        BoundedCapacity = 10,   // your configured limit
        MaxDegreeOfParallelism = 2 // your configured max
    });

要使用您的块，您可以执行以下操作：

using (var connection = new SqlConnection(...))
{
    while (await reader.ReadAsync().ConfigureAwait(false))
    {
        // offer a message to the processing block, but allow postponement
        // in case we've hit capacity
       await processBlock.SendAsync(item);
    }
}

【讨论】：