# Dotnet Channels > Implements producer/consumer queues. Channel, bounded/unbounded, backpressure, drain. - Skill: `rudironsoni/dotnet-channels` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add rudironsoni/dotnet-channels` - Raw SKILL.md: https://api.skillmd.com/api/skills/rudironsoni/dotnet-channels/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools, AI & ML - License: MIT - Author: rudironsoni (https://skillmd.com/u/rudironsoni) - Updated: 2026-09-10 - Page: https://skillmd.com/skills/rudironsoni/dotnet-channels --- # dotnet-channels Deep guide to `System.Threading.Channels` for high-performance, thread-safe producer/consumer communication in .NET. Covers channel creation, backpressure strategies, IAsyncEnumerable integration, and graceful shutdown patterns. ## Scope - Channel creation (bounded and unbounded) - Backpressure strategies and capacity management - IAsyncEnumerable integration with channel readers - Graceful shutdown and drain patterns ## Out of scope - Hosted service lifecycle and BackgroundService registration -- see [skill:dotnet-background-services] - Async/await fundamentals and cancellation token propagation -- see [skill:dotnet-csharp-async-patterns] Cross-references: [skill:dotnet-background-services] for integrating channels with hosted services, [skill:dotnet-csharp-async-patterns] for async patterns used in channel consumers. --- ## Channel Fundamentals A `Channel` is a thread-safe data structure with separate `ChannelWriter` and `ChannelReader` endpoints. Writers produce items, readers consume them -- the channel handles all synchronization. ````csharp // Create a channel and separate the endpoints Channel channel = Channel.CreateUnbounded(); ChannelWriter writer = channel.Writer; ChannelReader reader = channel.Reader; ```text ### Bounded vs Unbounded | Aspect | Bounded | Unbounded | | ------------- | ---------------------------------------------- | ---------------------------------- | | Creation | `Channel.CreateBounded(capacity)` | `Channel.CreateUnbounded()` | | Back-pressure | Yes -- `FullMode` controls behavior when full | No -- grows without limit | | Memory safety | Capped at `capacity` items | Can exhaust memory under load | | Use when | Production workloads, untrusted producer rates | Guaranteed-low-volume, prototyping | ```csharp // Bounded -- preferred for production var bounded = Channel.CreateBounded(new BoundedChannelOptions(capacity: 1000) { FullMode = BoundedChannelFullMode.Wait }); // Unbounded -- use only when you control the producer rate var unbounded = Channel.CreateUnbounded(); ```text --- ## BoundedChannelFullMode Controls what happens when a bounded channel is full and a producer attempts to write. | Mode | Behavior | Use case | | ------------ | ---------------------------------------------------------------- | ---------------------------------------------------- | | `Wait` | `WriteAsync` blocks until space is available | Default. Reliable delivery with back-pressure | | `DropOldest` | Drops the oldest item in the channel to make room | Telemetry, metrics -- latest data matters most | | `DropNewest` | Drops the item being written (newest) | Rate limiting -- discard excess incoming work | | `DropWrite` | Drops the item being written and returns `false` from `TryWrite` | Non-blocking fire-and-forget with overflow detection | ```csharp // DropOldest -- telemetry pipeline where stale readings are expendable var telemetryChannel = Channel.CreateBounded(new BoundedChannelOptions(500) { FullMode = BoundedChannelFullMode.DropOldest }); // DropWrite -- non-blocking enqueue with overflow awareness var logChannel = Channel.CreateBounded(new BoundedChannelOptions(10_000) { FullMode = BoundedChannelFullMode.DropWrite }); if (!logChannel.Writer.TryWrite(entry)) { // Channel full -- item was dropped; track overflow metric overflowCounter.Add(1); } ```text ### itemDropped Callback (.NET 7+) Starting in .NET 7, bounded channels with drop modes accept an `itemDropped` callback that fires whenever an item is discarded. Use this for metrics, logging, or resource cleanup on dropped items. ```csharp var channel = Channel.CreateBounded(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.DropOldest }, itemDropped: (item, writer) => { logger.LogWarning("Dropped item due to channel overflow: {Id}", item.Id); droppedItemsCounter.Add(1); // Clean up disposable items if needed (item as IDisposable)?.Dispose(); }); ```text The callback receives the dropped item and the `ChannelWriter` (useful if you need to re-route items to a fallback channel). --- ## Producer Patterns ### Single Producer ```csharp // Write with back-pressure (bounded channels) await writer.WriteAsync(item, cancellationToken); // Non-blocking write attempt (returns false if channel is full or completed) if (!writer.TryWrite(item)) { // Handle overflow -- log, retry, or discard } ```text ### Multiple Producers Multiple producers can call `WriteAsync` or `TryWrite` concurrently without external locking. The channel is internally thread-safe. ```csharp // Multiple API endpoints enqueueing work into a shared channel app.MapPost("/api/orders/{id}/process", async ( string id, ChannelWriter writer, CancellationToken ct) => { await writer.WriteAsync(new OrderCommand(id, "process"), ct); return Results.Accepted(); }); app.MapPost("/api/orders/{id}/cancel", async ( string id, ChannelWriter writer, CancellationToken ct) => { await writer.WriteAsync(new OrderCommand(id, "cancel"), ct); return Results.Accepted(); }); ```bash ### Signaling Completion Call `Complete()` or `TryComplete()` when no more items will be produced. This lets consumers detect the end of the stream. ```csharp // Signal completion -- no more items will be written writer.Complete(); // TryComplete is idempotent -- safe to call multiple times writer.TryComplete(); // Signal completion with an error writer.TryComplete(new InvalidOperationException("Source failed")); ```text --- ## Consumer Patterns ### Single Consumer -- ReadAsync Loop The classic pattern: wait for an item, process it, repeat. ```csharp while (await reader.WaitToReadAsync(cancellationToken)) { while (reader.TryRead(out var item)) { await ProcessAsync(item, cancellationToken); } } ```text This two-loop pattern is preferred over `ReadAsync` alone because it drains all available items before awaiting again, reducing async state machine overhead. ### Single Consumer -- ReadAsync (Simpler) For simpler cases where per-item overhead is acceptable: ```csharp try { while (true) { var item = await reader.ReadAsync(cancellationToken); await ProcessAsync(item, cancellationToken); } } catch (ChannelClosedException) { // Writer called Complete() -- no more items } ```text ### Multiple Consumers (Fan-Out) Scale processing by running multiple consumer tasks. The channel ensures each item is read by exactly one consumer. ```csharp public sealed class ScaledChannelProcessor( ChannelReader reader, IServiceScopeFactory scopeFactory, ILogger logger) : BackgroundService { private const int WorkerCount = 3; protected override async Task ExecuteAsync(CancellationToken stoppingToken) { var workers = Enumerable.Range(0, WorkerCount) .Select(i => ConsumeAsync(i, stoppingToken)); await Task.WhenAll(workers); } private async Task ConsumeAsync(int workerId, CancellationToken ct) { logger.LogDebug("Consumer {WorkerId} started", workerId); while (await reader.WaitToReadAsync(ct)) { while (reader.TryRead(out var item)) { try { using var scope = scopeFactory.CreateScope(); var handler = scope.ServiceProvider .GetRequiredService(); await handler.HandleAsync(item, ct); } catch (Exception ex) { logger.LogError(ex, "Consumer {WorkerId}: error processing {ItemId}", workerId, item.Id); } } } logger.LogDebug("Consumer {WorkerId} stopped", workerId); } } ```text --- ## IAsyncEnumerable Integration `ChannelReader.ReadAllAsync()` returns an `IAsyncEnumerable`, enabling `await foreach` consumption and integration with LINQ async operators. ### Basic await foreach ```csharp await foreach (var item in reader.ReadAllAsync(cancellationToken)) { await ProcessAsync(item, cancellationToken); } // Loop exits when writer calls Complete() and all items are consumed ```text `ReadAllAsync` is the simplest consumption pattern. It handles `WaitToReadAsync`/`TryRead` internally and completes when the channel is closed. ### Streaming from an API Endpoint Channels combine naturally with ASP.NET Core streaming responses. Return the `IAsyncEnumerable` directly -- minimal APIs will stream items as JSON array elements: ```csharp app.MapGet("/api/events/stream", ( ChannelReader reader, CancellationToken ct) => reader.ReadAllAsync(ct)); ```csharp ### LINQ Async Operators With the `System.Linq.Async` NuGet package, channel streams compose with familiar LINQ operators: ```csharp // NuGet: System.Linq.Async await foreach (var batch in reader.ReadAllAsync(ct) .Where(item => item.Priority >= Priority.High) .Buffer(50) // Collect into batches of 50 .WithCancellation(ct)) { await BulkProcessAsync(batch, ct); } ```text ### Producing an IAsyncEnumerable from a Channel ```csharp async IAsyncEnumerable StreamPricesAsync( string symbol, [EnumeratorCancellation] CancellationToken ct = default) { var channel = Channel.CreateUnbounded(); // Start producer in background _ = Task.Run(async () => { try { await foreach (var tick in marketFeed.SubscribeAsync(symbol, ct)) { await channel.Writer.WriteAsync(tick, ct); } channel.Writer.TryComplete(); } catch (Exception ex) { // Propagate error to reader -- ReadAllAsync will throw channel.Writer.TryComplete(ex); } }, ct); await foreach (var update in channel.Reader.ReadAllAsync(ct)) { yield return update; } } ```text --- ## Performance ### SingleReader / SingleWriter Flags Setting `SingleReader = true` or `SingleWriter = true` on channel options enables lock-free optimizations. The channel trusts these hints -- violating them (multiple concurrent readers when `SingleReader = true`) causes data corruption. ```csharp // Optimal for single-producer, single-consumer pipeline var channel = Channel.CreateBounded(new BoundedChannelOptions(1000) ## Detailed Examples See [references/detailed-examples.md](references/detailed-examples.md) for complete code samples and advanced patterns.