ValueTask / IAsyncEnumerable 的吞吐与背压：Channel 协作、取消/超时与反模式清单 🚀

---

---

1) 背景与目标🩺

痛点：高频异步 I/O、突发“爆冲”、无上限队列顶爆内存、取消边界不清导致句柄/缓冲泄漏。
目标：

• 在常同步完成的热点路径用 ValueTask 减少分配/调度（而不是全局替代 Task）
• 以 Channel<T> 建立有界缓冲与明确 FullMode（Wait/DropNewest/DropOldest/DropWrite）
• 用 IAsyncEnumerable<T> 的“拉式”消费，清晰传递 CancellationToken 并优雅收尾

🗺️ 总览：数据从哪里来，到哪里去？

---

2) 核心概念与选型 🧠

• ValueTask/ValueTask<T>：仅在高概率同步完成的热路径考虑；同一 ValueTask 不可重复 await，若需多次等待请 .AsTask()。默认仍优先 Task，ValueTask 是“进阶性能选项”。
• IAsyncEnumerable<T>：MoveNextAsync() 返回 ValueTask<bool>；取消可用 WithCancellation(ct) 或 GetAsyncEnumerator(ct)（[EnumeratorCancellation] 绑定外部 token）。
• Channel<T>：BoundedChannelOptions 指定容量与 FullMode；ReadAllAsync(ct) 将读端暴露为异步枚举。

✅ 结论：热路径同步概率高→可以用 ValueTask；流式 + 背压→Channel + IAsyncEnumerable 稳。

---

3) 可跑基线：有界 Channel + 多消费者 ⚙️

环境：.NET 8/9
创建：dotnet new console -n vtask-asyncenum-throughput

Program.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading.Channels;

// ===== 参数解析 =====
string GetArg(string key, string def)
{
    var argv = Environment.GetCommandLineArgs();
    var idx = Array.IndexOf(argv, key);
    return (idx >= 0 && idx + 1 < argv.Length) ? argv[idx + 1] : def;
}
int GetIntArg(string key, int def) => int.TryParse(GetArg(key, def.ToString()), out var v) ? v : def;
bool GetBoolArg(string key, bool def) => bool.TryParse(GetArg(key, def.ToString()), out var v) ? v : def;
BoundedChannelFullMode ParseMode(string s) => s.ToLowerInvariant() switch
{
    "wait" => BoundedChannelFullMode.Wait,
    "dropoldest" => BoundedChannelFullMode.DropOldest,
    "dropnewest" => BoundedChannelFullMode.DropNewest,
    "dropwrite" => BoundedChannelFullMode.DropWrite,
    _ => BoundedChannelFullMode.Wait
};

// ===== 运行参数（可通过命令行覆盖） =====
int capacity  = GetIntArg("--capacity", 50_000);
int consumers = GetIntArg("--consumers", Math.Max(Environment.ProcessorCount, 2));
int rps       = GetIntArg("--rps", 0); // 0 表示“尽力跑”（不做节流）
var fullMode  = ParseMode(GetArg("--mode", "Wait"));
bool syncCont = GetBoolArg("--synccont", true);

// ===== 监控计数器（业务路径仅做 Interlocked 累加）=====
static class Metrics
{
    public static long Sent;
    public static long Dropped;
    public static long Recv;
    public static long Enqueued;
    public static long Dequeued;
}

var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

// ===== 通道（同步延续可配，建议做 A/B 对比） =====
var ch = Channel.CreateBounded<MyItem>(new BoundedChannelOptions(capacity)
{
    SingleWriter = true,      // 单生产者
    SingleReader = false,     // 多消费者
    FullMode = fullMode,
    AllowSynchronousContinuations = syncCont
});

// ===== 背景监控任务（1s 打印一次，不阻塞业务路径）=====
_ = Task.Run(async () =>
{
    try
    {
        while (!cts.IsCancellationRequested)
        {
            await Task.Delay(1000, cts.Token);
            var sent     = Interlocked.Exchange(ref Metrics.Sent, 0);
            var dropped  = Interlocked.Exchange(ref Metrics.Dropped, 0);
            var recv     = Interlocked.Exchange(ref Metrics.Recv, 0);
            var enq      = Interlocked.Read(ref Metrics.Enqueued);
            var deq      = Interlocked.Read(ref Metrics.Dequeued);
            var depthEst = Math.Max(0, enq - deq); // 估算队列深度（近似）

            Console.WriteLine($"[Stats] sent={sent:N0} recv={recv:N0} dropped={dropped:N0} depth~={depthEst:N0}");
        }
    }
    catch (OperationCanceledException) { }
});

// ===== 启动生产者与消费者 =====
var producerTask  = Task.Run(() => Producer(ch.Writer, fullMode, rps, cts.Token));
var consumerTasks = Enumerable.Range(0, consumers)
    .Select(i => Task.Run(() => Consumer(i, ch.Reader, cts.Token)))
    .ToArray();

