Channel: producteur–consommateur ( Channel)

1. Introduction

Aujourd'hui on passe un cap ! Il est temps de découvrir un outil spécial — le Channel. Cet outil a été créé pour les applications asynchrones modernes en .NET : là où les verrous classiques soit n'aident pas, soit ralentissent énormément.

Vous allez voir le pattern "Producteur–Consommateur" (Producer-Consumer), populaire depuis les années 60. Vous construirez une simple "ligne de traitement" asynchrone où certaines tâches produisent des éléments (par ex. téléchargent des fichiers, calculent des valeurs, attendent des événements) et d'autres les consomment (par ex. sauvegardent, écrivent en base, mettent à jour l'UI).

Pourquoi est né le Channel ?

1. Le pattern "Producteur–Consommateur" était depuis longtemps résolu avec des queues : le producteur met des tâches dans une queue, le consommateur les prend. Mais ! BlockingCollection<T>, les queues basées sur ConcurrentQueue<T>, ou la synchronisation manuelle avec lock — tout cela n'est pas asynchrone. Autrement dit, un thread peut seulement se bloquer en attendant des données, au lieu de rendre la main au scheduler async/await.
2. L'asynchronisme en .NET n'est pas un mot à la mode du vendredi, c'est la base des architectures modernes. Bloquer des threads pour attendre des éléments est coûteux et inefficace. Il faut pouvoir attendre l'arrivée des données sans bloquer — c'est ce que résout le Channel.
3. Flexibilité : avec les channels on peut monter des pipelines complexes, séparer la logique des tâches, ajouter des étapes intermédiaires et de la répartition de charge — le tout sans souffrir de la synchronisation bas niveau.

Qu'est-ce qu'un Channel ? (Analogie et architecture)

Imaginez que vous avez un témoin de relais (ou un tapis roulant) par lequel on peut transmettre des objets d'un endroit à un autre, sans que les personnes aient besoin de se rencontrer en personne. L'essentiel est que le témoin ne se perde pas en chemin.

Channel est un mécanisme intégré à .NET pour la transmission asynchrone de données entre différentes tasks, threads ou parties d'une application. Il implémente une queue asynchrone avec support de "wait" aussi bien pour l'insertion que pour la lecture d'éléments.

• Le producteur met des éléments dans le channel (par ex. des requêtes à traiter) ;
• Le consommateur récupère les éléments — et le tour est joué !

2. La classe Channel<T> et son fonctionnement

Tout commence par l'espace de noms :

using System.Threading.Channels;

Contrairement aux collections habituelles, Channel est une usine qui crée des objets spécialisés pour la transmission de données.

Types principaux :

• ChannelWriter<T> — "writer" (producteur). Insère uniquement des éléments.
• ChannelReader<T> — "reader" (consommateur). Extrait uniquement des éléments.
• Le channel (Channel) sépare les responsabilités : le writer ne sait rien du reader et vice versa.

En .NET il existe plusieurs implémentations de channel, chacune avec ses caractéristiques : unbounded (pas de limite de taille), bounded (limité en nombre d'éléments), single-producer-single-consumer (SPSC), multi-producer-multi-consumer (MPMC), etc. On commence par la variante la plus universelle.

Exemple simple : une queue asynchrone de tâches

using System;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        // On crée un channel sans limite de taille
        var channel = Channel.CreateUnbounded<int>();

        // Tâche producteur
        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 10; i++)
            {
                Console.WriteLine($"Producteur : Dépose {i} dans le channel");
                await channel.Writer.WriteAsync(i); // Écriture asynchrone !
                await Task.Delay(100); // Simulation de travail
            }
            channel.Writer.Complete(); // On indique qu'on ne va plus écrire
        });

        // Tâche consommateur
        var consumer = Task.Run(async () =>
        {
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"Consommateur : A reçu {item} du channel");
                await Task.Delay(200); // Simulation du traitement
            }
            Console.WriteLine("Consommateur : Le channel est fermé");
        });

        await Task.WhenAll(producer, consumer);
    }
}

Que se passe-t-il ici ?

• Channel.CreateUnbounded<int>() — on crée un channel sans limite de taille.
• Le producteur écrit les nombres de 0 à 9 dans le channel via WriteAsync.
• Après la fin des écritures on appelle Complete() — signal "Il n'y aura plus d'éléments !".
• Le consommateur itère tous les éléments avec ReadAllAsync() (également asynchrone !), jusqu'à la fermeture du channel.
• Les délais (Task.Delay) simulent un travail réel : on voit que les nombres peuvent être écrits plus vite qu'ils ne sont lus.

3. Pourquoi tout cela fonctionne-t-il de manière asynchrone ?

Les queues bloquantes classiques (par ex. BlockingCollection ou celles protégées par lock) ne peuvent que bloquer un thread. Ce qui signifie qu'on gaspille des ressources précieuses si on a beaucoup de tâches ou qu'on vise une haute performance.

Avec les channels :

• Si le producteur est plus rapide, le channel accumule les éléments (limité seulement par la mémoire ou par la capacité si elle est définie).
• Si le consommateur est plus rapide, il attendra qu'un élément apparaisse (sans bloquer le thread, il libère le scheduler).

C'est parfait pour les scénarios où vous ne savez pas à l'avance qui sera le plus rapide — producteurs ou consommateurs.

Usage concret

