Rx.NET + Channels + BackgroundService avec les flux asynchrones en .NET
Dans l’article précédent "Rx.NET dans une API Web : modéliser les flux plutôt que les appels", nous avons posé une base importante : "arrêter de penser en appels, commencer à penser en flux".Sur le papier, Rx.NET coche toutes les cases. On compose, on transforme, on enchaîne. C’est lisible et élégant.Mais dès qu’on sort du labo et qu’on arrive sur une vraie API, le décor change. Les événements n’arrivent pas proprement, ils arrivent en rafale. L'ensemble des traitements prennent du temps et en plus les dépendances externes ralentissent tout.Et c'est là que le flux Rx.NET commence va montrer ses limites.## Le flux parfait… dans un monde imparfaitUn IObservable fonctionne très bien tant que le rythme est maîtrisé.```
var stream = Observable.FromEventPattern<OrderCreatedEvent>(handler);
stream
.Select(e => Process(e))
.Subscribe();
Ce genre de pipeline est agréable à lire. Il repose sur l'hypothèse implicite que le système est capable d’absorber tout ce qu’on lui envoie.Dans la réalité, ce n’est jamais vrai. Les événements s’accumulent, les traitements prennent du retard, et tu n’as aucun moyen simple de ralentir la source.C’est au niveau du débit que les problèmes commencent.## Remettre une notion essentielle : la pressionCe que Rx.NET ne force pas, les `Channel<T>` le rendent explicite.On arrête de consommer directement un flux. On introduit un point de passage contrôlé.
var channel = Channel.CreateBounded<Order>(100);
Ce simple changement modifie complètement le comportement du système. On bufferise, on rejette la demande quand le système est saturé.Côté API, ça devient presque trivial :
app.MapPost("/orders", async (Order order) =>
{
await channel.Writer.WriteAsync(order);
});
À partir de là, l’API n’est plus responsable du traitement. Elle se contente d’alimenter le pipeline. C’est un changement subtil, mais fondamental.## Faire tourner le système en continuLe BackgroundService vient donner une réalité à ce pipeline.On ne déclenche plus le traitement au moment de la requête et on laisse un worker consommer à son rythme.
public class OrderWorker : BackgroundService
{
private readonly Channel<Order> _channel;
public OrderWorker(Channel<Order> channel)
{
_channel = channel;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var order in _channel.Reader.ReadAllAsync(stoppingToken))
{
await Process(order);
}
}
}
Ce qui est intéressant ici, c’est le découplage.L’API pousse. Le worker consomme. Entre les deux, un buffer absorbe les variations.On commence à avoir quelque chose de stable.## Réintroduire Rx.NET au bon endroitÀ ce stade, on pourrait se contenter du ***Channel ***et du ***BackgroundService ***car ça fonctionne déjà très bien.Mais on perd la capacité à composer intelligemment les flux.C’est là que **Rx.NET** revient avec un rôle différent.On ne l’utilise plus pour recevoir les événements. On l’utilise pour les transformer.
var observable = channel.Reader
.ReadAllAsync()
.ToObservable();
observable
.Buffer(TimeSpan.FromSeconds(5))
.Where(batch => batch.Any())
.Subscribe(async batch =>
{
await ProcessBatch(batch);
});
On garde la robustesse du Channel, et on retrouve la puissance de composition de Rx.## Une architecture qui tient dans le tempsQuand on prend un peu de recul, on voit apparaître un pipeline assez clair.[](https://xavierraboteur.wordpress.com/wp-content/uploads/2026/03/mermaid-diagram.png)C’est un flux structuré, avec des responsabilités bien définies.## Un exemple concretPrenons quelque chose de simple, dans la continuité de l’article précédent (***"[Rx.NET dans une API Web : modéliser les flux plutôt que les appels](/articles/rx-net-dans-une-api-web-modeliser-les-flux-plutot-que-les-appels)"***).
public record EmployeeEvent(string Type, int EmployeeId);
L’API ne fait qu’émettre :
await channel.Writer.WriteAsync(new EmployeeEvent("CREATED", id));
Derrière, le pipeline devient intéressant :
channel.Reader
.ReadAllAsync()
.ToObservable()
.GroupBy(e => e.EmployeeId)
.Subscribe(group =>
{
group
.Throttle(TimeSpan.FromSeconds(1))
.Subscribe(e => HandleEmployeeEvent(e));
});
[https://github.com/xraboteu/rx-channels-worker-pipeline](https://github.com/xraboteu/rx-channels-worker-pipeline)