Utiliser les flux asynchrones avec IAsyncEnumerable<T>

A partir de la version 8 du langage C#, il est possible de créer des flux asynchrones, autrement dit des méthodes qui retournent un ensemble de données / d’objets de manière asynchrone, via un itérateur asynchrone qui possède les caractéristiques suivantes :

• Elle est déclarée avec le mot-clé async

• Elle retourne un IAsyncEnumerable<T>

• Elle utilise l'instruction yield return pour produire progressivement les données / objets.

Les flux asynchrones sont particulièrement adaptés lorsqu'une application doit produire ou recevoir des données progressivement, sans attendre que l'ensemble du traitement soit terminé. Ils permettent ainsi de réduire la latence, de diminuer l'encombrement mémoire et d'améliorer la réactivité de l'application.

Voici un exemple :

Console.Write("Nombres : ");

await foreach (int number in GetNumbersAsync())
{
    Console.Write($"{number} ");
}

/// <summary>
/// Produit progressivement 10 nombres aléatoires compris entre 1 et 100.
/// </summary>
static async IAsyncEnumerable<int> GetNumbersAsync()
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(1000);

        yield return Random.Shared.Next(1, 101);
    }
}

La méthode GetNumbersAsync() produit progressivement dix nombres aléatoires compris entre 1 et 100. Avant de produire chaque nombre, elle attend une seconde afin de simuler un traitement asynchrone.

L’instruction yield return rend chaque nombre disponible dès qu’il est généré, sans attendre que l’ensemble des nombres ait été produit.

Dans le code appelant, l’instruction await foreach attend de manière asynchrone chaque nouvel élément du flux. Le thread n’est donc pas bloqué pendant l’attente, et les nombres sont affichés au fur et à mesure de leur disponibilité.

Quelles sont les différences entre Task<List<T>> et IAsyncEnumerable<T> ?

Contrairement à une méthode retournant un Task<List<T>>, qui attend que l'ensemble des données soit disponible avant de retourner une collection complète, une méthode retournant un IAsyncEnumerable<T> produit les éléments progressivement, au fur et à mesure de leur disponibilité. Le code appelant peut ainsi commencer à traiter les premiers éléments immédiatement, sans attendre la fin du traitement. Cette approche limite également la quantité de données conservées simultanément en mémoire et s'avère particulièrement adaptée aux traitements de grands volumes de données ou de flux continus.

Interruption du flux

La gestion de l’annulation est particulièrement importante avec les flux longs ou continus, car elle permet au consommateur d’interrompre proprement la production des données / objets.

using System.Runtime.CompilerServices;

using var cancellationTokenSource = new CancellationTokenSource();

// Annule automatiquement le traitement après cinq secondes.
cancellationTokenSource.CancelAfter(TimeSpan.FromSeconds(5));

Console.Write("Nombres : ");

try
{
    await foreach (int number in GetNumbersAsync(
        cancellationTokenSource.Token))
    {
        Console.Write($"{number} ");
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine();
    Console.WriteLine("La production des nombres a été annulée.");
}

/// <summary>
/// Produit progressivement dix nombres aléatoires compris entre 1 et 100.
/// </summary>
static async IAsyncEnumerable<int> GetNumbersAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(1000, cancellationToken);

        yield return Random.Shared.Next(1, 101);
    }
}

Dans cet exemple, l’attribut [EnumeratorCancellation] indique que le paramètre de type CancellationToken fourni par le consommateur doit être utilisé pour annuler l’énumération asynchrone. L’utilisation de la méthode CancelAfter déclenche automatiquement une demande d’annulation après cinq secondes. Comme le paramètre de type CancellationToken est transmis à la méthode Task.Delay, l’attente en cours est interrompue par la levée d'une exception de type OperationCanceledException, qui est ensuite catchée par le code appelant.