Nisten warten in Parallel.ForEach

Die ganze Idee hinter Parallel.ForEach() ist, dass Sie haben eine Reihe von threads, und jeder thread bearbeitet einen Teil der Sammlung. Als Sie bemerkte, das funktioniert nicht mit async-await, wo Sie wollen, lassen Sie die Gewinde für die Dauer der async-Aufruf.

Könnte man "fix", dass durch die Blockierung der ForEach() threads, aber das Niederlagen der springende Punkt, der async-await.

Was Sie tun können, ist die Verwendung TPL Dataflow statt Parallel.ForEach() unterstützt asynchrone Tasks gut.

Insbesondere Ihren code geschrieben werden kann, mit einem TransformBlock, verwandelt jede id in ein Customer mit der async lambda. Dieser block kann so konfiguriert werden, parallel ausgeführt werden. Sie würde link-block auf eine ActionBlock schreibt jeder Customer auf der Konsole.
Nach einrichten der sperren-Netzwerk können Sie Post() jede id, um die TransformBlock.

Code:

var ids = new List<string> { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };

var getCustomerBlock = new TransformBlock<string, Customer>(
    async i =>
    {
        ICustomerRepo repo = new CustomerRepo();
        return await repo.GetCustomer(i);
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    });
var writeCustomerBlock = new ActionBlock<Customer>(c => Console.WriteLine(c.ID));
getCustomerBlock.LinkTo(
    writeCustomerBlock, new DataflowLinkOptions
    {
        PropagateCompletion = true
    });

foreach (var id in ids)
    getCustomerBlock.Post(id);

getCustomerBlock.Complete();
writeCustomerBlock.Completion.Wait();

Obwohl Sie wahrscheinlich wollen für die Begrenzung der Parallelität der TransformBlock auf einige kleine Konstante. Außerdem könnten Sie begrenzen die Kapazität der TransformBlock und fügen Sie die Einträge asynchron mit SendAsync() zum Beispiel, wenn die Sammlung zu groß ist.

Als zusätzlichen Vorteil, wenn im Vergleich zu deinem code (wenn es funktionierte) ist, dass der Schreibvorgang wird gestartet, sobald ein einzelnes Element fertig ist, und nicht warten, bis die Verarbeitung abgeschlossen ist.

InformationsquelleAutor der Antwort svick

Mit DataFlow als svick vorgeschlagen zuviel, und Stephen ' s Antwort nicht liefern die Mittel zur Steuerung der Parallelität der operation. Das kann jedoch erreicht werden, eher einfach:

public static async Task RunWithMaxDegreeOfConcurrency<T>(
     int maxDegreeOfConcurrency, IEnumerable<T> collection, Func<T, Task> taskFactory)
{
    var activeTasks = new List<Task>(maxDegreeOfConcurrency);
    foreach (var task in collection.Select(taskFactory))
    {
        activeTasks.Add(task);
        if (activeTasks.Count == maxDegreeOfConcurrency)
        {
            await Task.WhenAny(activeTasks.ToArray());
            //observe exceptions here
            activeTasks.RemoveAll(t => t.IsCompleted); 
        }
    }
    await Task.WhenAll(activeTasks.ToArray()).ContinueWith(t => 
    {
        //observe exceptions in a manner consistent with the above   
    });
}

Den ToArray() - Aufrufe optimiert werden kann, mit einem array statt einer Liste und ersetzt abgeschlossene Aufgaben, aber ich bezweifle es würde viel von einem Unterschied in den meisten Szenarien. Beispiel für die Nutzung pro die OP ' s Frage:

RunWithMaxDegreeOfConcurrency(10, ids, async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    var cust = await repo.GetCustomer(i);
    customers.Add(cust);
});

BEARBEITEN Kerl SO Benutzer-und TPL-wiz Eli Arbel wies mich auf eine Verwandte Artikel von Stephen Toub. Wie gewohnt, seine Umsetzung ist sowohl elegant und effizient:

public static Task ForEachAsync<T>(
      this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll( 
        from partition in Partitioner.Create(source).GetPartitions(dop) 
        select Task.Run(async delegate { 
            using (partition) 
                while (partition.MoveNext()) 
                    await body(partition.Current).ContinueWith(t => 
                          {
                              //observe exceptions
                          });

        })); 
}

InformationsquelleAutor der Antwort Ohad Schneider

Wickeln Sie die Parallele.Foreach in eine Aufgabe.Run() und statt dem await-Schlüsselwort verwenden [yourasyncmethod].Ergebnis

(Sie tun müssen, um die Aufgabe.Ausführen Sache nicht blockieren der UI-thread)

