跳至主要內容

Channel 与异步生产者消费者

SunnyFan大约 12 分钟约 3588 字

Channel 与异步生产者消费者

简介

System.Threading.Channels 提供高性能的异步生产者-消费者模式实现。相比 BlockingCollection 和 ConcurrentQueue，Channel 支持异步等待和背压控制，是数据流管道的理想选择。

Channel 是 .NET Core 2.1 引入的类型，专为异步场景设计。它的内部使用无锁数据结构（单生产者单消费者场景使用环形缓冲区，多生产者或多消费者场景使用混合锁），在不阻塞线程的前提下实现了高效的线程安全消息传递。

特点

Channel vs 其他队列方案

方案对比

// ==========================================
// 生产者-消费者方案对比
// ==========================================

// 1. ConcurrentQueue + 轮询 — 简单但浪费 CPU
var queue = new ConcurrentQueue<int>();
// 消费者需要不断轮询: while (queue.TryDequeue(out var item))
// 没有"等待数据"的机制，空转浪费 CPU

// 2. BlockingCollection — 同步阻塞
var blocking = new BlockingCollection<int>(boundedCapacity: 100);
// 消费者: blocking.GetConsumingEnumerable()
// 满时生产者阻塞，空时消费者阻塞
// 问题: 阻塞线程，不适合高并发异步场景

// 3. Channel<T> — 异步非阻塞（推荐）
var channel = Channel.CreateBounded<int>(100);
// 消费者: await foreach (var item in channel.Reader.ReadAllAsync())
// 满时生产者 await，空时消费者 await
// 不阻塞线程，最适合异步场景

// 4. System.Threading.Tasks.Dataflow — 功能最丰富
// 提供了 TransformBlock、ActionBlock、BatchBlock 等高级块
// 适合复杂的消息处理拓扑
// 但 API 复杂，学习成本高

// 选择建议：
// - 简单异步队列 → Channel<T>
// - 复杂数据流拓扑 → TPL Dataflow
// - 同步场景 → BlockingCollection<T>

实现

基础 Channel

using System.Threading.Channels;

// 无界通道 — 无限缓冲
var unbounded = Channel.CreateUnbounded<string>();

// 有界通道 — 限制容量（背压）
var bounded = Channel.CreateBounded<string>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait, // 满时等待
    SingleReader = true,   // 单消费者优化
    SingleWriter = false,  // 多生产者
});

// ==========================================
// BoundedChannelFullMode 详解
// ==========================================
// Wait — 等待直到有空间（默认）
// DropWrite — 丢弃当前写入的消息，不等待
// DropNewest — 丢弃队列中最新的消息，腾出空间

// DropWrite 示例 — 适合日志、监控等可丢弃场景
var logChannel = Channel.CreateBounded<LogEntry>(
    new BoundedChannelOptions(1000)
    {
        FullMode = BoundedChannelFullMode.DropWrite,
        SingleWriter = true,
    });

// DropNewest 示例 — 适合实时数据，新数据比旧数据更重要
var realtimeChannel = Channel.CreateBounded<SensorData>(
    new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.DropNewest,
    });

// 生产者
async Task ProduceAsync(ChannelWriter<string> writer, int count)
{
    for (int i = 0; i < count; i++)
    {
        await writer.WriteAsync($"消息-{i}");
        // 或 TryWrite（非阻塞，返回是否成功）
        // writer.TryWrite($"消息-{i}");
    }
    writer.Complete(); // 通知消费者：不会再写入
}

// 消费者
async Task ConsumeAsync(ChannelReader<string> reader)
{
    // 方式 1: ReadAllAsync（推荐）
    await foreach (var item in reader.ReadAllAsync())
        Console.WriteLine($"消费: {item}");

    // 方式 2: WaitToReadAsync + TryRead
    while (await reader.WaitToReadAsync())
        while (reader.TryRead(out var item))
            Console.WriteLine(item);
}

// 使用
var channel = Channel.CreateBounded<int>(1000);
var producer = ProduceAsync(channel.Writer, 100);
var consumer = ConsumeAsync(channel.Reader);
await Task.WhenAll(producer, consumer);

Channel 的内部实现原理

// ==========================================
// Channel 的内部结构
// ==========================================
// Channel<T> 内部维护：
// 1. 一个环形缓冲区（存储消息）
// 2. 等待队列（等待读取/写入的 Task）
// 3. 状态标志（是否完成、是否有异常）

