Verwenden von async / await für mehrere Aufgaben

406

Ich verwende einen API-Client, der vollständig asynchron ist, dh jede Operation gibt entweder zurück Taskoder Task<T>z.

static async Task DoSomething(int siteId, int postId, IBlogClient client)
{
    await client.DeletePost(siteId, postId); // call API client
    Console.WriteLine("Deleted post {0}.", siteId);
}

Was ist mit den asynchronen / wartenden C # 5-Operatoren die richtige / effizienteste Methode, um mehrere Aufgaben zu starten und darauf zu warten, dass sie alle abgeschlossen sind:

int[] ids = new[] { 1, 2, 3, 4, 5 };
Parallel.ForEach(ids, i => DoSomething(1, i, blogClient).Wait());

oder:

int[] ids = new[] { 1, 2, 3, 4, 5 };
Task.WaitAll(ids.Select(i => DoSomething(1, i, blogClient)).ToArray());

Da der API-Client HttpClient intern verwendet, würde ich erwarten, dass dies sofort 5 HTTP-Anforderungen ausgibt, die nach Abschluss jeweils in die Konsole geschrieben werden.

Ben Foster
quelle
Und was ist das Problem?
Serg Shevchenko
1
@SergShevchenko Das Problem ist, dass sein Parallel.ForEach falsch ausgeführt wird (siehe Antworten) - er fragt, ob seine Versuche, asynchronen Code parallel auszuführen, korrekt sind und zwei Lösungsversuche anbieten, und ob einer besser ist als der andere (und vermutlich warum) ).
AnorZaken

Antworten:

572
int[] ids = new[] { 1, 2, 3, 4, 5 };
Parallel.ForEach(ids, i => DoSomething(1, i, blogClient).Wait());

Obwohl Sie die Operationen parallel zum obigen Code ausführen, blockiert dieser Code jeden Thread, auf dem jede Operation ausgeführt wird. Wenn der Netzwerkanruf beispielsweise 2 Sekunden dauert, bleibt jeder Thread 2 Sekunden lang hängen, ohne etwas anderes zu tun als zu warten.

int[] ids = new[] { 1, 2, 3, 4, 5 };
Task.WaitAll(ids.Select(i => DoSomething(1, i, blogClient)).ToArray());

Andererseits WaitAllblockiert der obige Code mit auch die Threads und Ihre Threads können keine anderen Arbeiten verarbeiten, bis der Vorgang endet.

Empfohlener Ansatz

Ich würde es vorziehen, wenn WhenAllIhre Operationen asynchron parallel ausgeführt werden.

public async Task DoWork() {

    int[] ids = new[] { 1, 2, 3, 4, 5 };
    await Task.WhenAll(ids.Select(i => DoSomething(1, i, blogClient)));
}

In dem oben genannten Fall müssen Sie nicht einmal awaitdirekt von der Methode zurückkehren, da Sie keine Fortsetzungen haben:

public Task DoWork() 
{
    int[] ids = new[] { 1, 2, 3, 4, 5 };
    return Task.WhenAll(ids.Select(i => DoSomething(1, i, blogClient)));
}

Um dies zu belegen, finden Sie hier einen detaillierten Blog-Beitrag, in dem alle Alternativen und ihre Vor- und Nachteile erläutert werden: Wie und wo gleichzeitige asynchrone E / A mit der ASP.NET-Web-API

