Параллельный foreach с асинхронным lambda

Question

Я хотел бы обрабатывать коллекцию параллельно, но у меня возникли проблемы с ее реализацией, и поэтому я надеюсь на помощь.

Проблема возникает, если я хочу вызвать метод, помеченный async в C#, в пределах лямбда параллельного цикла. Например:

var bag = new ConcurrentBag<object>(); 
Parallel.ForEach(myCollection, async item => 
{ 
    // some pre stuff 
    var response = await GetData(item); 
    bag.Add(response); 
    // some post stuff 
} 
var count = bag.Count; 

Проблема возникает с графом, равным 0, так как все потоки, созданные эффективны только фоновыми потоки и Parallel.ForEach вызова не ждать завершения. Если удалить асинхронной ключевое слово, метод выглядит следующим образом:

var bag = new ConcurrentBag<object>(); 
Parallel.ForEach(myCollection, item => 
{ 
    // some pre stuff 
    var responseTask = await GetData(item); 
    responseTask.Wait(); 
    var response = responseTask.Result; 
    bag.Add(response); 
    // some post stuff 
} 
var count = bag.Count; 

Он работает, но он полностью отключает ждут сообразительности, и я должен сделать некоторые ручной обработки исключений .. (Удалено для краткости).

Как я могу реализовать цикл Parallel.ForEach, который использует ключевое слово ожидания в лямбда? Является ли это возможным?

Прототип метода Parallel.ForEach принимает параметр Action<T> как параметр, но я хочу, чтобы он дождался моей асинхронной лямбда.

Answer 1

Если вы просто хотите простой параллелизм, вы можете сделать это:

var bag = new ConcurrentBag<object>(); 
var tasks = myCollection.Select(async item => 
{ 
    // some pre stuff 
    var response = await GetData(item); 
    bag.Add(response); 
    // some post stuff 
}); 
await Task.WhenAll(tasks); 
var count = bag.Count; 

Если вам нужно что-то более сложное, проверить Stephen Toub's ForEachAsync post.

Answer 2

Вы можете использовать метод ParallelForEachAsync расширения от AsyncEnumerator NuGet Package:

using System.Collections.Async; 

var bag = new ConcurrentBag<object>(); 
await myCollection.ParallelForEachAsync(async item => 
{ 
    // some pre stuff 
    var response = await GetData(item); 
    bag.Add(response); 
    // some post stuff 
}, maxDegreeOfParallelism: 10); 
var count = bag.Count;