dotnet-channels

dotnet-channels

System.Threading.Channels 深度指南，介绍如何在 .NET 中使用它实现高性能、线程安全的生产者/消费者通信。内容涵盖通道创建、背压策略、与 IAsyncEnumerable 集成，以及优雅关闭模式。

不在本指南范围内： 托管服务生命周期和 BackgroundService 注册由 [skill:dotnet-background-services] 负责。Async/await 基础知识和取消令牌传递由 [skill:dotnet-csharp-async-patterns] 负责。

交叉参考：[skill:dotnet-background-services] 介绍如何将通道集成到托管服务中；[skill:dotnet-csharp-async-patterns] 介绍通道消费者使用的异步模式。

---

Channel 基础知识

Channel<T> 是一种线程安全的数据结构，具有分离的 ChannelWriter<T> 和 ChannelReader<T> 端点。写入者生成项，读取者消费项——通道负责处理所有同步。

// Create a channel and separate the endpoints
Channel<WorkItem> channel = Channel.CreateUnbounded<WorkItem>();
ChannelWriter<WorkItem> writer = channel.Writer;
ChannelReader<WorkItem> reader = channel.Reader;

有界与无界

方面	有界	无界
创建	Channel.CreateBounded<T>(capacity)	Channel.CreateUnbounded<T>()
背压	有——FullMode 控制通道已满时的行为	无——无限增长
内存安全	上限为 capacity 项	高负载下可能耗尽内存
适用场景	生产工作负载、不受信任的生产者速率	可确保低流量的场景、原型开发

// Bounded -- preferred for production
var bounded = Channel.CreateBounded<WorkItem>(new BoundedChannelOptions(capacity: 1000)
{
    FullMode = BoundedChannelFullMode.Wait
});

// Unbounded -- use only when you control the producer rate
var unbounded = Channel.CreateUnbounded<WorkItem>();

---

BoundedChannelFullMode

控制有界通道已满且生产者尝试写入时的行为。

模式	行为	适用场景
Wait	WriteAsync 会阻塞，直到有可用空间	默认模式。通过背压确保可靠交付
DropOldest	丢弃通道中最旧的项以腾出空间	遥测、指标——最新数据最重要
DropNewest	丢弃正在写入的项（最新项）	限速——丢弃超出的传入工作
DropWrite	丢弃正在写入的项，且 TryWrite 返回 false	非阻塞即发即弃，并检测溢出

// DropOldest -- telemetry pipeline where stale readings are expendable
var telemetryChannel = Channel.CreateBounded<SensorReading>(new BoundedChannelOptions(500)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

// DropWrite -- non-blocking enqueue with overflow awareness
var logChannel = Channel.CreateBounded<LogEntry>(new BoundedChannelOptions(10_000)
{
    FullMode = BoundedChannelFullMode.DropWrite
});

if (!logChannel.Writer.TryWrite(entry))
{
    // Channel full -- item was dropped; track overflow metric
    overflowCounter.Add(1);
}

itemDropped 回调（.NET 7+）

从 .NET 7 开始，使用丢弃模式的有界通道支持 itemDropped 回调；每当有项被丢弃时都会触发该回调。可用于记录指标、日志，或清理被丢弃项所占用的资源。

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();
});

回调会接收被丢弃的项和 ChannelWriter<T>（如果需要将项重新路由到备用通道，这会很有用）。

---

生产者模式

单个生产者

// 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
}

多个生产者

多个生产者可以并发调用 WriteAsync 或 TryWrite，无需外部锁定。通道内部是线程安全的。

// Multiple API endpoints enqueueing work into a shared channel
app.MapPost("/api/orders/{id}/process", async (
    string id,
    ChannelWriter<OrderCommand> writer,
    CancellationToken ct) =>
{
    await writer.WriteAsync(new OrderCommand(id, "process"), ct);
    return Results.Accepted();
});