// 单生产者单消费者 (SPSC) 场景：
// - 使用环形缓冲区 + volatile 读写指针
// - 完全无锁，性能极高
// - SingleReader = true + SingleWriter = true 时启用

// 多生产者或多消费者 (MPSC/SPMC/MPMC) 场景：
// - 使用混合锁（自旋 + 内部锁）
// - 比无锁慢，但仍比传统锁高效

// ==========================================
// ValueTask 的使用
// ==========================================
// Channel 的读写方法返回 ValueTask 而不是 Task
// 这意味着在"大部分操作立即完成"的场景下，
// 可以避免堆分配，显著减少 GC 压力

// TryRead/TryWrite 返回 bool，完全零分配
// WriteAsync/ReadAsync 返回 ValueTask，可能零分配

数据处理管道

// 多阶段管道: 读取 → 解析 → 处理 → 输出
public class DataPipeline<TInput, TOutput>
{
    private readonly Channel<TInput> _input = Channel.CreateBounded<TInput>(1000);
    private readonly Channel<TOutput> _output = Channel.CreateBounded<TOutput>(1000);

    public ChannelReader<TOutput> Output => _output.Reader;
    public ChannelWriter<TInput> Input => _input.Writer;

    public async Task RunAsync(
        Func<TInput, CancellationToken, Task<TOutput>> processor,
        int degreeOfParallelism,
        CancellationToken ct)
    {
        var workers = Enumerable.Range(0, degreeOfParallelism)
            .Select(_ => ProcessWorkerAsync(_input.Reader, _output.Writer, processor, ct))
            .ToArray();

        await Task.WhenAll(workers);
        _output.Writer.Complete();
    }

    private async Task ProcessWorkerAsync(
        ChannelReader<TInput> reader,
        ChannelWriter<TOutput> writer,
        Func<TInput, CancellationToken, Task<TOutput>> processor,
        CancellationToken ct)
    {
        await foreach (var item in reader.ReadAllAsync(ct))
        {
            try
            {
                var result = await processor(item, ct);
                await writer.WriteAsync(result, ct);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"处理失败: {ex.Message}");
            }
        }
    }
}

// 使用 — 并行处理订单
var pipeline = new DataPipeline<Order, ProcessedOrder>();
var workerTask = pipeline.RunAsync(async (order, ct) =>
{
    await Task.Delay(10, ct); // 模拟处理
    return new ProcessedOrder(order.Id, order.Total * 1.1m);
}, degreeOfParallelism: 4, CancellationToken.None);

// 写入输入
for (int i = 0; i < 100; i++)
    await pipeline.Input.WriteAsync(new Order(i, 100 * i));
pipeline.Input.Complete();

// 读取输出
await foreach (var result in pipeline.Output.ReadAllAsync())
    Console.WriteLine($"处理完成: {result.Id}");

多阶段管道 — Fan-Out / Fan-In 模式

// ==========================================
// 扇出-扇入模式 — 分发到多个 Worker 并行处理
// ==========================================
public static class PipelineBuilder
{
    /// <summary>
    /// 扇出: 将一个通道的数据分发到 N 个 Worker
    /// </summary>
    public static async Task FanOutAsync<T>(
        ChannelReader<T> source,
        int workerCount,
        Func<ChannelReader<T>, Task> workerFactory,
        CancellationToken ct)
    {
        // 注意：默认情况下一个消息只能被一个消费者读取
        // 如果需要广播（多个消费者都收到），需要使用 BroadcastChannel

        // 这里用"工作窃取"模式：所有 Worker 竞争同一个通道
        var workers = Enumerable.Range(0, workerCount)
            .Select(_ => workerFactory(source))
            .ToArray();

        await Task.WhenAll(workers);
    }

    /// <summary>
    /// 扇入: 将多个通道合并到一个通道
    /// </summary>
    public static async Task<ChannelReader<T>> FanInAsync<T>(
        IEnumerable<ChannelReader<T>> sources,
        CancellationToken ct)
    {
        var output = Channel.CreateBounded<T>(1000);
        var tasks = sources.Select(source => CopyAsync(source, output.Writer, ct));
        var completion = Task.WhenAll(tasks);

        // 当所有源完成后，通知输出通道
        _ = completion.ContinueWith(
            _ => output.Writer.TryComplete(),
            CancellationToken.None,
            TaskContinuationOptions.None,
            TaskScheduler.Default);

        return output.Reader;
    }