Tugberk
quelle
31
"Der obige Code WaitAllblockiert auch die Threads" - blockiert er nicht nur einen Thread, den aufgerufenen WaitAll?
Rawling
5
In der Dokumentation heißt es: "Typ: System.Threading.Tasks.Task [] Ein Array von Task-Instanzen, auf die gewartet werden soll." Es blockiert also alle Threads.
Mixxiphoid
30
@Mixxiphoid: Das von Ihnen angegebene Bit bedeutet nicht, dass es alle Threads blockiert. Es blockiert nur den aufrufenden Thread, während die angegebenen Aufgaben ausgeführt werden. Wie diese Aufgaben tatsächlich ausgeführt werden, hängt vom Scheduler ab. Normalerweise wird der Thread, auf dem er ausgeführt wurde, nach Abschluss jeder Aufgabe in den Pool zurückgeführt. Jeder Thread würde nicht blockiert bleiben, bis andere abgeschlossen sind.
Musaul
3
@tugberk, So wie ich es verstehe, besteht der einzige Unterschied zwischen den "klassischen" Task-Methoden und den Async-Gegenstücken darin, wie sie mit Threads interagieren, wenn eine Task gestartet und beendet wird. Die klassische Methode unter einem Standardplaner blockiert einen Thread während dieses Zeitraums (auch wenn er "schläft"), während die asynchronen dies nicht tun. Außerhalb dieses Zeitraums gibt es keinen Unterschied, dh die Aufgabe ist geplant, aber nicht gestartet, und wenn sie abgeschlossen ist, wartet der Anrufer noch.
Musaul
3
@tugberk Siehe stackoverflow.com/a/6123432/750216 Der Unterschied besteht darin, ob der aufrufende Thread blockiert ist oder nicht. Der Rest ist der gleiche. Möglicherweise möchten Sie die Antwort zur Verdeutlichung bearbeiten.
Răzvan Flavius ​​Panda
45

Ich war neugierig auf die Ergebnisse der in der Frage angegebenen Methoden sowie auf die akzeptierte Antwort und habe sie auf die Probe gestellt.

Hier ist der Code:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace AsyncTest
{
    class Program
    {
        class Worker
        {
            public int Id;
            public int SleepTimeout;

            public async Task DoWork(DateTime testStart)
            {
                var workerStart = DateTime.Now;
                Console.WriteLine("Worker {0} started on thread {1}, beginning {2} seconds after test start.",
                    Id, Thread.CurrentThread.ManagedThreadId, (workerStart-testStart).TotalSeconds.ToString("F2"));
                await Task.Run(() => Thread.Sleep(SleepTimeout));
                var workerEnd = DateTime.Now;
                Console.WriteLine("Worker {0} stopped; the worker took {1} seconds, and it finished {2} seconds after the test start.",
                   Id, (workerEnd-workerStart).TotalSeconds.ToString("F2"), (workerEnd-testStart).TotalSeconds.ToString("F2"));
            }
        }

        static void Main(string[] args)
        {
            var workers = new List<Worker>
            {
                new Worker { Id = 1, SleepTimeout = 1000 },
                new Worker { Id = 2, SleepTimeout = 2000 },
                new Worker { Id = 3, SleepTimeout = 3000 },
                new Worker { Id = 4, SleepTimeout = 4000 },
                new Worker { Id = 5, SleepTimeout = 5000 },
            };

            var startTime = DateTime.Now;
            Console.WriteLine("Starting test: Parallel.ForEach...");
            PerformTest_ParallelForEach(workers, startTime);
            var endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            startTime = DateTime.Now;
            Console.WriteLine("Starting test: Task.WaitAll...");
            PerformTest_TaskWaitAll(workers, startTime);
            endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            startTime = DateTime.Now;
            Console.WriteLine("Starting test: Task.WhenAll...");
            var task = PerformTest_TaskWhenAll(workers, startTime);
            task.Wait();
            endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            Console.ReadKey();
        }

        static void PerformTest_ParallelForEach(List<Worker> workers, DateTime testStart)
        {
            Parallel.ForEach(workers, worker => worker.DoWork(testStart).Wait());
        }

        static void PerformTest_TaskWaitAll(List<Worker> workers, DateTime testStart)
        {
            Task.WaitAll(workers.Select(worker => worker.DoWork(testStart)).ToArray());
        }

        static Task PerformTest_TaskWhenAll(List<Worker> workers, DateTime testStart)
        {
            return Task.WhenAll(workers.Select(worker => worker.DoWork(testStart)));
        }
    }
}

Und die daraus resultierende Ausgabe:

