C# 有界 Channel：把背压、故障传播与优雅停止设计成一个协议

一个采集程序每秒收到 800 条测量数据，计算模块每秒只能处理 500 条。把线程安全队列换成 Channel，并不会使缺少的每秒 300 次计算凭空出现。真正要决定的是：积压放在哪里，最多允许等待多久，哪些数据允许丢，以及退出时如何解释已经接收的数据。

先算容量，再选 API

设平均到达率为 λ，持续处理能力为 μ。如果 λ 长期大于 μ，任何有限内存队列都会满；无界队列只是把“拒绝”变成更晚发生的内存耗尽。

假设单条数据连同对象开销按 4 KiB 估算，容量 2,000 条约占 7.8 MiB，这还没有算正在处理的项目、生产者等待写入时持有的对象，以及下游缓存。更关键的是，在处理能力为每秒 500 条时，队尾项目仅排队就可能等待约 4 秒。对实时监测界面，这种“没有丢数据”可能意味着一直显示过期状态。

容量应同时满足内存预算和可接受等待时间。粗略约束可以写成 B ≤ μ × W，其中 B 是容量，W 是允许的队列等待时间。它只适合做第一轮设计；突发输入、处理时间长尾和多消费者竞争仍需实测。

还要检查入口：如果上游先为一百万条数据各创建一个 Task，再在 Task 内等待 WriteAsync，有界 Channel 并没有约束这一百万个任务。背压必须沿调用链一直传回真正的数据源，或在入口明确拒绝任务。

满了以后，业务允许发生什么

Channel 的 Wait 模式会让异步写入等待空间；DropOldest、DropNewest、DropWrite 则是不同的丢弃策略。它们是数据语义，不只是性能开关。Microsoft 的 Channel 文档说明了各模式及读写 API。

测量原始记录如果要作为质量追溯依据，通常不能默默丢弃；可以等待上游、写入持久化暂存区，或显式记录采集失败。预览画面只关心最新状态时，丢掉旧帧反而合理。订单、付款等必须可靠交付的任务不能只靠进程内 Channel：进程崩溃后，队列和正在处理的项目都会消失。

尤其不要把“WriteAsync 正常完成”解释为“下游已经成功保存”。在 Wait 模式，它最多表示项目进入通道；在丢弃模式，写入调用成功也不等于每项都被处理。业务确认应发生在真正的提交点。

三个状态需要不同的取消信号

优雅退出至少包含三个阶段：

1. 停止入口，不再接受新的项目。
2. 标记写端完成，让消费者排空已经接收的项目。
3. 排空超时或关键组件失败时，强制中止仍在执行的工作。

如果生产者和消费者共用一个“用户点停止”令牌，用户停止采集时消费者也会立即退出，剩余队列就无法排空。反过来，如果消费者异常后没有通知生产者，生产者可能永久等待一个再也无人释放的队列空间。

下面把“停止生产”和“强制中止”分开。环境为 .NET 8、C# 12 控制台项目；核心类已在 SDK 8.0.421 的独立临时项目编译，并用内存数据验证正常排空、停止生产、消费者故障和强制中止四条路径。项目的 HandleAsync 必须合作响应取消，才能遵守强制停止预算。

using System.Threading.Channels;

public static class BoundedPipeline
{
    public static async Task RunAsync<T>(
        IAsyncEnumerable<T> source,
        Func<T, CancellationToken, Task> handle,
        CancellationToken stopProducing,
        CancellationToken abort)
    {
        var queue = Channel.CreateBounded<T>(
            new BoundedChannelOptions(256)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleWriter = true,
                SingleReader = true,
                AllowSynchronousContinuations = false
            });

        using var failed = new CancellationTokenSource();
        using var producerStop = CancellationTokenSource
            .CreateLinkedTokenSource(stopProducing, abort, failed.Token);
        using var consumerStop = CancellationTokenSource
            .CreateLinkedTokenSource(abort, failed.Token);

        async Task ProduceAsync()
        {
            Exception? error = null;
            try
            {
                await foreach (var item in source
                    .WithCancellation(producerStop.Token))
                {
                    await queue.Writer.WriteAsync(item, producerStop.Token);
                }
            }
            catch (OperationCanceledException)
                when (producerStop.IsCancellationRequested) { }
            catch (Exception ex)
            {
                error = ex;
                failed.Cancel();
                throw;
            }
            finally
            {
                queue.Writer.TryComplete(error);
            }
        }