Console.WriteLine($"Running... --capacity {capacity} --consumers {consumers} --mode {fullMode} --rps {rps} --synccont {syncCont}");
Console.WriteLine("Press Ctrl+C to stop.");

try
{
    await Task.Delay(Timeout.Infinite, cts.Token);
}
catch (OperationCanceledException)
{
    // 正常停止信号
}
finally
{
    ch.Writer.TryComplete();
    try { await Task.WhenAll(consumerTasks.Append(producerTask)); }
    catch (OperationCanceledException) { }
    catch (AggregateException ae) when (ae.InnerExceptions.All(e =>
        e is OperationCanceledException || e is ChannelClosedException)) { }
}

// ===== 生产者：Wait 用 WriteAsync 背压；Drop* 用 TryWrite 计丢弃 =====
static async Task Producer(ChannelWriter<MyItem> writer,
                           BoundedChannelFullMode mode,
                           int rps,
                           CancellationToken ct)
{
    // rps>0 使用 PeriodicTimer 做平滑节流；否则尽力跑
    using PeriodicTimer? timer = (rps > 0) ? new PeriodicTimer(TimeSpan.FromMilliseconds(1)) : null;
    double perMs = (rps > 0 ? rps / 1000.0 : 0.0);
    double carry = 0.0;

    try
    {
        while (!ct.IsCancellationRequested)
        {
            int countThisMs;
            if (perMs <= 0)
            {
                // 不节流：默认每 1ms 产 200 条（可调）
                countThisMs = 200;
            }
            else
            {
                carry += perMs;
                countThisMs = (int)Math.Floor(carry);
                carry -= countThisMs;
                if (countThisMs <= 0) countThisMs = 1; // 保底
            }

            for (int i = 0; i < countThisMs; i++)
            {
                var item = new MyItem(Stopwatch.GetTimestamp(), Random.Shared.Next());

                if (mode == BoundedChannelFullMode.Wait)
                {
                    await writer.WriteAsync(item, ct);           // 可能抛 OCE
                    Interlocked.Increment(ref Metrics.Sent);
                    Interlocked.Increment(ref Metrics.Enqueued);
                }
                else
                {
                    if (writer.TryWrite(item))
                    {
                        Interlocked.Increment(ref Metrics.Sent);
                        Interlocked.Increment(ref Metrics.Enqueued);
                    }
                    else
                    {
                        Interlocked.Increment(ref Metrics.Dropped);
                    }
                }
            }

            if (timer is not null)
                await timer.WaitForNextTickAsync(ct);
        }
    }
    catch (OperationCanceledException) { /* 正常停止 */ }
    catch (ChannelClosedException)     { /* 通道完成，正常停止 */ }
}

// ===== 消费者：ReadAllAsync + 协作式取消 =====
static async Task Consumer(int id, ChannelReader<MyItem> reader, CancellationToken ct)
{
    try
    {
        await foreach (var item in reader.ReadAllAsync(ct))
        {
            // 模拟 30% I/O（真实场景可替换为网络/磁盘）
            if ((item.Value % 10) < 3) await Task.Delay(1, ct);

            await HandleAsync(item, ct); // 常同步完成 → ValueTask.CompletedTask
            Interlocked.Increment(ref Metrics.Recv);
            Interlocked.Increment(ref Metrics.Dequeued);
        }
    }
    catch (OperationCanceledException) { /* 正常停止 */ }
    catch (ChannelClosedException)     { /* 正常停止 */ }
}

static ValueTask HandleAsync(MyItem item, CancellationToken ct)
    => ValueTask.CompletedTask;

readonly record struct MyItem(long Ticks, int Value);

🧭 背压/丢弃策略是怎么决定的？

---

4) ValueTask：何时真香，何时别用 🧩

适用：缓存命中、连接/对象池已就绪、内存计算等常同步完成的热点方法。
边界：同一 ValueTask 不可重复 await、不应跨层长期持有；若需多次等待请 .AsTask()。
默认：除非确认是“热点+常同步”，否则保持 Task 简洁度更高。

示例：同步命中快速返回（并发安全容器）

using System.Collections.Concurrent;

static readonly ConcurrentDictionary<string, User> Cache = new();

static ValueTask<User?> GetUserAsync(string id, CancellationToken ct)
{
    if (Cache.TryGetValue(id, out var u))
        return new(u);                    // 同步路径：零分配
    return new(FetchFromDbAsync(id, ct)); // 异步路径：包 Task
}

---

5) IAsyncEnumerable 的取消/超时边界 ⏱️

