In-Process Queues with Channels
Hand work from endpoints to a background worker through a bounded Channel, apply backpressure and know exactly what is lost on restart.
var queue = Channel.CreateBounded<ReceiptJob>(new BoundedChannelOptions(500)
{ FullMode = BoundedChannelFullMode.Wait, SingleReader = true });
if (!queue.Writer.TryWrite(job)) return Results.StatusCode(503);An endpoint that sends an email or renders a PDF inline makes the client wait for it. The usual in-process fix is a producer/consumer queue: the endpoint writes a job and returns 202 Accepted, and a BackgroundService reads and processes jobs. System.Threading.Channels is the built-in, allocation-light, async queue for this.
The moving parts
Channel.CreateBounded<T>(new BoundedChannelOptions(500) { ... })registered as a singleton.- Producers use
channel.Writer:TryWritereturns at once,WriteAsyncwaits for space. - The consumer uses
await foreach (var job in channel.Reader.ReadAllAsync(ct)), which ends when the writer is completed and the buffer is empty. SingleReader = trueis a promise that only one loop reads.CreateUnboundeduses it to pick a faster single-consumer implementation;CreateBoundedignores it today. Set it only when it is true.
Bounded, and what to do when full
Never use CreateUnbounded for request-driven work. If the consumer slows (SMTP server throttling), an unbounded queue grows until the pod is OOM-killed, and every queued job dies with it. A bound turns that into a decision:
BoundedChannelFullMode | When full | Use for |
|---|---|---|
Wait (default) | WriteAsync waits; TryWrite returns false | Work that must not be dropped. Pair TryWrite with a 503 and Retry-After. |
DropOldest | Removes the oldest buffered item | Latest-value data: stock levels, presence, metrics. |
DropNewest | Removes the newest buffered item | Rare; keeps the earliest backlog. |
DropWrite | Discards the item being written | Best-effort work such as cache warming. |
In the drop modes, TryWrite always returns true, so a silent loss needs the itemDropped callback overload to count drops in a metric. For HTTP producers prefer TryWrite and reject: awaiting WriteAsync in a handler holds the request open and moves the backlog into Kestrel's connection queue, where it is invisible and still unbounded. Capacity counts items, not bytes; size it from throughput times the slowest tolerable delay.
Losing work on restart
The buffer lives in process memory. Everything in it is gone on a deploy, a crash, an OOM kill or a node drain. A graceful stop can drain: complete the writer, finish the buffered items within ShutdownTimeout, then exit. Anything beyond that window is lost, and the client already received 202.
So the decision is about the job, not the technology. A channel is right when losing a job is acceptable or recoverable: thumbnail generation that can be redone, cache refreshes, analytics events, emails that a nightly sweep re-sends from database state. It is wrong for anything that must happen exactly because the client was told it would: payment capture, order fulfilment, webhooks to partners. Those belong in a database row (the outbox) or a durable broker, with a channel at most as a wake-up signal in front of it.
Example
When to use it
- Return 202 from POST /orders/{id}/receipt immediately and send the email from a worker, answering 503 with Retry-After when 500 receipts are already waiting.
- Render invoice PDFs with four concurrent consumers while capping memory with a bounded buffer of 200 jobs.
- Push the latest stock level per SKU to the search index, dropping superseded readings when the indexer falls behind.
More examples
Drain buffered jobs on graceful shutdown
The default <code>StopAsync</code> cancels first and waits second, which abandons the buffer. Completing the writer and waiting for <code>ExecuteTask</code> lets buffered jobs finish within <code>ShutdownTimeout</code>; only then is the stopping token cancelled. Endpoints must treat a failed <code>TryWrite</code> after completion as 503.
using System.Threading.Channels;
public sealed record ThumbnailJob(int ProductId, string ImagePath);
public sealed class ThumbnailWorker(Channel<ThumbnailJob> queue, ILogger<ThumbnailWorker> logger) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// No token on ReadAllAsync: the loop ends when the writer is completed and the buffer is empty.
await foreach (var job in queue.Reader.ReadAllAsync())
{
try
{
await RenderAsync(job, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
logger.LogWarning("Shutdown timeout reached; {Remaining} thumbnails lost", queue.Reader.Count + 1);
return;
}
}
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
queue.Writer.TryComplete(); // refuse new work
if (ExecuteTask is not null)
await ExecuteTask.WaitAsync(cancellationToken) // drain until the host's shutdown deadline
.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
await base.StopAsync(cancellationToken); // then cancel whatever is still running
}
private static Task RenderAsync(ThumbnailJob job, CancellationToken ct) => Task.Delay(100, ct);
}Several consumers with a parallelism limit
<code>Parallel.ForEachAsync</code> over <code>ReadAllAsync</code> gives four concurrent consumers with one line of coordination. Set <code>SingleReader = false</code> on this channel. Each job gets its own DI scope, and the per-item catch matters: one exception from the body cancels the remaining iterations and faults the worker.
using System.Threading.Channels;
public sealed record InvoicePdfJob(string InvoiceNumber);
public interface IInvoiceRenderer { Task RenderAsync(string invoiceNumber, CancellationToken ct); }
public sealed class InvoicePdfWorker(
Channel<InvoicePdfJob> queue,
IServiceScopeFactory scopes,
ILogger<InvoicePdfWorker> logger) : BackgroundService
{
protected override Task ExecuteAsync(CancellationToken stoppingToken) =>
Parallel.ForEachAsync(
queue.Reader.ReadAllAsync(stoppingToken),
new ParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = stoppingToken },
async (job, ct) =>
{
try
{
await using var scope = scopes.CreateAsyncScope();
await scope.ServiceProvider.GetRequiredService<IInvoiceRenderer>().RenderAsync(job.InvoiceNumber, ct);
}
catch (Exception ex) when (!ct.IsCancellationRequested)
{
// An unhandled exception here would stop Parallel.ForEachAsync and the whole worker.
// A renderer timeout is caught too: only a cancelled loop token means shutdown.
logger.LogError(ex, "Rendering {Invoice} failed", job.InvoiceNumber);
}
});
}Latest-value data with DropOldest and a drop counter
For stock levels pushed to a search index, a newer reading supersedes an older one, so dropping the oldest is correct and the producer never blocks. <code>TryWrite</code> always succeeds in this mode, so the <code>itemDropped</code> callback is the only signal that the consumer is falling behind; alert on the counter.
using System.Diagnostics.Metrics;
using System.Threading.Channels;
public sealed record StockLevel(string Sku, int Available, DateTimeOffset At);
public static class StockFeedRegistration
{
private static readonly Meter Meter = new("Inventory.StockFeed");
private static readonly Counter<long> Dropped = Meter.CreateCounter<long>("stock_feed.dropped");
public static IServiceCollection AddStockFeed(this IServiceCollection services) =>
services.AddSingleton(_ => Channel.CreateBounded<StockLevel>(
new BoundedChannelOptions(10_000)
{
FullMode = BoundedChannelFullMode.DropOldest,
SingleReader = true,
},
// A bounded tag only: one tag value per SKU would create one time series per SKU.
itemDropped: _ => Dropped.Add(1, new KeyValuePair<string, object?>("feed", "search-index"))));
}
Discussion