    private static async Task CopyAsync<T>(
        ChannelReader<T> source,
        ChannelWriter<T> destination,
        CancellationToken ct)
    {
        await foreach (var item in source.ReadAllAsync(ct))
            await destination.WriteAsync(item, ct);
    }
}

// ==========================================
// 广播通道 — 一条消息多个消费者都收到
// ==========================================
public class BroadcastChannel<T>
{
    private readonly List<Channel<T>> _outputs = new();
    private readonly Channel<T> _input;
    private int _subscriberCount;

    public BroadcastChannel(int bufferSize = 1000)
    {
        _input = Channel.CreateBounded<T>(bufferSize);
    }

    public ChannelWriter<T> Writer => _input.Writer;

    public ChannelReader<T> Subscribe()
    {
        var output = Channel.CreateBounded<T>(bufferSize: 1000);
        _outputs.Add(output);
        Interlocked.Increment(ref _subscriberCount);

        // 启动转发任务
        _ = ForwardAsync(_input.Reader, output.Writer);

        return output.Reader;
    }

    private async Task ForwardAsync(
        ChannelReader<T> source,
        ChannelWriter<T> destination)
    {
        try
        {
            await foreach (var item in source.ReadAllAsync())
            {
                if (!await destination.WaitToWriteAsync())
                    break;
                await destination.WriteAsync(item);
            }
        }
        finally
        {
            destination.TryComplete();
        }
    }
}

// 使用广播
var broadcast = new BroadcastChannel<string>();
var sub1 = broadcast.Subscribe();
var sub2 = broadcast.Subscribe();

await broadcast.Writer.WriteAsync("消息1");
// sub1 和 sub2 都能收到 "消息1"

限流消费

// 使用 Channel + SemaphoreSlim 实现限流
public class RateLimitedProcessor<T>
{
    private readonly Channel<T> _channel = Channel.CreateBounded<T>(100);

    public async Task EnqueueAsync(T item, CancellationToken ct)
        => await _channel.Writer.WriteAsync(item, ct);

    public async Task StartAsync(
        Func<T, CancellationToken, Task> handler,
        int maxConcurrency,
        CancellationToken ct)
    {
        var semaphore = new SemaphoreSlim(maxConcurrency);
        var tasks = new List<Task>();

        await foreach (var item in _channel.Reader.ReadAllAsync(ct))
        {
            await semaphore.WaitAsync(ct);
            tasks.Add(Task.Run(async () =>
            {
                try { await handler(item, ct); }
                finally { semaphore.Release(); }
            }, ct));
        }

        await Task.WhenAll(tasks);
    }
}

// 使用 — 限制最多 5 个并发 HTTP 请求
var processor = new RateLimitedProcessor<string>();
var startTask = processor.StartAsync(async (url, ct) =>
{
    using var client = new HttpClient();
    var content = await client.GetStringAsync(url, ct);
    Console.WriteLine($"完成: {url} ({content.Length} chars)");
}, maxConcurrency: 5, CancellationToken.None);

foreach (var url in urls)
    await processor.EnqueueAsync(url, CancellationToken.None);

_channel.Writer.Complete();
await startTask;

实际应用 — 后台任务队列

/// <summary>
/// 通用后台任务队列 — 基于 Channel 实现
/// </summary>
public class BackgroundTaskQueue
{
    private readonly Channel<Func<CancellationToken, ValueTask>> _queue;
    private readonly ILogger<BackgroundTaskQueue> _logger;

    public BackgroundTaskQueue(ILogger<BackgroundTaskQueue> logger, int capacity = 1000)
    {
        _logger = logger;
        _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(
            new BoundedChannelOptions(capacity)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = true,
            });
    }

    /// <summary>
    /// 入队一个后台任务
    /// </summary>
    public async ValueTask QueueBackgroundWorkItemAsync(
        Func<CancellationToken, ValueTask> workItem)
    {
        if (workItem == null) throw new ArgumentNullException(nameof(workItem));
        await _queue.Writer.WriteAsync(workItem);
    }

    /// <summary>
    /// 获取待处理的任务（Dequeuer 调用）
    /// </summary>
    public IAsyncEnumerable<Func<CancellationToken, ValueTask>> ReadAllAsync(
        CancellationToken ct)
    {
        return _queue.Reader.ReadAllAsync(ct);
    }

    /// <summary>
    /// 获取队列中待处理的任务数
    /// </summary>
    public int Count => _queue.Reader.Count;

    /// <summary>
    /// 标记队列完成（不再接受新任务）
    /// </summary>
    public void Complete() => _queue.Writer.TryComplete();
}

