Parallele foreach mit asynchronem Lambda

Question

Ich würde gerne eine Sammlung parallel behandeln, aber ich habe Probleme bei der Umsetzung und ich hoffe daher auf etwas Hilfe.

Der Fehler tritt auf, wenn ich eine Methode aufrufen möchte, die in C# async innerhalb des Lambda der parallelen Schleife markiert ist. Zum Beispiel:

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; 

Das Problem tritt bei der Zählung 0 ist, weil alle erstellten Threads effektiv nur Hintergrund-Threads sind und der Parallel.ForEach Anruf wartet nicht für die Fertigstellung. Wenn ich das Asynchron-Schlüsselwort zu entfernen, sieht das Verfahren wie folgt aus:

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; 

Es funktioniert, aber es vollständig deaktiviert erwarten die Klugheit, und ich habe einige manuelle Ausnahmebehandlung zu tun .. (der Kürze halber entfernt).

Wie kann ich eine Parallel.ForEach Schleife implementieren, die das Schlüsselwort wait innerhalb des Lambda verwendet? Ist es möglich?

Der Prototyp der Parallel.ForEach-Methode benötigt einen Action<T> als Parameter, aber ich möchte, dass er auf mein asynchrones Lambda wartet.

Answer 1

Wenn Sie nur einfache Parallelität möchten, können Sie dies tun:

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; 

Wenn Sie etwas komplexere benötigen, Stephen Toub's ForEachAsync post überprüfen.

Answer 2

Sie können die ParallelForEachAsync Erweiterungsmethode von 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;