Etwas wie dieses:

var yourForeachTask = Task.Run(() =>
        {
            Parallel.ForEach(ids, i =>
            {
                ICustomerRepo repo = new CustomerRepo();
                var cust = repo.GetCustomer(i).Result;
                customers.Add(cust);
            });
        });
await yourForeachTask;

InformationsquelleAutor der Antwort ofcoursedude

Nach Einführung einer Reihe von helper-Methoden, werden Sie in der Lage sein, parallel-Abfragen mit diesem einfachen Syntax:

const int DegreeOfParallelism = 10;
IEnumerable<double> result = await Enumerable.Range(0, 1000000)
    .Split(DegreeOfParallelism)
    .SelectManyAsync(async i => await CalculateAsync(i).ConfigureAwait(false))
    .ConfigureAwait(false);

Was hier passiert, ist, dass wir split-source-Sammlung in 10 Blöcken (.Split(DegreeOfParallelism)), dann laufen Sie 10 Aufgaben, jede Bearbeitung seiner Elemente nacheinander (.SelectManyAsync(...)) und ordnen diese wieder in einer Liste angezeigt.

Erwähnenswert, es ist ein einfacher Ansatz:

double[] result2 = await Enumerable.Range(0, 1000000)
    .Select(async i => await CalculateAsync(i).ConfigureAwait(false))
    .WhenAll()
    .ConfigureAwait(false);

Aber es braucht eine vorsorge: wenn Sie eine source-Sammlung, die zu groß ist, wird es chedule eine Task für jedes Element gleich, das kann zu erheblichen Leistungseinbußen.

Erweiterung Methoden wie in den Beispielen oben wie folgt Aussehen:

public static class CollectionExtensions
{
    ///<summary>
    ///Splits collection into number of collections of nearly equal size.
    ///</summary>
    public static IEnumerable<List<T>> Split<T>(this IEnumerable<T> src, int slicesCount)
    {
        if (slicesCount <= 0) throw new ArgumentOutOfRangeException(nameof(slicesCount));

        List<T> source = src.ToList();
        var sourceIndex = 0;
        for (var targetIndex = 0; targetIndex < slicesCount; targetIndex++)
        {
            var list = new List<T>();
            int itemsLeft = source.Count - targetIndex;
            while (slicesCount * list.Count < itemsLeft)
            {
                list.Add(source[sourceIndex++]);
            }

            yield return list;
        }
    }

    ///<summary>
    ///Takes collection of collections, projects those in parallel and merges results.
    ///</summary>
    public static async Task<IEnumerable<TResult>> SelectManyAsync<T, TResult>(
        this IEnumerable<IEnumerable<T>> source,
        Func<T, Task<TResult>> func)
    {
        List<TResult>[] slices = await source
            .Select(async slice => await slice.SelectListAsync(func).ConfigureAwait(false))
            .WhenAll()
            .ConfigureAwait(false);
        return slices.SelectMany(s => s);
    }

    ///<summary>Runs selector and awaits results.</summary>
    public static async Task<List<TResult>> SelectListAsync<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, Task<TResult>> selector)
    {
        List<TResult> result = new List<TResult>();
        foreach (TSource source1 in source)
        {
            TResult result1 = await selector(source1).ConfigureAwait(false);
            result.Add(result1);
        }
        return result;
    }

    ///<summary>Wraps tasks with Task.WhenAll.</summary>
    public static Task<TResult[]> WhenAll<TResult>(this IEnumerable<Task<TResult>> source)
    {
        return Task.WhenAll<TResult>(source);
    }
}

InformationsquelleAutor der Antwort Vitaliy Ulantikov

Erweiterung Methode was SemaphoreSlim und auch erlaubt, um maximale Grad der Parallelität

    ///<summary>
    ///Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    ///</summary>
    ///<typeparam name="T">Type of IEnumerable</typeparam>
    ///<param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    ///<param name="action">an async <see cref="Action" /> to execute</param>
    ///<param name="maxDegreeOfParallelism">Optional, An integer that represents the maximum degree of parallelism,
    ///Must be grater than 0</param>
    ///<returns>A Task representing an async operation</returns>
    ///<exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxDegreeOfParallelism = null)
    {
        if (maxDegreeOfParallelism.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxDegreeOfParallelism.Value, maxDegreeOfParallelism.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    //Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item);

                        //action is completed, so decrement the number of currently running tasks
                        semaphoreSlim.Release();
                    }));
                }

                //Wait for all tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

Beispiel Für Die Nutzung:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);

InformationsquelleAutor der Antwort Jay Shah