Starting test: Parallel.ForEach...
Worker 1 started on thread 1, beginning 0.21 seconds after test start.
Worker 4 started on thread 5, beginning 0.21 seconds after test start.
Worker 2 started on thread 3, beginning 0.21 seconds after test start.
Worker 5 started on thread 6, beginning 0.21 seconds after test start.
Worker 3 started on thread 4, beginning 0.21 seconds after test start.
Worker 1 stopped; the worker took 1.90 seconds, and it finished 2.11 seconds after the test start.
Worker 2 stopped; the worker took 3.89 seconds, and it finished 4.10 seconds after the test start.
Worker 3 stopped; the worker took 5.89 seconds, and it finished 6.10 seconds after the test start.
Worker 4 stopped; the worker took 5.90 seconds, and it finished 6.11 seconds after the test start.
Worker 5 stopped; the worker took 8.89 seconds, and it finished 9.10 seconds after the test start.
Test finished after 9.10 seconds.

Starting test: Task.WaitAll...
Worker 1 started on thread 1, beginning 0.01 seconds after test start.
Worker 2 started on thread 1, beginning 0.01 seconds after test start.
Worker 3 started on thread 1, beginning 0.01 seconds after test start.
Worker 4 started on thread 1, beginning 0.01 seconds after test start.
Worker 5 started on thread 1, beginning 0.01 seconds after test start.
Worker 1 stopped; the worker took 1.00 seconds, and it finished 1.01 seconds after the test start.
Worker 2 stopped; the worker took 2.00 seconds, and it finished 2.01 seconds after the test start.
Worker 3 stopped; the worker took 3.00 seconds, and it finished 3.01 seconds after the test start.
Worker 4 stopped; the worker took 4.00 seconds, and it finished 4.01 seconds after the test start.
Worker 5 stopped; the worker took 5.00 seconds, and it finished 5.01 seconds after the test start.
Test finished after 5.01 seconds.

Starting test: Task.WhenAll...
Worker 1 started on thread 1, beginning 0.00 seconds after test start.
Worker 2 started on thread 1, beginning 0.00 seconds after test start.
Worker 3 started on thread 1, beginning 0.00 seconds after test start.
Worker 4 started on thread 1, beginning 0.00 seconds after test start.
Worker 5 started on thread 1, beginning 0.00 seconds after test start.
Worker 1 stopped; the worker took 1.00 seconds, and it finished 1.00 seconds after the test start.
Worker 2 stopped; the worker took 2.00 seconds, and it finished 2.00 seconds after the test start.
Worker 3 stopped; the worker took 3.00 seconds, and it finished 3.00 seconds after the test start.
Worker 4 stopped; the worker took 4.00 seconds, and it finished 4.00 seconds after the test start.
Worker 5 stopped; the worker took 5.00 seconds, and it finished 5.00 seconds after the test start.
Test finished after 5.00 seconds.
RiaanDP
quelle
2
Wenn Sie die Zeit auf jedes dieser Ergebnisse setzen, wäre dies nützlicher
Serj Sagan
8
@SerjSagan Meine ursprüngliche Idee war nur zu überprüfen, ob die Worker jeweils gleichzeitig gestartet werden, aber ich habe Zeitstempel hinzugefügt, um die Klarheit des Tests zu verbessern. Danke für den Vorschlag.
RiaanDP
Danke für den Test. Es fühlt sich jedoch etwas seltsam an, dass Sie thread.sleep auf einem Thread ausführen, der vom "Worker-Thread" getrennt ist. Nicht, dass es in diesem Fall wichtig wäre, aber wäre es für Task nicht sinnvoller, die Worker-Threads auszuführen, wenn wir Rechenarbeit simulieren, oder nur Task.Delay anstelle von Sleep, wenn wir I / O simulieren? Überprüfen Sie einfach, was Sie darüber denken würden.
AnorZaken
24

