Producer/consumer en C# : pourquoi Channel change la donne

 

De ConcurrentQueue à BlockingCollection, puis à l’asynchrone natif : ce que chaque brique résout, et ce qu’elle laisse de côté

Un service qui encaisse des événements à la volée, une API qui absorbe des pics de requêtes, un traitement de fichiers qui arrive par lots irréguliers : dans chacun de ces cas, quelque chose produit du travail à un rythme qu’on ne maîtrise pas complètement, celui d’un système externe, d’un utilisateur, d’une intégration tierce. Et à côté, un composant est censé traiter ce travail, à son propre rythme, souvent plus lent. Dès que ces deux rythmes divergent, une question se pose : que fait-on des données en attente ? Les traiter immédiatement au risque de saturer le système ? Les mettre de côté, et alors où, avec quelle garantie qu’aucune ne se perd ? C’est exactement ce que résout le pattern producer/consumer : un ou plusieurs producers alimentent une file d’attente, un ou plusieurs consumers la vident à leur rythme. Le producer n’attend jamais le consumer, et le consumer ne se fait jamais déborder. En C#, on va suivre la progression naturelle : la brique de base, ConcurrentQueue, la réponse historique bâtie dessus, BlockingCollection, et une approche plus récente pensée spécifiquement pour l’asynchrone, Channel.

 

ConcurrentQueue : la brique de base, sans blocage natif

Depuis .NET 4.0, ConcurrentQueue est la collection thread-safe de référence en C# pour ce genre de scénario : plusieurs threads peuvent y écrire et y lire sans se marcher dessus. Mais elle ne résout qu’une partie du problème. TryDequeue() renvoie immédiatement, qu’il y ait un élément ou non ; elle ne bloque jamais pour attendre qu’un élément arrive. Le réflexe le plus simple pour combler ce manque, c’est une boucle qui vérifie régulièrement si un élément est disponible :

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

ConcurrentQueue n’ayant aucune notion de fin intégrée, il faut la construire soi-même : ici, un simple booléen producerDone que le producer positionne une fois terminé, et que le consumer surveille en plus de la file elle-même (pour ne pas s’arrêter tant qu’il reste des éléments à traiter). Volatile.Write/Volatile.Read garantissent que ce changement est bien visible d’un thread à l’autre. Rien de compliqué en soi, mais c’est un état partagé de plus à gérer soi-même. Ça fonctionne, mais avec un compromis frustrant. Un Task.Delay court consomme du CPU pour rien quand la file est vide, un délai plus long ajoute de la latence avant que chaque nouvel élément soit traité. On est en train d’échanger un problème (savoir quand un élément est disponible) contre un autre (choisir le bon intervalle de vérification). C’est exactement ce que propose BlockingCollection, disponible dans le même namespace depuis .NET 4.0 : un vrai blocage du thread consommateur, sans polling, tant qu’aucun élément n’est disponible, et une notion de fin intégrée. Voyons ce que ça change.

 

BlockingCollection : la même mécanique, avec le blocage en plus

BlockingCollection est une couche de blocage posée par-dessus ConcurrentQueue. Sous le capot, BlockingCollection stocke ses éléments dans une ConcurrentQueue (c’est son type de collection sous-jacent par défaut, selon la doc Microsoft), et gère le réveil du consumer avec un SemaphoreSlim interne. Chaque Add() libère ce sémaphore, chaque Take() attend dessus : le même besoin de synchronisation qu’on a contourné plus tôt avec du polling, mais résolu ici proprement avec un signal plutôt qu’une boucle d’attente active, capacité maximale comprise et gérée correctement en interne. Add() insère un élément, Take() bloque jusqu’à ce qu’un élément soit disponible : plus de polling, plus de délai à calibrer. On reprend exactement le même scénario que pour ConcurrentQueue, avec BlockingCollection à la place :

 

 

 

 

 

 

 

 

 

 

 

 

 

 

CompleteAdding() signale la fin de la production, ce qui permet au foreach du consumer de se terminer proprement.

 

Deux réponses, encore aucune vraiment asynchrone

Le Task.Delay de ConcurrentQueue n’est qu’un bricolage qui simule une attente, pas un vrai producer/consumer async : le thread continue de vérifier activement, juste avec des pauses entre deux tentatives. BlockingCollection, elle, ne fait aucun semblant : elle bloque réellement le thread appelant, point final. En revanche, Channel est la première des trois à être pensée nativement pour l’asynchrone, sans bricolage ni blocage réel.

 

Channel <T> : le même pattern, pensé pour l’asynchrone