app.MapPost("/api/orders/{id}/cancel", async (
    string id,
    ChannelWriter<OrderCommand> writer,
    CancellationToken ct) =>
{
    await writer.WriteAsync(new OrderCommand(id, "cancel"), ct);
    return Results.Accepted();
});

发出完成信号

不再有项要生成时，调用 Complete() 或 TryComplete()。这样消费者就能检测到流的结束。

// 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"));

---

消费者模式

单个消费者——ReadAsync 循环

经典模式：等待一个项，处理它，然后重复。

while (await reader.WaitToReadAsync(cancellationToken))
{
    while (reader.TryRead(out var item))
    {
        await ProcessAsync(item, cancellationToken);
    }
}

与单独使用 ReadAsync 相比，更推荐这种双循环模式，因为它会先读取所有可用项，再次等待，从而减少异步状态机的开销。

单个消费者——ReadAsync（更简单）

对于可以接受每项开销的简单场景：

try
{
    while (true)
    {
        var item = await reader.ReadAsync(cancellationToken);
        await ProcessAsync(item, cancellationToken);
    }
}
catch (ChannelClosedException)
{
    // Writer called Complete() -- no more items
}

多个消费者（扇出）

通过运行多个消费者任务来扩展处理能力。通道可确保每个项只由一个消费者读取。

public sealed class ScaledChannelProcessor(
    ChannelReader<WorkItem> reader,
    IServiceScopeFactory scopeFactory,
    ILogger<ScaledChannelProcessor> 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<IWorkItemHandler>();
                    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);
    }
}

---

与 IAsyncEnumerable 集成

ChannelReader<T>.ReadAllAsync() 会返回 IAsyncEnumerable<T>，支持使用 await foreach 消费，并与 LINQ 异步运算符集成。

基本的 await foreach

await foreach (var item in reader.ReadAllAsync(cancellationToken))
{
    await ProcessAsync(item, cancellationToken);
}
// Loop exits when writer calls Complete() and all items are consumed

ReadAllAsync 是最简单的消费模式。它会在内部处理 WaitToReadAsync/TryRead，并在通道关闭时完成。

从 API 端点流式传输

通道可以自然地与 ASP.NET Core 流式响应结合使用。直接返回 IAsyncEnumerable<T>——最小 API 会将各项作为 JSON 数组元素进行流式传输：

app.MapGet("/api/events/stream", (
    ChannelReader<ServerEvent> reader,
    CancellationToken ct) => reader.ReadAllAsync(ct));

LINQ 异步运算符

使用 System.Linq.Async NuGet 包，通道流可以与熟悉的 LINQ 运算符组合使用：

// 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);
}

从通道生成 IAsyncEnumerable

async IAsyncEnumerable<PriceUpdate> StreamPricesAsync(
    string symbol,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    var channel = Channel.CreateUnbounded<PriceUpdate>();

    // 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;
    }
}

---

性能

SingleReader / SingleWriter 标志

在通道选项中设置 SingleReader = true 或 SingleWriter = true 可启用无锁优化。通道会信任这些提示——违反提示（例如在 SingleReader = true 时使用多个并发读取器）会导致数据损坏。

// Optimal for single-producer, single-consumer pipeline
var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(1000)
{
    SingleReader = true,   // One consumer task
    SingleWriter = true,   // One producer task
    FullMode = BoundedChannelFullMode.Wait
});

WaitToReadAsync + TryRead 模式

这是最高效的使用者模式。WaitToReadAsync 会挂起，直到数据可用；随后 TryRead 会同步清空所有缓冲项，从而避免每个项都产生异步状态机开销。

while (await reader.WaitToReadAsync(ct))
{
    // Drain all currently buffered items synchronously
    while (reader.TryRead(out var item))
    {
        Process(item);
    }
}

TryWrite 快速路径

当通道有空余空间时，TryWrite 是同步且无需分配内存的。在可以处理 false 返回值的高频路径中，优先使用它，而不是 WriteAsync。