/// <summary>
/// 队列消费者 — 在 IHostedService 中运行
/// </summary>
public class QueuedHostedService : BackgroundService
{
    private readonly BackgroundTaskQueue _taskQueue;
    private readonly ILogger<QueuedHostedService> _logger;

    public QueuedHostedService(
        BackgroundTaskQueue taskQueue,
        ILogger<QueuedHostedService> logger)
    {
        _taskQueue = taskQueue;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("后台任务队列消费者已启动");

        await foreach (var workItem in _taskQueue.ReadAllAsync(stoppingToken))
        {
            try
            {
                await workItem(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "后台任务执行失败");
            }
        }

        _logger.LogInformation("后台任务队列消费者已停止");
    }
}

// 注册到 DI 容器
// services.AddSingleton<BackgroundTaskQueue>();
// services.AddHostedService<QueuedHostedService>();

// 使用
public class OrderService
{
    private readonly BackgroundTaskQueue _queue;

    public OrderService(BackgroundTaskQueue queue) => _queue = queue;

    public async Task CreateOrderAsync(Order order)
    {
        // 快速返回，将耗时操作放入后台队列
        await _queue.QueueBackgroundWorkItemAsync(async ct =>
        {
            await SendConfirmationEmailAsync(order, ct);
            await UpdateInventoryAsync(order, ct);
            await LogAnalyticsAsync(order, ct);
        });
    }
}

实际应用 — 批量处理

/// <summary>
/// 批量处理器 — 累积消息到一定数量或超时后批量处理
/// </summary>
public class BatchingProcessor<T>
{
    private readonly Channel<T> _input;
    private readonly TimeSpan _maxWaitTime;
    private readonly int _batchSize;
    private readonly Func<IReadOnlyList<T>, CancellationToken, Task> _batchHandler;
    private readonly CancellationTokenSource _cts = new();

    public BatchingProcessor(
        Func<IReadOnlyList<T>, CancellationToken, Task> batchHandler,
        int batchSize = 100,
        TimeSpan? maxWaitTime = null,
        int channelCapacity = 10000)
    {
        _batchHandler = batchHandler;
        _batchSize = batchSize;
        _maxWaitTime = maxWaitTime ?? TimeSpan.FromSeconds(5);
        _input = Channel.CreateBounded<T>(channelCapacity);
    }

    public ChannelWriter<T> Writer => _input.Writer;

    public async Task StartAsync(CancellationToken ct = default)
    {
        var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ct, _cts.Token);
        var batch = new List<T>(_batchSize);
        var timerCts = new CancellationTokenSource();

        try
        {
            await foreach (var item in _input.Reader.ReadAllAsync(linkedCts.Token))
            {
                batch.Add(item);

                // 重置超时计时器
                timerCts.Cancel();
                timerCts = new CancellationTokenSource();

                if (batch.Count >= _batchSize)
                {
                    await ProcessBatchAsync(batch, linkedCts.Token);
                    batch.Clear();
                }
                else if (batch.Count == 1)
                {
                    // 第一个元素到达时启动超时计时器
                    _ = TimeoutProcessAsync(batch, timerCts.Token, linkedCts.Token);
                }
            }

            // 处理剩余的元素
            if (batch.Count > 0)
                await ProcessBatchAsync(batch, linkedCts.Token);
        }
        finally
        {
            timerCts.Dispose();
        }
    }

    private async Task TimeoutProcessAsync(
        List<T> batch,
        CancellationToken timerCt,
        CancellationToken linkedCt)
    {
        try
        {
            await Task.Delay(_maxWaitTime, timerCt);
        }
        catch (OperationCanceledException)
        {
            return; // 计时器被取消（批量已处理）
        }

        // 超时，处理当前批次
        if (batch.Count > 0)
        {
            var itemsToProcess = batch.ToList();
            batch.Clear();
            await ProcessBatchAsync(itemsToProcess, linkedCt);
        }
    }

    private async Task ProcessBatchAsync(List<T> batch, CancellationToken ct)
    {
        try
        {
            await _batchHandler(batch, ct);
        }
        catch (Exception ex)
        {
            Console.Error.WriteLine($"批量处理失败: {ex.Message}");
        }
    }

    public void Stop() => _cts.Cancel();
}