Da die von Ihnen aufgerufene API asynchron ist, macht die Parallel.ForEachVersion wenig Sinn. Sie sollten .Waitin der WaitAllVersion nicht verwenden, da dies die Parallelität verlieren würde. Eine andere Alternative, wenn der Aufrufer asynchron ist, ist die Verwendung Task.WhenAllnach dem Ausführen Selectund ToArrayzum Generieren des Array von Aufgaben. Eine zweite Alternative ist die Verwendung von Rx 2.0

James Manning
quelle
10

Sie können eine Task.WhenAllFunktion verwenden, mit der Sie n Aufgaben übergeben können. Task.WhenAllgibt eine Aufgabe zurück, die Task.WhenAllvollständig ausgeführt wird, wenn alle Aufgaben, die Sie übergeben haben, abgeschlossen sind. Sie müssen asynchron warten, Task.WhenAlldamit Sie Ihren UI-Thread nicht blockieren:

   public async Task DoSomeThing() {

       var Task[] tasks = new Task[numTasks];
       for(int i = 0; i < numTask; i++)
       {
          tasks[i] = CallSomeAsync();
       }
       await Task.WhenAll(tasks);
       // code that'll execute on UI thread
   }
Ahmed Wasim
quelle
8

Parallel.ForEacherfordert eine Liste benutzerdefinierter Worker und eine nicht asynchrone Action Ausführung für jeden Worker.

Task.WaitAllund Task.WhenAllerfordern eine List<Task>, die per Definition asynchron sind.

Ich fand die Antwort von RiaanDP sehr nützlich, um den Unterschied zu verstehen, aber es muss korrigiert werden . Nicht genug Ruf, um auf seinen Kommentar zu antworten, daher meine eigene Antwort.Parallel.ForEach

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace AsyncTest
{
    class Program
    {
        class Worker
        {
            public int Id;
            public int SleepTimeout;

            public void DoWork(DateTime testStart)
            {
                var workerStart = DateTime.Now;
                Console.WriteLine("Worker {0} started on thread {1}, beginning {2} seconds after test start.",
                    Id, Thread.CurrentThread.ManagedThreadId, (workerStart - testStart).TotalSeconds.ToString("F2"));
                Thread.Sleep(SleepTimeout);
                var workerEnd = DateTime.Now;
                Console.WriteLine("Worker {0} stopped; the worker took {1} seconds, and it finished {2} seconds after the test start.",
                   Id, (workerEnd - workerStart).TotalSeconds.ToString("F2"), (workerEnd - testStart).TotalSeconds.ToString("F2"));
            }

            public async Task DoWorkAsync(DateTime testStart)
            {
                var workerStart = DateTime.Now;
                Console.WriteLine("Worker {0} started on thread {1}, beginning {2} seconds after test start.",
                    Id, Thread.CurrentThread.ManagedThreadId, (workerStart - testStart).TotalSeconds.ToString("F2"));
                await Task.Run(() => Thread.Sleep(SleepTimeout));
                var workerEnd = DateTime.Now;
                Console.WriteLine("Worker {0} stopped; the worker took {1} seconds, and it finished {2} seconds after the test start.",
                   Id, (workerEnd - workerStart).TotalSeconds.ToString("F2"), (workerEnd - testStart).TotalSeconds.ToString("F2"));
            }
        }

        static void Main(string[] args)
        {
            var workers = new List<Worker>
            {
                new Worker { Id = 1, SleepTimeout = 1000 },
                new Worker { Id = 2, SleepTimeout = 2000 },
                new Worker { Id = 3, SleepTimeout = 3000 },
                new Worker { Id = 4, SleepTimeout = 4000 },
                new Worker { Id = 5, SleepTimeout = 5000 },
            };

            var startTime = DateTime.Now;
            Console.WriteLine("Starting test: Parallel.ForEach...");
            PerformTest_ParallelForEach(workers, startTime);
            var endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            startTime = DateTime.Now;
            Console.WriteLine("Starting test: Task.WaitAll...");
            PerformTest_TaskWaitAll(workers, startTime);
            endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            startTime = DateTime.Now;
            Console.WriteLine("Starting test: Task.WhenAll...");
            var task = PerformTest_TaskWhenAll(workers, startTime);
            task.Wait();
            endTime = DateTime.Now;
            Console.WriteLine("Test finished after {0} seconds.\n",
                (endTime - startTime).TotalSeconds.ToString("F2"));

            Console.ReadKey();
        }

        static void PerformTest_ParallelForEach(List<Worker> workers, DateTime testStart)
        {
            Parallel.ForEach(workers, worker => worker.DoWork(testStart));
        }

        static void PerformTest_TaskWaitAll(List<Worker> workers, DateTime testStart)
        {
            Task.WaitAll(workers.Select(worker => worker.DoWorkAsync(testStart)).ToArray());
        }

        static Task PerformTest_TaskWhenAll(List<Worker> workers, DateTime testStart)
        {
            return Task.WhenAll(workers.Select(worker => worker.DoWorkAsync(testStart)));
        }
    }
}