        async Task ConsumeAsync()
        {
            try
            {
                await foreach (var item in queue.Reader
                    .ReadAllAsync(consumerStop.Token))
                {
                    await handle(item, consumerStop.Token);
                }
            }
            catch
            {
                failed.Cancel();
                throw;
            }
        }

        var consumer = ConsumeAsync();
        var producer = ProduceAsync();
        await Task.WhenAll(producer, consumer);
    }
}

这里采用“关键故障立即中止”策略：生产或处理失败，会取消另一侧，而不是继续处理剩余数据。正常停止入口则只取消生产，写端完成后消费者继续排空。强制中止会使任务结束为取消或故障，不能把它汇报成全部成功。

调用方可以在收到停止请求后启动排空倒计时，超过例如 10 秒再取消 abort。这个 10 秒应来自服务关闭窗口和数据重要性，不能直接当作通用常数。若下游驱动无视取消，取消令牌也无法杀掉它，必须额外设计 I/O 超时或进程隔离。

单读单写、顺序与资源所有权

SingleReader、SingleWriter 是对使用方式的承诺，不会替你把多个调用者串行化。AllowSynchronousContinuations 为 true 时，生产线程可能承担消费者的延续工作；把它设为 false 更容易隔离入口延迟。微软对 Channel 的设计介绍解释了这些选项的含义。

增加消费者后，读取顺序和完成顺序是两件事。第一个项目处理 300 毫秒，第二个只用 10 毫秒，最终写出的结果可能反过来。如果后续要求按采样序号提交，可以采用单消费者、按设备分区，或添加带容量上限的重排缓冲。重排缓冲如果无限等待缺失序号，又会成为新的无界队列。

若项目携带租用数组或非托管缓冲区，必须明确所有权：成功入队后由消费者释放，入队取消前由生产者释放；强制停止时，残留队列也需要逐项清理。丢弃策略还需要处理被丢项目的释放。最小示例故意使用普通值或托管对象，没有替读者隐藏这部分成本。

从排队时间反推“系统还能撑多久”

在前面的每秒 800 条输入、500 条输出场景里，空的 2,000 条队列约在 6.7 秒后填满。这个估算忽略启动和抖动，但足以说明：把容量翻倍只多买到几秒，并没有修复持续超载。若峰值只持续 2 秒，积压约 600 条，恢复到每秒 200 条输入时，每秒可净排空 300 条，约 2 秒后回到正常。这种突发模型才适合用缓冲吸收。

还应区分“写入等待时间”和“项目在队列中的年龄”。第一项反映入口受压程度，第二项反映结果新鲜度。生产者等待很短也可能是因为数据被丢弃，而不是系统处理得快；平均排队时间很短也可能掩盖少量已经等待几十秒的项目。

对每个项目携带单调递增的序号和入队时刻，可以在提交点计算年龄，并对过期项目作业务判断。预览状态可以丢弃已经无意义的旧值，审计记录则应保留并标记延迟。两者不能共用一个没有语义区分的丢弃策略。

当一个项目触发外部写入时，取消发生在请求发送之后，外部动作可能已经执行。即使通道内没有重复读取，也不能推出外部系统没有重复效果。若任务允许恢复重放，仍需为外部提交设计稳定操作 ID 或可比较版本；队列只约束进程内的传递，不能扩展为端到端事务。

用故障注入验证，而不是只测吞吐

第一轮用容量 4、每项延迟 100 毫秒、连续生产 100 个整数。记录入口等待、入队、开始处理和成功提交四个时间点，确认队列满后生产速度被压住，且已接收序号最终恰好出现一次。

第二轮处理到第 7 项时抛异常。生产任务应尽快退出，整体任务必须以故障结束，不能停在 WriteAsync。第三轮在队列满时只取消 stopProducing，确认已接收项目能够处理完；再把 HandleAsync 延迟改长，触发 abort，确认未提交项目数被明确记录。

最后用真实大小的负载观察进程工作集、分配率、队列等待时间的 P95，以及停止耗时。队列长度低不代表健康：消费者可能刚取出一个项目，然后在下游挂住。至少同时记录处理中数量和最长处理时长。

上述独立验证不包含真实采集设备、外部提交或性能基准，本文没有给出吞吐提升百分比。可复用的结论是：容量、满载策略、确认时机和停止协议需要共同设计，Channel 只是承载这个协议的工具。