// 使用 — 批量写入数据库
var batcher = new BatchingProcessor<Order>(
    batchHandler: async (orders, ct) =>
    {
        await _dbContext.Orders.AddRangeAsync(orders, ct);
        await _dbContext.SaveChangesAsync(ct);
        Console.WriteLine($"批量写入 {orders.Count} 条订单");
    },
    batchSize: 50,
    maxWaitTime: TimeSpan.FromSeconds(3));

// 后台启动
_ = batcher.StartAsync();

// 写入数据
foreach (var order in orders)
    await batcher.Writer.WriteAsync(order);

batcher.Writer.Complete();

优点

缺点

总结

Channel<T> 是 .NET 中实现异步生产者-消费者模式的首选。无界通道适合不限速场景，有界通道提供背压控制。ReadAllAsync 简化消费逻辑。多阶段管道通过多个 Channel 串联实现。建议在日志处理、消息队列本地缓冲、数据 ETL 等场景使用 Channel 替代 BlockingCollection。

选择指南：

  1. 简单队列 — Channel.CreateBounded 直接使用
  2. 并行处理 — Fan-Out 模式，多个 Worker 消费同一 Channel
  3. 数据管道 — 多阶段 Channel 串联
  4. 广播 — 自行实现 BroadcastChannel
  5. 批量处理 — Channel + 超时触发器
  6. 后台任务 — BackgroundTaskQueue + IHostedService

关键知识点

  • 先明确这个主题影响的是语法层、运行时层，还是性能与可维护性层。
  • 学习时要同时关注语言表面写法和编译器、JIT、GC 等底层行为。
  • 真正有价值的是知道"为什么这样写"和"在什么边界下不能这样写"。

项目落地视角

  • 把示例改成最小可运行样例，并观察编译输出、运行结果和异常行为。
  • 如果它会进入团队代码规范，最好同步补充命名约定、禁用场景和替代方案。
  • 涉及性能结论时，优先用 Benchmark 或实际热点链路验证，而不是凭感觉判断。

常见误区

  • 只记语法糖，不知道底层成本。
  • 把适用于小样例的写法直接搬到高并发或大对象场景里。
  • 忽略框架版本、语言版本和运行时差异，导致结论失真。
  • 使用无界通道导致内存溢出（生产速度大于消费速度时）。
  • 忘记调用 Writer.Complete() 导致消费者永远等待。
  • 消费者异常未捕获导致管道中断。
  • SingleReader/SingleWriter 设置不当导致性能未充分优化。

进阶路线

  • 继续向源码、IL、JIT 行为和 BCL 实现层深入。
  • 把知识点和代码评审、性能诊断、面试复盘结合起来。
  • 把同类主题做横向对比，例如值类型与引用类型、迭代器与 async 状态机、反射与 Source Generator。
  • 学习 System.Threading.Tasks.Dataflow 的高级数据流拓扑。
  • 研究 RabbitMQ/Kafka 等专业消息队列的本地缓冲实现。

适用场景

  • 当你准备把《Channel 与异步生产者消费者》真正落到项目里时，最适合先在一个独立模块或最小样例里验证关键路径。
  • 适合在需要理解语言特性、运行时行为或 API 边界时阅读。
  • 当代码开始出现性能瓶颈、可维护性问题或语义歧义时，这类主题会直接影响实现质量。

落地建议

  • 先写最小可运行样例，再把结论迁移到真实业务代码。
  • 同时记录这个特性的收益、限制和替代方案，避免为了"高级"而使用。
  • 涉及内存、并发或序列化时，最好配合调试器或基准测试验证。

排错清单

  • 先确认问题属于编译期、运行期还是语义误用。
  • 检查是否存在隐式转换、装箱拆箱、闭包捕获或上下文切换等隐藏成本。
  • 查看异常栈、日志和最小复现代码，优先排除使用姿势问题。
  • 检查 Writer.Complete() 是否被正确调用。
  • 检查有界通道是否因满而导致生产者无限等待。
  • 监控通道的 Reader.Count 确认消息消费进度。

复盘问题

  • 如果把《Channel 与异步生产者消费者》放进你的当前项目，最先要验证的输入、输出和失败路径分别是什么？
  • 《Channel 与异步生产者消费者》最容易在什么规模、什么边界条件下暴露问题？你会用什么指标或日志去确认？
  • 相比默认实现或替代方案，采用《Channel 与异步生产者消费者》最大的收益和代价分别是什么？

延伸阅读

Copyright © 2026 SunnyFan