Die resultierende Ausgabe ist unten. Ausführungszeiten sind vergleichbar. Ich habe diesen Test ausgeführt, während mein Computer den wöchentlichen Antivirenscan durchführte. Durch Ändern der Reihenfolge der Tests wurden die Ausführungszeiten geändert.

Starting test: Parallel.ForEach...
Worker 1 started on thread 9, beginning 0.02 seconds after test start.
Worker 2 started on thread 10, beginning 0.02 seconds after test start.
Worker 3 started on thread 11, beginning 0.02 seconds after test start.
Worker 4 started on thread 13, beginning 0.03 seconds after test start.
Worker 5 started on thread 14, beginning 0.03 seconds after test start.
Worker 1 stopped; the worker took 1.00 seconds, and it finished 1.02 seconds after the test start.
Worker 2 stopped; the worker took 2.00 seconds, and it finished 2.02 seconds after the test start.
Worker 3 stopped; the worker took 3.00 seconds, and it finished 3.03 seconds after the test start.
Worker 4 stopped; the worker took 4.00 seconds, and it finished 4.03 seconds after the test start.
Worker 5 stopped; the worker took 5.00 seconds, and it finished 5.03 seconds after the test start.
Test finished after 5.03 seconds.

Starting test: Task.WaitAll...
Worker 1 started on thread 9, beginning 0.00 seconds after test start.
Worker 2 started on thread 9, beginning 0.00 seconds after test start.
Worker 3 started on thread 9, beginning 0.00 seconds after test start.
Worker 4 started on thread 9, beginning 0.00 seconds after test start.
Worker 5 started on thread 9, beginning 0.01 seconds after test start.
Worker 1 stopped; the worker took 1.00 seconds, and it finished 1.01 seconds after the test start.
Worker 2 stopped; the worker took 2.00 seconds, and it finished 2.01 seconds after the test start.
Worker 3 stopped; the worker took 3.00 seconds, and it finished 3.01 seconds after the test start.
Worker 4 stopped; the worker took 4.00 seconds, and it finished 4.01 seconds after the test start.
Worker 5 stopped; the worker took 5.00 seconds, and it finished 5.01 seconds after the test start.
Test finished after 5.01 seconds.

Starting test: Task.WhenAll...
Worker 1 started on thread 9, beginning 0.00 seconds after test start.
Worker 2 started on thread 9, beginning 0.00 seconds after test start.
Worker 3 started on thread 9, beginning 0.00 seconds after test start.
Worker 4 started on thread 9, beginning 0.00 seconds after test start.
Worker 5 started on thread 9, beginning 0.00 seconds after test start.
Worker 1 stopped; the worker took 1.00 seconds, and it finished 1.00 seconds after the test start.
Worker 2 stopped; the worker took 2.00 seconds, and it finished 2.00 seconds after the test start.
Worker 3 stopped; the worker took 3.00 seconds, and it finished 3.00 seconds after the test start.
Worker 4 stopped; the worker took 4.00 seconds, and it finished 4.00 seconds after the test start.
Worker 5 stopped; the worker took 5.00 seconds, and it finished 5.01 seconds after the test start.
Test finished after 5.01 seconds.
JPortillo
quelle