// Hot path -- avoid async overhead when channel has space
if (!writer.TryWrite(item))
{
    // Slow path -- wait for space (or handle overflow)
    await writer.WriteAsync(item, ct);
}

有界通道的内存行为

有界通道会预先分配一个包含 capacity 个槽位的内部数组。项按引用存储（对于引用类型），因此通道会一直持有这些引用，直到项被消费。对于对内存敏感的工作负载：

• 根据预期的单项大小乘以数量来选择容量
• TryRead/ReadAsync 返回项后，这些项即可进行 GC
• 丢弃模式（DropOldest、DropNewest）可以保持内存稳定，但会丢失数据

---

取消与优雅关闭

基本取消

将 CancellationToken 传递给所有异步通道操作。取消时，操作会抛出 OperationCanceledException。

try
{
    await foreach (var item in reader.ReadAllAsync(stoppingToken))
    {
        await ProcessAsync(item, stoppingToken);
    }
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
    // Expected during shutdown
}

排空模式

完成写入器以指示不会再有新项到达，然后在停止前排空剩余项。这可以防止关闭期间丢失数据。

public sealed class DrainableProcessor(
    Channel<WorkItem> channel,
    IServiceScopeFactory scopeFactory,
    ILogger<DrainableProcessor> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var reader = channel.Reader;

        try
        {
            while (await reader.WaitToReadAsync(stoppingToken))
            {
                while (reader.TryRead(out var item))
                {
                    using var scope = scopeFactory.CreateScope();
                    var handler = scope.ServiceProvider
                        .GetRequiredService<IWorkItemHandler>();
                    await handler.HandleAsync(item, stoppingToken);
                }
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // Shutdown requested -- fall through to drain
        }

        // Signal producers to stop -- any concurrent WriteAsync will throw ChannelClosedException
        channel.Writer.TryComplete();

        // Drain remaining items with a deadline
        logger.LogInformation("Draining remaining work items");
        using var drainCts = new CancellationTokenSource(TimeSpan.FromSeconds(25));

        while (reader.TryRead(out var remaining))
        {
            try
            {
                using var scope = scopeFactory.CreateScope();
                var handler = scope.ServiceProvider
                    .GetRequiredService<IWorkItemHandler>();
                await handler.HandleAsync(remaining, drainCts.Token);
            }
            catch (Exception ex)
            {
                logger.LogWarning(ex, "Error during drain");
            }
        }

        logger.LogInformation("Drain complete");
    }
}

Host 关闭超时

默认的 Host 关闭超时时间为 30 秒。如果排空需要更多时间，请进行配置：

builder.Services.Configure<HostOptions>(options =>
{
    options.ShutdownTimeout = TimeSpan.FromSeconds(60);
});

---

Agent 注意事项

1. 不要在生产环境中使用无界通道而不进行速率控制——持续的生产者压力可能会耗尽内存。始终优先使用显式指定容量的有界通道。
2. 不要违反 SingleReader/SingleWriter 承诺——这些标志会启用无锁优化。在 SingleReader = true 时使用多个并发读取器会导致数据损坏，而不是抛出异常。
3. 不要忘记对写入器调用 Complete()——如果不完成，使用 ReadAllAsync() 或 WaitToReadAsync 的使用者会在最后一个项之后无限期等待。
4. 不要全局捕获 ChannelClosedException——它表示写入器调用了 Complete()，且可能传入了错误。只在 ReadAsync 调用附近捕获它；WaitToReadAsync/TryRead 循环会通过返回 false 来处理完成状态。
5. 不要在高频路径中使用 ReadAsync——优先采用 WaitToReadAsync + TryRead 模式，同步排空缓冲项并减少异步状态机分配。
6. 不要在 itemDropped 回调中阻塞——它会在写入器线程上同步运行。请保持回调快速（增加计数器、记录日志），或将繁重工作移出回调。

---

参考资料

• System.Threading.Channels 概述
• Channel API 参考
• BoundedChannelOptions
• ChannelReader.ReadAllAsync