Я хотел бы запустить кучу асинхронных задач с ограничением количества заданий, ожидающих завершения в любой момент времени.
Скажите, что у вас 1000 URL-адресов, и вы хотите, чтобы одновременно открывалось 50 запросов; но как только один запрос завершается, вы открываете соединение со следующим URL-адресом в списке. Таким образом, всегда открыто ровно 50 подключений за раз, пока список URL не исчерпан.
Я также хочу использовать заданное количество потоков, если это возможно.
Я придумал метод расширения ThrottleTasksAsync, который делает то, что я хочу. Есть ли еще более простое решение? Я бы предположил, что это общий сценарий.
Использование:
class Program
{
static void Main(string[] args)
{
Enumerable.Range(1, 10).ThrottleTasksAsync(5, 2, async i => { Console.WriteLine(i); return i; }).Wait();
Console.WriteLine("Press a key to exit...");
Console.ReadKey(true);
}
}
Вот код:
static class IEnumerableExtensions
{
public static async Task<Result_T[]> ThrottleTasksAsync<Enumerable_T, Result_T>(this IEnumerable<Enumerable_T> enumerable, int maxConcurrentTasks, int maxDegreeOfParallelism, Func<Enumerable_T, Task<Result_T>> taskToRun)
{
var blockingQueue = new BlockingCollection<Enumerable_T>(new ConcurrentBag<Enumerable_T>());
var semaphore = new SemaphoreSlim(maxConcurrentTasks);
// Run the throttler on a separate thread.
var t = Task.Run(() =>
{
foreach (var item in enumerable)
{
// Wait for the semaphore
semaphore.Wait();
blockingQueue.Add(item);
}
blockingQueue.CompleteAdding();
});
var taskList = new List<Task<Result_T>>();
Parallel.ForEach(IterateUntilTrue(() => blockingQueue.IsCompleted), new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism },
_ =>
{
Enumerable_T item;
if (blockingQueue.TryTake(out item, 100))
{
taskList.Add(
// Run the task
taskToRun(item)
.ContinueWith(tsk =>
{
// For effect
Thread.Sleep(2000);
// Release the semaphore
semaphore.Release();
return tsk.Result;
}
)
);
}
});
// Await all the tasks.
return await Task.WhenAll(taskList);
}
static IEnumerable<bool> IterateUntilTrue(Func<bool> condition)
{
while (!condition()) yield return true;
}
}
В методе используются BlockingCollection и SemaphoreSlim, чтобы он работал. Дроссель запускается на одном потоке, и все асинхронные задачи запускаются на другом потоке. Чтобы достичь parallelism, я добавил параметр maxDegreeOfParallelism, который передавался в цикл Parallel.ForEach, повторно заданный как цикл while.
Старая версия:
foreach (var master = ...)
{
var details = ...;
Parallel.ForEach(details, detail => {
// Process each detail record here
}, new ParallelOptions { MaxDegreeOfParallelism = 15 });
// Perform the final batch updates here
}
Но пул потоков быстро исчерпается, и вы не можете сделать async/await.
Бонус:
Чтобы обойти проблему в BlockingCollection, где исключение выбрано в Take(), когда вызывается CompleteAdding(), я использую перегрузку TryTake с таймаутом. Если бы я не использовал таймаут в TryTake, он бы победил цель использования BlockingCollection, так как TryTake не будет блокироваться. Есть ли способ лучше? В идеале, существует метод TakeAsync.