Channel <T> arrive avec .NET Core 3.0, dans le namespace System.Threading.Channels : plus de thread bloqué à attendre, juste un await qui libère le thread tant qu’aucun élément n’est disponible. Un Channel s’utilise via deux interfaces distinctes : ChannelWriter pour écrire, ChannelReader pour lire.

 

 

 

 

 

 

 

 

 

 

 

 

 

WriteAsync() et ReadAllAsync() sont tous deux asynchrones : le thread qui attend n’est jamais bloqué au sens classique, il est simplement suspendu et rendu disponible pour d’autres tâches jusqu’à ce qu’un élément arrive. Writer.Complete() joue le même rôle que CompleteAdding() sur BlockingCollection : plus rien n’arrivera, et ReadAllAsync() se termine proprement une fois le channel vidé. Channel.CreateUnbounded() crée un channel sans limite de taille. Pour retrouver l’équivalent du boundedCapacity de BlockingCollection, on utilise Channel.CreateBounded(capacité), qui force le producer à attendre (de façon asynchrone, là aussi) dès que la capacité est atteinte. Ce comportement par défaut (attendre) n’est qu’une option parmi d’autres, contrôlée par BoundedChannelFullMode : Wait (le défaut, celui qu’on vient de voir) fait patienter le producer jusqu’à ce qu’une place se libère. • DropOldest évince l’élément le plus ancien du channel pour faire de la place au nouveau. • DropNewest évince l’élément le plus récent encore non lu, au profit du nouveau. • DropWrite rejette directement le nouvel élément qu’on essayait d’écrire. Utile par exemple pour un flux de télémétrie où seule la donnée la plus récente compte, et où bloquer le producer n’a aucun sens.

 

 

 

 

À côté de WriteAsync() et ReadAllAsync(), Channel expose aussi des variantes synchrones, non bloquantes : TryWrite() et TryRead(). Elles renvoient immédiatement un booléen (écriture ou lecture réussie ou non) sans jamais attendre, utiles quand on ne veut pas payer le coût d’un await alors qu’une valeur est probablement déjà disponible, ou quand on écrit depuis un contexte qui ne peut pas être asynchrone. Dernier point qui compte en pratique : WriteAsync(), ReadAsync() et ReadAllAsync() acceptent tous un CancellationToken, tout comme Add() et Take() sur BlockingCollection. Sur BlockingCollection, l’annulation réveille un thread réellement bloqué en attente. Sur Channel, elle interrompt une attente asynchrone qui ne mobilisait déjà aucun thread pendant qu’elle patientait.

 

Les autres briques du langage

Channel <T> n’est pas la seule option du langage pour ce pattern. TPL Dataflow (System.Threading.Tasks.Dataflow, package NuGet séparé) propose des blocs comme BufferBlock ou ActionBlock, également asynchrones, mais avec une portée plus large orientée pipeline : composition de blocs, parallélisme intégré, routage conditionnel. Plus riche que Channel, aussi plus lourd à apprendre pour un simple besoin producer/consumer. IAsyncEnumerable peut aussi jouer ce rôle dans un cas particulier : un seul producer, un seul consumer, sans vrai découplage entre les deux. La raison tient à son fonctionnement même : un IAsyncEnumerable est pull, le consumer déclenche chaque production en appelant MoveNextAsync(), alors qu’un Channel est push, le producer produit indépendamment du rythme du consumer. ChannelReader.ReadAllAsync() en est d’ailleurs un exemple direct, puisqu’elle retourne justement un IAsyncEnumerable.

 

Conclusion

Channel <T> n’est pas qu’une nouvelle collection à ajouter à la liste. Elle marque un changement de posture : longtemps, l’asynchrone en C# a été une option qu’on ajoutait par-dessus des briques pensées pour le monde synchrone, avec les frictions que ça implique. Avec Channel, l’async devient le point de départ de la conception, pas une couche ajoutée après coup. Elle reflète une direction plus générale du framework : proposer des primitives natives pour l’asynchrone plutôt que d’attendre que les développeurs les reconstruisent eux-mêmes à partir de briques plus anciennes. IAsyncEnumerable, les async streams, et maintenant Channel s’inscrivent dans la même logique. Pour les équipes qui maintiennent du code écrit avant .NET Core 3.0, ça vaut le coup de regarder d’un œil neuf les BlockingCollection qui traînent dans la base de code. Pas pour les remplacer systématiquement, elles restent parfaitement valables dans un contexte synchrone, mais pour repérer celles qui bloquent des threads dans un contexte qui, lui, est devenu asynchrone au fil du temps. Le même regard vaut pour un ConcurrentQueue couplé à du polling ou à un SemaphoreSlim fait maison : c’est exactement le genre de solution artisanale que Channel <T> rend aujourd’hui inutile.