Usando async / await para múltiples tareas

406

Estoy usando un cliente API que es completamente asíncrono, es decir, cada operación devuelve Tasko Task<T>, por ejemplo:

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

Usando los operadores asíncronos / en espera de C # 5, ¿cuál es la forma correcta / más eficiente de iniciar múltiples tareas y esperar a que todas se completen?

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

o:

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

Dado que el cliente API está utilizando HttpClient internamente, esperaría que esto emita 5 solicitudes HTTP de inmediato, escribiendo en la consola a medida que se completa cada una.

Ben Foster
fuente
Y cuál es el problema ?
Serg Shevchenko
1
@SergShevchenko El problema es que su Parallel.ForEach se hace incorrectamente (ver respuestas): está preguntando si sus intentos de ejecutar código asíncrono en paralelo son correctos, ofreciendo dos intentos de solución y si uno es mejor que el otro (y presumiblemente por qué) )
AnorZaken

Respuestas:

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

Aunque ejecuta las operaciones en paralelo con el código anterior, este código bloquea cada subproceso en el que se ejecuta cada operación. Por ejemplo, si la llamada de red tarda 2 segundos, cada subproceso se cuelga durante 2 segundos sin hacer nada más que esperar.

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

Por otro lado, el código anterior WaitAlltambién bloquea los hilos y sus hilos no serán libres de procesar ningún otro trabajo hasta que finalice la operación.

Enfoque recomendado

Prefiero WhenAllque realice sus operaciones de forma asincrónica en paralelo.

public async Task DoWork() {

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

De hecho, en el caso anterior, ni siquiera necesita hacerlo await, solo puede regresar directamente del método ya que no tiene ninguna continuación:

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

Para respaldar esto, aquí hay una publicación de blog detallada que detalla todas las alternativas y sus ventajas / desventajas: Cómo y dónde E / S asincrónicas concurrentes con API web ASP.NET

tugberk
fuente
31
"el código anterior WaitAlltambién bloquea los hilos" - ¿no bloquea solo un hilo, el que llamó WaitAll?
Rawling
55
@Rawling la documentación indica que "Tipo: System.Threading.Tasks.Task [] Una matriz de instancias de tareas en las que esperar". Entonces, bloquea todos los hilos.
Mixxiphoid
30
@Mixxiphoid: el bit que citó no significa que bloquee todos los hilos. Solo bloquea el hilo de llamada mientras se ejecutan las tareas proporcionadas. La forma en que se ejecutan esas tareas depende del planificador. Normalmente, después de que se completa cada tarea, el hilo en el que se estaba ejecutando volvería al grupo. Cada hilo no permanecerá bloqueado hasta que otros estén completos.
musaul 01 de
3
@tugberk, según tengo entendido, la única diferencia entre los métodos de tareas "clásicos" y los equivalentes de Async es cómo interactúan con los hilos entre cuando una tarea comienza a ejecutarse y termina de ejecutarse. El método clásico en un planificador predeterminado acaparará un hilo durante ese período (incluso si está "inactivo"), mientras que los asíncronos no lo harán. No hay diferencia fuera de ese período, es decir, la tarea es programada pero no iniciada, y cuando se ha completado pero la persona que llama aún está esperando.
musaul
3
@tugberk Consulte stackoverflow.com/a/6123432/750216 la diferencia está en si el hilo de llamada está bloqueado o no, el resto es el mismo. Es posible que desee editar la respuesta para aclarar.
Răzvan Flavius ​​Panda
45

Tenía curiosidad por ver los resultados de los métodos proporcionados en la pregunta, así como la respuesta aceptada, así que lo puse a prueba.

Aquí está el código:

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)));
        }
    }
}

Y el resultado resultante:

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
fuente
2
Si pones el tiempo en cada uno de estos resultados, esto sería más útil
Serj Sagan el
8
@SerjSagan mi idea inicial era solo verificar que los trabajadores se inicien simultáneamente en cada caso, pero agregué marcas de tiempo para mejorar la claridad de la prueba. Gracias por la sugerencia.
RiaanDP
Gracias por la prueba Sin embargo, se siente un poco extraño que esté ejecutando thread.sleep en un subproceso separado del "subproceso de trabajo". No es que importe en este caso, pero no tendría más sentido para Task. Ejecute los hilos de trabajo si estamos simulando trabajo computacional, o simplemente Task. ¿Retraso en lugar de dormir si estamos simulando E / S? Solo comprobando lo que piensas sobre eso.
AnorZaken
24

Como la API a la que llama es asíncrona, la Parallel.ForEachversión no tiene mucho sentido. No debe usar .Waiten la WaitAllversión ya que eso perdería el paralelismo. Otra alternativa si la persona que llama es asíncrona está usando Task.WhenAlldespués de hacer Selecty ToArraygenerar la matriz de tareas. Una segunda alternativa es usar Rx 2.0

James Manning
fuente
10

Puede usar la Task.WhenAllfunción que puede pasar n tareas; Task.WhenAlldevolverá una tarea que se ejecutará hasta completar cuando se completen todas las tareas que pasó Task.WhenAll. Tienes que esperar asincrónicamente Task.WhenAllpara que no bloquees tu hilo de la interfaz de usuario:

   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
fuente
8

Parallel.ForEachrequiere una lista de trabajadores definidos por el usuario y una no asíncrona Action para realizar con cada trabajador.

Task.WaitAlly Task.WhenAllrequieren un List<Task>, que son por definición asíncronos.

La respuesta de RiaanDP me pareció muy útil para entender la diferencia, pero necesita una corrección . No hay suficiente reputación para responder a su comentario, de ahí mi propia respuesta.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)));
        }
    }
}

La salida resultante está debajo. Los tiempos de ejecución son comparables. Ejecuté esta prueba mientras mi computadora realizaba el análisis antivirus semanal. Cambiar el orden de las pruebas cambió los tiempos de ejecución en ellas.

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
fuente