Producer / Consumer

Decouples work generation from work processing using an in-memory queue as an intermediate buffer.

Context

You have a flow that generates work faster than it can be processed. If the generator waits for the processor, you block the request. If you couple them directly, any slowness in the processor propagates upstream.

Solution

Split into two independent roles:

Producer: generates work and deposits it into a queue. It doesn't know who will process it or when.
Consumer: reads from the queue at its own pace and processes each item.

The queue acts as a buffer: it absorbs the speed difference between both sides.

[Producer]  →  [ queue / Channel<T> ]  →  [Consumer]
   fast              buffer                   slow

The consumer sleeps when the queue is empty — no polling, no wasted CPU. It wakes up only when an item arrives.

#How it works internally

The magic is in TaskCompletionSource<T>. When the consumer calls WaitToReadAsync and the queue is empty, the runtime suspends that code and completely releases the thread. When the producer calls WriteAsync, it internally calls TrySetResult(true) — that resolves the Task that was waiting and the runtime schedules the consumer to continue.

CSHARP

Producer: WriteAsync("item")
  → enqueues the item
  → TrySetResult(true)        ← wakes up the consumer
    → runtime schedules continuation
      → consumer receives the item
        → processes it
          → goes back to sleep

No polling. No Thread.Sleep. No CPU consumed while waiting.

#C# Example

CSHARP

// Registration (singleton — same instance for producer and consumer)
services.AddSingleton(Channel.CreateBounded<string>(new BoundedChannelOptions(500)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = false
}));

services.AddHostedService<MyConsumer>();

CSHARP

// Producer — can be a middleware, an endpoint, any service
public class MyProducer(Channel<string> channel)
{
    public async Task EnqueueAsync(string item)
    {
        await channel.Writer.WriteAsync(item);
    }
}

CSHARP

// Consumer — runs in background, independent of the HTTP cycle
public class MyConsumer(Channel<string> channel, ILogger<MyConsumer> logger)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var item in channel.Reader.ReadAllAsync(ct))
        {
            try
            {
                await ProcessAsync(item);
            }
            catch (Exception ex)
            {
                // try/catch INSIDE the loop — if one item fails, the consumer stays alive
                logger.LogError(ex, "Failed processing {Item}", item);
            }
        }
    }
}

Warning

The try/catch must be inside the await foreach, not outside. If you put it outside, one exception kills the entire loop and the consumer never processes anything again — with no visible error.

#Backpressure

With BoundedChannelOptions you control what happens when the queue fills up:

FullMode	Behavior
`Wait`	Producer waits until a slot is available. The HTTP request stalls.
`DropNewest`	Discards the incoming item. Producer never blocks.
`DropOldest`	Discards the oldest item in the queue.

For HTTP endpoints that cannot block, TryWrite is the alternative:

CSHARP

if (!channel.Writer.TryWrite(item))
    return Results.StatusCode(503); // saturated, retry later

#Multiple consumers in parallel

If processing is slow, you can spin up N consumers on the same channel:

CSHARP

// Register the same HostedService N times
services.AddHostedService<MyConsumer>();
services.AddHostedService<MyConsumer>();
services.AddHostedService<MyConsumer>();

All read from the same singleton Channel<T>. Each item is processed by exactly one consumer.

#Monitoring

CSHARP

app.MapGet("/health/queue", (Channel<string> channel) =>
    Results.Ok(new { pending = channel.Reader.Count }));

If pending grows indefinitely, the bottleneck is in the consumer — you need more instances or caching.

Tradeoffs

Pro	Con
Producer and consumer completely decoupled	Messages are lost if the process dies (no persistence)
No polling — 0 CPU when the queue is empty	No dead-letter queue — a failing item is discarded
Natural backpressure with `Bounded`	No automatic retries
N consumers in parallel without changing the producer	Does not scale across processes — in-memory only

#When NOT to use it

If you need durability: a process crash loses all queued messages. Use Outbox Pattern + Service Bus instead.
If you need automatic retries with dead-letter queues. Channel<T> has none of that.
If you need to scale horizontally across multiple instances. Channel<T> is in-memory — each process has its own queue. Use RabbitMQ, Azure Service Bus, or Kafka.
If processing is so fast that the enqueue latency matters — a direct call is simpler.

#Comparison with Service Bus

	`Channel<T>`	Service Bus / RabbitMQ
Persistence	No — RAM	Yes — disk/network
Dead-letter	No	Yes
Retries	Manual	Automatic
Multi-process	No	Yes
Latency	Microseconds	Milliseconds
Infrastructure	None	External broker

Channel<T> is a Service Bus in memory. No broker, no network, no operational cost — in exchange for durability.

#Availability

Channel<T> is available since .NET Core 3.0. No additional NuGet required.

CSHARP

using System.Threading.Channels;

Not available in .NET Framework. The legacy equivalent is BlockingCollection<T>, but it blocks real threads instead of using async/await.

#concurrency #async #channel #queue #background-service