• Logging asynchrone : écrire des messages dans un fichier ou une base depuis un thread séparé ;
• Traitement de requêtes web : une tâche télécharge un lot de pages, une autre les analyse ;
• Scan et indexation de dossiers : certaines tâches parcourent le système de fichiers, d'autres compilent des statistiques ;
• Pipelines de traitement de données complexes : par ex. dans des tâches ETL (Extract–Transform–Load) une étape transforme la matière première en semi-produits, une autre en produit final.

4. Channel limité (Bounded Channel)

Les channels "illimités" sont amusants, mais la mémoire n'est pas infinie (même si votre machine vous semble énorme).

Un channel limité (bounded) permet de fixer un nombre maximal d'éléments pouvant se trouver simultanément à l'intérieur. Si le channel est plein — le producteur attend que le consommateur retire quelque chose.

Exemple :

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(3)
{
    FullMode = BoundedChannelFullMode.Wait // (par défaut) - attendre qu'un emplacement se libère
});

Ici, seulement trois éléments peuvent être présents dans le channel en même temps. Si le producteur essaie d'en écrire un quatrième — il attendra.

Plusieurs producteurs et consommateurs

var channel = Channel.CreateUnbounded<int>();

// 2 producteurs
for (int producerId = 0; producerId < 2; producerId++)
{
    Task.Run(async () =>
    {
        for (int i = 0; i < 5; i++)
        {
            int value = producerId * 100 + i;
            Console.WriteLine($"Producteur {producerId} : Dépose {value}");
            await channel.Writer.WriteAsync(value);
            await Task.Delay(50);
        }
        // Chaque producteur appelle Complete() — dangereux !
    });
}
// Astuce : Complete() ne doit être appelé qu'une seule fois, quand TOUS les producteurs ont fini.
// Pour l'exemple on laisse une seule tâche-consommateur :
Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Le consommateur a reçu {item}");
        await Task.Delay(100);
    }
});

Attention ! Le channel doit être fermé (via Complete()) seulement après que tous les producteurs ont fini. Sinon certains tenteront d'écrire alors que le channel est déjà fermé — et ça provoque une exception. Dans la pratique on utilise généralement un compteur de tâches ou Task.WhenAll.

5. Pratique : Traitement d'images via un channel

Complexifions un peu ! Imaginons qu'on ait un dossier d'images. Une tâche cherche les images et place leurs chemins dans le channel, une autre prend le chemin et fait quelque chose d'utile avec le fichier (par ex. calcule la taille ou convertit).

Remarque : pour simplifier l'exemple on travaillera sur les noms de fichiers (sans traitement réel des images), mais le principe est identique.

using System;
using System.IO;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        var channel = Channel.CreateBounded<string>(5);

        // Producteur : recherche des fichiers .jpg dans le dossier
        var producer = Task.Run(async () =>
        {
            foreach (var file in Directory.EnumerateFiles(@"images", "*.jpg"))
            {
                await channel.Writer.WriteAsync(file);
                Console.WriteLine($"Ajouté au channel : {file}");
                await Task.Delay(50); // on simule le délai de recherche
            }
            channel.Writer.Complete(); // fin du channel
        });

        // Consommateur : lit et "traite" les fichiers
        var consumer = Task.Run(async () =>
        {
            await foreach (var file in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"Traitement du fichier : {file}");
                await Task.Delay(200); // simulation du traitement
            }
            Console.WriteLine("Toutes les images ont été traitées !");
        });

        await Task.WhenAll(producer, consumer);
    }
}

6. Configurer le Channel : options et nuances

Les channels se configurent via des options à la création — voici les principaux paramètres pour les channels limités :

Capacity

SingleWriter

SingleReader

FullMode

Exemple avec options :

var options = new BoundedChannelOptions(10)
{
    SingleWriter = false,
    SingleReader = true,
    FullMode = BoundedChannelFullMode.Wait
};
var channel = Channel.CreateBounded<string>(options);

7. Méthodes asynchrones : ReadAsync, WriteAsync, ReadAllAsync

Pourquoi async est-il si important ?

Les méthodes WriteAsync et ReadAsync n'empêchent pas le thread d'avancer ! S'il n'y a rien à lire, la task est mise en pause, libérant le thread pour d'autres tâches. C'est crucial pour les applications serveurs et UI où un blocage peut causer des "freezes".

ReadAllAsync — la commodité du C# moderne

On peut itérer de façon asynchrone :

await foreach (var item in channel.Reader.ReadAllAsync())
{
    // On travaille avec item
}

Channel<T> vs collections thread-safe : quelle différence ?

ConcurrentQueue<T>/BlockingCollection<T> sont bien pour les scénarios multi-thread classiques, mais elles ne conviennent pas pour l'asynchronisme pur (scénarios avec await).

Channel<T> a été conçu pour les pipelines asynchrones. En termes de sécurité thread, les deux approches fonctionnent, mais les channels offrent plus de flexibilité et une meilleure intégration avec les fonctionnalités modernes de C# (comme IAsyncEnumerable et autres).

8. Erreurs et pièges typiques

N'oubliez pas d'appeler Complete() sur le writer quand tous les éléments ont été ajoutés ! Sinon le consommateur restera bloqué en attente de nouveaux éléments indéfiniment.

N'appelez pas Complete() plusieurs fois si vous avez plusieurs writers — faites-le seulement après que absolument tous les producteurs ont terminé leur travail.

Après la fermeture du channel on ne peut plus écrire d'éléments, mais on peut lire ceux qui restent.

Race condition lors d'écritures simultanées : si le channel est fermé et que quelqu'un essaie encore d'écrire — vous obtiendrez une exception.