Flujos asíncronos en C# (IAsyncEnumerable)

En las aplicaciones modernas de .NET, los datos suelen llegar por partes y a lo largo del tiempo: flujos de red, líneas de registro, datos de sensores… IAsyncEnumerable<T> permite consumir estos flujos de forma asíncrona y perezosa (lazy). En el lado del productor se usa async + yield return, y en el lado del consumidor await foreach.

¿Qué es IAsyncEnumerable?

IAsyncEnumerable<T> permite producir y consumir una secuencia de datos asíncronamente. A diferencia de IEnumerable<T>, cada paso puede implicar una espera (por ejemplo, una operación I/O). Para consumir, se usa await foreach; en cada iteración se espera hasta que el siguiente elemento esté disponible.

// Consumo simple
await foreach (var item in GetNumbersAsync())
{
    Console.WriteLine(item);
}

Iterador asíncrono: async + yield return

Al escribir un método productor asíncrono, el tipo de retorno es IAsyncEnumerable<T>. Dentro del cuerpo, se puede usar await para esperar y yield return para generar elementos.

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

static async IAsyncEnumerable<int> GetNumbersAsync()
{
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(500); // Simula una operación I/O o un trabajo prolongado
        yield return i;        // un nuevo número cada 500 ms
    }
}

// Consumo
await foreach (var n in GetNumbersAsync())
{
    Console.WriteLine($"Recibido: {n}");
}

Consumo con await foreach

await foreach obtiene elementos de un flujo asíncrono sin bloquear la interfaz de usuario ni el hilo. En cada paso espera a que los datos estén listos, y una vez disponibles, el ciclo continúa.

await foreach (var registro in LeerLogsAsync())
{
    // Procesar el registro tan pronto como llegue
    Procesar(registro);
}

Manejo de errores y using

Al consumir flujos asíncronos se puede usar try/catch normalmente. Si ocurre un error dentro del flujo o durante el consumo, este se lanzará durante la ejecución de await foreach.

try
{
    await foreach (var item in GetNumbersAsync())
        Console.WriteLine(item);
}
catch (Exception ex)
{
    Console.WriteLine("Error: " + ex.Message);
}

Para recursos asíncronos, use IAsyncDisposable y await using.

await using var recurso = await AbrirRecursoAsync();
await foreach (var x in recurso.StreamAsync())
{
    // ...
}

Soporte de cancelación (CancellationToken)

Para detener flujos prolongados, se puede cancelar el consumo. En el lado del productor, capture el token con [EnumeratorCancellation] y revíselo periódicamente.

using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

static async IAsyncEnumerable<int> ContarAsync(
    int inicio, int cantidad, [EnumeratorCancellation] CancellationToken ct = default)
{
    for (int i = 0; i < cantidad; i++)
    {
        ct.ThrowIfCancellationRequested();
        await Task.Delay(300, ct);
        yield return inicio + i;
    }
}

// Consumo
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1)); // cancelar después de 1 s
await foreach (var n in ContarAsync(10, 100, cts.Token))
{
    Console.WriteLine(n);
}

Comparación: IEnumerable<Task<T>> vs IAsyncEnumerable<T>

• IEnumerable<Task<T>>: Se generan todas las tareas primero y luego se consumen; el control de flujo (backpressure) es débil.
• IAsyncEnumerable<T>: Los elementos se generan a demanda; el uso de memoria y el tiempo de procesamiento son más eficientes.

Lectura asíncrona línea por línea (Ejemplo con wrapper)

El siguiente ejemplo genera un flujo que lee un archivo asíncronamente, línea por línea. Cada línea se entrega al consumidor mediante yield return tan pronto como está disponible.

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

static async IAsyncEnumerable<string> LeerLineasAsync(string ruta)
{
    using var fs = new FileStream(ruta, FileMode.Open, FileAccess.Read, FileShare.Read,
                                  bufferSize: 4096, useAsync: true);
    using var sr = new StreamReader(fs);

    while (!sr.EndOfStream)
    {
        var linea = await sr.ReadLineAsync(); // cada vez que se lee una línea
        if (linea is not null)
            yield return linea;
    }
}

// Consumo
await foreach (var linea in LeerLineasAsync("log.txt"))
{
    Console.WriteLine(linea);
}

Filtrado simple sobre un flujo (Pipeline)

Los flujos asíncronos se pueden encadenar para crear un pipeline: producir → filtrar → transformar → consumir.

static async IAsyncEnumerable<int> FiltrarAsync(IAsyncEnumerable<int> fuente)
{
    await foreach (var n in fuente)
    {
        if (n % 2 == 0)          // pasar solo números pares
            yield return n * 10; // transformar
    }
}

// Uso
await foreach (var x in FiltrarAsync(GetNumbersAsync()))
{
    Console.WriteLine(x); // 20, 40, ...
}

Escenario de UI: await foreach en WPF

En WPF, los flujos prolongados no bloquean la interfaz; esta puede actualizarse cada vez que llega un nuevo elemento. Si se necesita actualizar la UI, vuelva al hilo de la interfaz usando Dispatcher.

// Ejemplo: leer datos del sensor desde un flujo y mostrarlos en la UI
await foreach (var v in LeerSensorAsync())
{
    Dispatcher.Invoke(() => txtEstado.Text = $"Último valor: {v}");
}

Consejos y consideraciones

• Genere datos de forma incremental con yield return; evite cargar grandes listas completas en memoria.
• Proporcione soporte de cancelación (CancellationToken); los flujos largos deben poder detenerse.
• Maneje los errores adecuadamente con try/catch; capture y registre en el consumidor cuando sea necesario.
• Use Dispatcher para actualizaciones de UI; no acceda a controles desde hilos que no sean de la interfaz.

TL;DR

Artículos relacionados