17370845950

用 Task + Channel 替代 goroutine + channel

的核心思路是：以 async/await 模拟 goroutine 的异步并发，用线程安全、真异步的 Channel 替代 channel；避免 BlockingCollection 和盲目 Task.Run，正确处理关闭、取消与资源清理。

用 Task + Channel 替代 goroutine + channel 的核心思路

Go 的 goroutine 本质是轻量级协作式并发单元，C# 没有语言级等价物，但 Task（配合 async/await）在绝大多数 I/O 密集场景下能达到相似效果：高并发、低栈开销、自动调度。而 Go 的 channel 在 .NET 6+ 中已有原生替代 —— System.Threading.Channels.Channel，它线程安全、支持异步读写、可配置缓冲策略，是目前最贴近的实现。

Channel 的创建与基础用法区别于 BlockingCollection

很多人第一反应是用 BlockingCollection，但它基于锁 + 阻塞，不支持真正的异步等待（TakeAsync 是伪异步），且无法取消；而 Channel 的 Writer.WriteAsync() 和 Reader.ReadAsync() 是真异步、可取消、无锁（单生产者/单消费者模式下）。

• 创建无界 channel：

  var channel = Channel.CreateUnbounded();

• 创建带容量限制的 channel：

  var channel = Channel.CreateBounded(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.Wait });

• 写入必须检查是否完成：

  await channel.Writer.WriteAsync("hello"); // 不要忽略返回值

• 读取需处理关闭信号：

  while (await channel.Reader.WaitToReadAsync()) { if (channel.Reader.TryRead(out var msg)) { /* 处理 msg */ } }

模拟 goroutine 启动：别直接 Task.Run，优先用 async + await

Go 的 go fn() 是隐式启动，C# 若盲目套用 Task.Run(() => { ... })，会把本该异步的 I/O 操作强行拉到线程池，浪费资源，还可能引发死锁（尤其在 UI 或 ASP.NET 同步上下文里）。正确做法是让工作函数本身是 async Task，再用 Task.Run 包裹仅当它含 CPU 密集逻辑时。

• I/O 密集型（如 HTTP 请求、文件读取）：

  async Task WorkerAsync(ChannelReader reader) { while (await reader.WaitToReadAsync()) { if (reader.TryRead(out var msg)) { await SomeHttpCallAsync(msg); } } }

• CPU 密集型（如图像处理）才考虑 Task.Run：

  Task.Run(() => HeavyCompute(msg));

• 启动多个“goroutine”风格协程：

  _ = WorkerAsync(channel.Reader); _ = WorkerAsync(channel.Reader); // 注意：这里用 _ 忽略 Task 引用，实际应妥善管理生命周期

关闭 channel 和清理资源的常见漏点

Go 的 close(ch) 对应 C# 的 channel.Writer.Complete()，但容易被忽略的是：一旦调用 Complete()，后续所有 WriteAsync 都会抛 InvalidOperationException；且 Reader 不会自动退出循环，必须靠 WaitToReadAsync() 返回 false 才知道 channel 已关闭并写入完毕。

• 生产者结束前务必调用：

  channel.Writer.Complete();

• 消费者循环中必须检查 WaitToReadAsync() 返回值：

  while (await channel.Reader.WaitToReadAsync()) { /* ... */ } // 循环退出即表示 channel 关闭且无更多数据

• 若需等待所有消费者完成，不能只等 channel.Reader.Completion，而应单独跟踪 Task 实例并 await Task.WhenAll(...)

真正难处理的是跨多层嵌套的取消和超时——Channel 本身不持有 CancellationToken，所有 WriteAsync/ReadAsync 调用都得显式传入，漏一个就可能卡死。这点比 Go 的 channel 更易出错。