Служба Windows и конкретное число задач async

Question

Для проекта нам необходимо закодировать службу Windows, которая будет обрабатывать очередь заданий и выполнять необходимую обработку и вызовы для разных API, и она будет работать на автономной машине.

Контекст: Эта служба будет опроса базы данных SQL каждые x (например: 5) секунд, получить верхние задания Y в соответствии с приоритетом и датой создания, а затем начать их обработку. Мы ожидаем огромный объем для этой службы Windows, поэтому мы хотим сделать ее многопоточной или асинхронной. Если достигнут максимальный объем параллельной обработки, мы не хотим, чтобы он запускал больше потоков. Но в то же время, если у нас есть 7 рабочих мест, занимающих 30 секунд, а один занимает 5 минут, мы не хотим, чтобы он дождался завершения 5-минутной работы до начала цикла и начала другой партии по 8.

Первый вариант искал BackgroundWorker. Каждая итерация таймера проверяет статус каждого BackgroundWorker, и, если он доступен, будет инструктировать его для обработки нового задания. Но с более новыми версиями .NET Framework они устарели с помощью метода async.

Второй вариант, который мы искали, - это Parallel.ForEach и awaitAll. Но проблема в том, что если 7 из 8 потоков занимают 1 минуту, но последний занимает 6 минут, мы не хотим дождаться окончания последнего, прежде чем запускать 7 новых рабочих процессов.

Самый подходящий способ сделать это - задачи.

Мои вопросы: Есть ли способ отслеживать состояние выполнения нескольких задач? И ограничить количество запущенных одновременно задач? Должен ли я инициировать все свои задачи в OnStart() службы, и каждый OnElapsed моего таймера проверяет состояние каждой задачи и, если доступно, снова запускает ее с новым заданием? Или я совершенно не прав, как работают задачи?

Количество разрешенных параллельных вычислений будет определено в файле конфигурации приложения и инициализировано в службе OnStart для Windows.

Answer 1

Один из возможных способов сделать это - использовать библиотеку TPL Dataflow. Вы можете настроить блок (ActionBlock в этом случае) для обработки обработки и отправки элементов в блок асинхронно (или синхронно, если хотите). Вы можете ограничить максимальные уровни параллелизма, которые вы хотите разрешить, а также установить способность дросселировать количество элементов в буфере.

Используя SendAsync, вы можете позволить асинхронному процессу ждать, пока в буфере появятся новые места. Используя библиотеку TPL DataFlow это все довольно просто:

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); 
    } 
}