• await foreach (var x in source.WithCancellation(ct)) 把 token 传给枚举器；也可 GetAsyncEnumerator(ct)，配合 [EnumeratorCancellation] 从方法签名绑定 token。
• 协作式取消：不要假设“取消=立刻硬停”。若通道中已有就绪项，枚举器可能先把它们吐出后才抛 OperationCanceledException。实践中需要在写端使用 TryComplete() 或写入侧超时来收束流动。

🔄 取消时序

---

6) 多阶段流水线与批处理 🧱

• Stage1（接入）→ Stage2（分拣/聚合）→ Stage3（落盘/发送）
• 批处理：拉满 M 条或等待 T ms，孰先触发，减少 syscalls/事务成本
• 内存：配合 ArrayPool<T>/MemoryPool<byte> 承载批次缓冲，用后归还（降低 LOH 压力）

批处理消费者

using System.Buffers;
using System.Diagnostics;
using System.Threading.Channels;

static async Task BatchConsumer(ChannelReader<MyItem> reader,
                                int batchSize, TimeSpan maxWait,
                                CancellationToken ct)
{
    var pool = ArrayPool<MyItem>.Shared;
    var buf  = pool.Rent(batchSize);
    try
    {
        while (!ct.IsCancellationRequested)
        {
            int count = 0;
            var sw = Stopwatch.StartNew();

            if (!await reader.WaitToReadAsync(ct)) break;
            if (reader.TryRead(out var tmp1)) { buf[count] = tmp1; count++; }

            while (count < batchSize && sw.Elapsed < maxWait && reader.TryRead(out var tmp2))
            { buf[count] = tmp2; count++; }

            await FlushAsync(buf.AsSpan(0, count), ct);
        }
    }
    catch (OperationCanceledException) { }
    catch (ChannelClosedException)     { }
    finally { pool.Return(buf, clearArray: true); }
}

static Task FlushAsync(ReadOnlySpan<MyItem> batch, CancellationToken ct)
    => Task.CompletedTask;

🧭 三阶段流水线结构

---

7) 实验：压力场景与指标 🧪

场景

1. 恒定产出：1k/5k/10k RPS，对比 Wait 与 DropOldest 的吞吐/延迟/丢弃率
2. 突发洪峰：1 秒 10× 峰值，观察队列深度与丢弃曲线
3. 取消/超时：消费者刻意减速，验证“先吐已就绪数据再抛 OCE”的协作式取消，以及写端 TryComplete 的收敛速度
4. 混合 IO/CPU：30% 模拟 I/O，评估线程与上下文切换

指标

• 吞吐：in_rps / out_rps
• 延迟：端到端 p50/p95/p99（Stopwatch + 直方统计）
• 队列深度（估算）、丢弃率、GC 分配（B/s）、CPU（System.Runtime 计数器）

dotnet-counters

# 从一开始就监控 System.Runtime 计数器运行你的程序
dotnet-counters monitor System.Runtime -- \
  dotnet run -c Release --project src/App -- --capacity 50000 --consumers 8 --mode Wait --rps 50000 --synccont true

---

8) 反模式清单 ⛔

• 滥用 ValueTask：内部总是异步 await 还强行返回 ValueTask → 无收益涨复杂
• 重复 await 同一个 ValueTask / 长期持有：需要多次等待请 .AsTask()
• IAsyncEnumerable 一把梭 ToListAsync()：失去流式/背压，内存暴涨
• 在 await foreach 内做阻塞 I/O（Result/Wait/ReadAllBytes）：线程池耗尽
• 无上限 Channel：峰值“靠内存硬抗” → OOM
• 误解取消：以为 ReadAllAsync(ct) 会“立刻硬停”，而忽略协作式语义

---

9) 可观测与告警基线 🛰️

• 指标：in_rps / out_rps / lag_ms / queue_depth / dropped / alloc_bytes / gc_time / cpu_usage
• 告警：队列深度持续 > 80% 容量；丢弃率 > 1%；p95 延迟超 SLO；连续取消失败
• Trace：为 enqueue/dequeue/batch-flush 打 Activity，用 Baggage 传 flow-id，便于跨阶段关联

---

10) 代码结构建议 🗂️

vtask-asyncenum-throughput/
  src/
    App/                          # Program.cs（基线管线，含参数化/监控）
    Pipelines/
      BoundedPipe.cs              # 包装 Channel：Wait/DropOldest/DropWrite 策略
      BatchConsumer.cs            # M/T 触发的批处理消费者
      RateProducer.cs             # 恒定/洪峰产出器（可替换 Program 内部实现）
    Metrics/
      Hist.cs                     # 近似 HDR 直方图
      Counters.cs                 # EventSource + EventCounters（可选）
  scripts/
    run.ps1                       # 一键压测：构建→运行→dotnet-counters 监控
  README.md                       # 怎么跑、怎么看、期望曲线

---