【.NET并发编程 - 16】IAsyncEnumerable 异步流：边加载边处理惩罚的优雅之道

16. IAsyncEnumerable 异步流：边加载边处理惩罚的优雅之道

本章 GitHub 堆栈：csharp-concurrency-cookbook ⭐
欢迎 Star 和 Fork！全部代码示例都可以在堆栈中找到并运行。

🎯 本章导读

📌 本文目标：把握 IAsyncEnumerable 的利用场景、焦点语法和性能优化本事，实现真正的流式异步数据处理惩罚。

你是否遇到过如许的场景：

• 从数据库查询 10 万条记载，要等全部加载完才气开始处理惩罚？
  
• 调用分页 API 获取数据，必须期待全部页都下载完？
  
• 实时日志流、传感器数据流，怎样边汲取边处理惩罚？
  
• 大文件读取时，怎样克制一次性加载到内存导致 OOM？
  

本日，我们就来彻底搞懂 IAsyncEnumerable——.NET 异步流的标准办理方案（C# 8.0 引入，.NET Core 3.0+ 支持）。

⚠️ 前置知识：本文涉及 async/await、yield return、Task 等概念，发起先把握前面章节的异步编程根本（第 03-06 章）。

0️⃣ 一个真实的故事：GitHub API 的分页加载窘境

0.1 场景重现：获取某个堆栈的全部 Issue

假设你要写一个工具，获取 GitHub 某个热门堆栈的全部 Issue（大概有上万个），然后统计标签分布、关键词频率等。
GitHub API 是分页的：每次哀求最多返回 100 条，必要通过 page 参数逐页获取。
你写出了第一版代码：

1. // ❌ 方案1：全部加载完再处理 —— 慢，还占内存
2. public async Task<List<GitHubIssue>> GetAllIssuesAsync(string repo)
3. {
4.         var allIssues = new List<GitHubIssue>();
5.         int page = 1;
7.         while (true)
8.         {
9.                 var issues = await FetchPageAsync(repo, page); // 等待每一页
10.                 if (issues.Count == 0) break;
12.                 allIssues.AddRange(issues); // 堆积在内存里
13.                 page++;
14.         }
16.         return allIssues; // 全部返回后才能开始处理
17. }
19. // 调用方必须等待所有数据加载完
20. var allIssues = await GetAllIssuesAsync("dotnet/runtime"); // 等了 30 秒...
21. foreach (var issue in allIssues)
22. {
23.         ProcessIssue(issue); // 终于可以开始处理了
24. }

复制代码

0.2 这个实现的标题

运行后你发现了几个标题：

• ❌ 相应慢：用户要期待全部数据加载完（大概几十秒）才气看到第一条效果
  
• ❌ 内存占用高：1 万条 Issue 全部加载到内存，大概占用几十 MB
  
• ❌ 不敷流式：数据是"批量"的，不是"流式"的，无法边加载边处理惩罚
  
• ❌ 无法提前取消：如果用户只必要前 100 条，也要等全部加载完
  

你开始思索：能不能边加载边返回？就像水龙头一样，一边出水一边用，而不是先接满一桶再倒出来？
0.3 IAsyncEnumerable 的登场

.NET 提供了一套完备的异步流机制：IAsyncEnumerable。
利用 IAsyncEnumerable 重写后：

1. // ✅ 方案2：流式返回 —— 快，省内存
2. public async IAsyncEnumerable<GitHubIssue> GetAllIssuesStreamAsync(string repo)
3. {
4.         int page = 1;
6.         while (true)
7.         {
8.                 var issues = await FetchPageAsync(repo, page);
9.                 if (issues.Count == 0) break;
11.                 foreach (var issue in issues)
12.                 {
13.                         yield return issue; // 逐个返回，不等全部加载完
14.                 }
16.                 page++;
17.         }
18. }
20. // 调用方立刻就能开始处理第一条数据
21. await foreach (var issue in GetAllIssuesStreamAsync("dotnet/runtime"))
22. {
23.         ProcessIssue(issue); // 边加载边处理！
24. }

复制代码

神奇的效果：

• ✅ 相应快：第一页加载完（1 秒）就开始处理惩罚，无需期待全部
  
• ✅ 内存低：同时在内存中的只有当前处理惩罚的那一页（100 条）
  
• ✅ 真正的流式：数据像水流一样源源不停，边来边处理惩罚
  
• ✅ 可以提前竣事：用 break 或 return 立即克制，后续页不会再加载
  

看到这里，你是不是开始心动了？别急，我们从原理讲起。
1️⃣ IAsyncEnumerable 是什么？它与 IEnumerable 有什么区别?

1.1 回顾：IEnumerable 的同步天下

我们先回顾一下 IEnumerable，它是 .NET 中最常见的"可摆列"接口：

1. // 同步版本：生成 1 到 10 的数字
2. public IEnumerable<int> GetNumbers()
3. {
4.         for (int i = 1; i <= 10; i++)
5.         {
6.                 yield return i; // 每次返回一个数字
7.         }
8. }
10. // 消费方式
11. foreach (var num in GetNumbers())
12. {
13.         Console.WriteLine(num); // 同步逐个处理
14. }

复制代码

6️⃣ 性能优化：ValueTask vs Task、缓冲战略

6.1 为什么 MoveNextAsync 返回 ValueTask？

你大概留意到了，IAsyncEnumerator.MoveNextAsync() 返回的是 ValueTask，而不是 Task。
缘故原由：对于高频调用的场景（如数据库游标、文件流），如果每次 MoveNextAsync 都分配一个 Task 对象，会产生大量 GC 压力。
ValueTask 的上风：

• 如果利用同步完成（如数据已在缓冲区），直接返回效果，不分配堆内存
  
• 如果利用异步完成（必要期待 I/O），才分配 Task
  

这就是为什么 IAsyncEnumerable 比本身手写 Task 更高效的缘故原由之一。
6.2 缓冲战略：镌汰网络来回

对于网络 API，可以实现一个"预加载"战略：当前页在处理惩罚时，背景已经在加载下一页。

1. // ❌ 错误示范：IEnumerable 里不能用 await
2. public IEnumerable<GitHubIssue> GetIssues()
3. {
4.         var issues = await FetchPageAsync(1); // ❌ 编译错误：yield return 方法不能是 async
5.         foreach (var issue in issues)
6.                 yield return issue;
7. }

复制代码

效果：当前页在处理惩罚时，下一页已经在背景加载，镌汰用户期待时间。
6.3 ⚠️ 克制 CA2024 告诫：不要在异步方法中利用 EndOfStream

在读取文件流时，你大概会遇到这个告诫：

1. // ❌ 不够优雅：失去了"流式"的优势
2. public async Task<IEnumerable<GitHubIssue>> GetIssuesAsync()
3. {
4.         var allIssues = new List<GitHubIssue>();
5.         var issues = await FetchPageAsync(1);
6.         allIssues.AddRange(issues);
7.         return allIssues; // 一次性返回
8. }

复制代码

标题：EndOfStream 是一个同步属性，在异步方法中调用大概导致壅闭，违反了异步编程的原则。
精确做法：通过 ReadLineAsync 的返回值（null）来判定流竣事：

1. public interface IAsyncEnumerable<out T>
2. {
3.         IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
4. }
6. public interface IAsyncEnumerator<out T> : IAsyncDisposable
7. {
8.         T Current { get; }
9.         ValueTask<bool> MoveNextAsync(); // 注意：这里是 ValueTask，不是 Task
10. }

复制代码

6.4 高性能方案：利用 System.IO.Pipelines（.NET Core 3.0+）

对于大文件、高吞吐量场景，System.IO.Pipelines 提供了更高效的流处理惩罚方式：
上风：

• 利用内存池（Memory Pool），镌汰 GC 压力
  
• 更高效的缓冲管理
  
• 支持背压控制
  
• 范例性能提升：1.5-3x，内存分配镌汰 50-80%
  

实用场景：

• 大文件（> 100 MB）
  
• 高吞吐量场景（如日志处理惩罚、数据导入）
  
• 网络流处理惩罚
  

实现示例：

1. // ✅ 正确示范：异步流式返回
2. public async IAsyncEnumerable<int> GetNumbersAsync()
3. {
4.         for (int i = 1; i <= 10; i++)
5.         {
6.                 await Task.Delay(100); // 模拟异步操作（如 API 调用）
7.                 yield return i; // 每次返回一个数字
8.         }
9. }
11. // 消费方式：await foreach
12. await foreach (var num in GetNumbersAsync())
13. {
14.         Console.WriteLine(num); // 边加载边处理
15. }

复制代码

性能对比（10 万行文件）：
方案耗时内存分配吞吐量StreamReader150 ms~20 MB666K 行/秒Pipeline100 ms~5 MB1M 行/秒提升1.5x镌汰 75%1.5x选择发起：
场景保举方案缘故原由小文件（< 10 MB）StreamReader代码简朴，性可以大概用大文件（> 100 MB）Pipeline明显的性能提升和内存节流实时监控（tail -f）StreamReader实时性更告急，性能差异不显着高吞吐量场景Pipeline镌汰 GC 压力，进步稳固性

完备示例代码：AsyncEnumerable/LogStreamProcessor.cs 和 StreamReaderVsPipelineComparison.cs

7️⃣ 常见陷阱与最佳实践

7.1 ❌ 陷阱1：多次遍历会重新实行

1. public async Task<List<int>> GetDataTraditionalAsync()
2. {
3.         var result = new List<int>();
4.         for (int page = 1; page <= 10; page++)
5.         {
6.                 await Task.Delay(200); // 模拟 API 延迟
7.                 for (int i = 1; i <= 100; i++)
8.                         result.Add(page * 100 + i);
9.         }
10.         return result;
11. }
13. // 使用方式
14. var sw = Stopwatch.StartNew();
15. var data = await GetDataTraditionalAsync(); // 等待 2 秒...
16. Console.WriteLine($"开始处理第一条数据，已过 {sw.ElapsedMilliseconds}ms");
18. foreach (var item in data)
19.         ProcessItem(item); // 终于可以开始了

复制代码

办理方案：如果必要多次遍历，先转换为 List：

1. 开始处理第一条数据，已过 2000ms
2. 处理完成，总耗时 2500ms

复制代码

7.2 ❌ 陷阱2：忘记 await foreach

1. public async IAsyncEnumerable<int> GetDataStreamAsync()
2. {
3.         for (int page = 1; page <= 10; page++)
4.         {
5.                 await Task.Delay(200); // 模拟 API 延迟
6.                 for (int i = 1; i <= 100; i++)
7.                         yield return page * 100 + i;
8.         }
9. }
11. // 使用方式
12. var sw = Stopwatch.StartNew();
13. await foreach (var item in GetDataStreamAsync())
14. {
15.         if (sw.ElapsedMilliseconds < 500) // 只记录前 500ms 的日志
16.                 Console.WriteLine($"处理第 {item} 条数据，已过 {sw.ElapsedMilliseconds}ms");
18.         ProcessItem(item);
19. }

复制代码

7.3 ❌ 陷阱3：在 yield return 之前做太多工作

1. 处理第 101 条数据，已过 201ms
2. 处理第 102 条数据，已过 201ms
3. ...（第一页的数据立刻就开始处理了）
4. 处理第 201 条数据，已过 401ms
5. 处理完成，总耗时 2500ms

复制代码

7.4 ✅ 最佳实践总结

场景保举做法必要取消利用 [EnumeratorCancellation] + WithCancellation()库代码利用 ConfigureAwait(false)多次遍历先 ToListAsync()必要总数返回 (IAsyncEnumerable data, int totalCount) 元组错误处理惩罚在 yield 之前 try-catch，决定是跳过照旧克制进度反馈在 yield 时更新进度（如日志、UI）8️⃣ EF Core 中的 IAsyncEnumerable 支持

Entity Framework Core 原生支持异步流：

1. // 基础语法
2. public async IAsyncEnumerable<int> GetNumbersAsync(int count)
3. {
4.         for (int i = 1; i <= count; i++)
5.         {
6.                 await Task.Delay(100); // 每个数字都需要异步操作
7.                 yield return i;
8.         }
9. }

复制代码

对比传统方式：

1. // 使用 [EnumeratorCancellation] 特性接收取消令牌
2. public async IAsyncEnumerable<int> GetNumbersAsync(
3.         int count,
4.         [EnumeratorCancellation] CancellationToken cancellationToken = default)
5. {
6.         for (int i = 1; i <= count; i++)
7.         {
8.                 cancellationToken.ThrowIfCancellationRequested(); // 检查取消
9.                 await Task.Delay(100, cancellationToken);
10.                 yield return i;
11.         }
12. }

复制代码

9️⃣ 实战案例2：实时日志流处理惩罚

完备代码位置：AsyncEnumerable/LogStreamProcessor.cs

9.1 需求

监控一个日志文件，实时读取新增的行（雷同 tail -f）。
9.2 实现

1. public async IAsyncEnumerable<string> FetchUrlsAsync(string[] urls)
2. {
3.         foreach (var url in urls)
4.         {
5.                 string? content = null;
6.                 try
7.                 {
8.                         content = await FetchAsync(url);
9.                 }
10.                 catch (HttpRequestException ex)
11.                 {
12.                         Console.WriteLine($"获取 {url} 失败: {ex.Message}");
13.                         continue; // 跳过失败的 URL
14.                 }
16.                 if (content != null)
17.                         yield return content;
18.         }
19. }

复制代码

🔟 IAsyncEnumerable 与 SSE（Server-Sent Events）的关系

10.1 什么是 SSE？

SSE（Server-Sent Events） 是一种 HTTP 长毗连技能，允许服务器主动向客户端推送数据流，常用于：

• 💬 实时谈天消息推送
  
• 📊 股票代价、角逐分数实时更新
  
• 📝 AI 流式输出（如 ChatGPT 的打字机效果）
  
• 🔔 体系关照推送
  

技能特点：

• 基于 HTTP，客户端发起毗连后，服务器一连推送 data: 格式的文本流
  
• 单向通讯（服务器 → 客户端）
  
• 主动重连机制
  
• 兼容性好（支持 EventSource API）
  

10.2 IAsyncEnumerable 与 SSE 的天然契合

在 ASP.NET Core 中，IAsyncEnumerable 是实现 SSE 的完善载体：

1. await foreach (var item in GetNumbersAsync(10))
2. {
3.         Console.WriteLine(item);
4. }

复制代码

10.3 实战案例：AI 流式输出（ChatGPT 效果）

场景：调用 AI API，逐字返回天生的内容，而不是等全部天生完。

1. var cts = new CancellationTokenSource();
2. cts.CancelAfter(TimeSpan.FromSeconds(2)); // 2 秒后取消
4. try
5. {
6.         await foreach (var item in GetNumbersAsync(100).WithCancellation(cts.Token))
7.         {
8.                 Console.WriteLine(item);
9.         }
10. }
11. catch (OperationCanceledException)
12. {
13.         Console.WriteLine("操作已取消");
14. }

复制代码

前端效果（打字机动画）：

1. await foreach (var item in GetNumbersAsync(10).ConfigureAwait(false))
2. {
3.         // 不捕获 SynchronizationContext，适合库代码
4.         Console.WriteLine(item);
5. }

复制代码

用户体验：

• ❌ 传统方式：期待 5 秒，忽然体现完备答案
  
• ✅ 流式方式：立即开始体现，逐字出现，体验流通
  

10.4 IAsyncEnumerable 与 SSE 的区别与接洽

维度IAsyncEnumerableSSE本质C# 语言特性（异步迭代器）HTTP 传输协议作用范围服务器端数据流处理惩罚服务器 → 客户端推送关系SSE 的服务端实现方式SSE 的传输格式范例用法await foreach 处理惩罚流式数据EventSource 汲取服务器推送ASP.NET Core 集成返回 IAsyncEnumerable 主动转换为 SSE框架主动处理惩罚 SSE 格式焦点明白：

• IAsyncEnumerable 是 生产者端的抽象（怎样天生流式数据）
  
• SSE 是 传输层的协议（怎样通过 HTTP 推送数据）
  
• 在 ASP.NET Core 中，IAsyncEnumerable 主动映射为 SSE
  

10.5 其他流式技能对比

技能通讯方式协议实用场景IAsyncEnumerable 支持SSE单向（服务器 → 客户端）HTTP实时推送、股票行情、日志流✅ 原生支持WebSocket双向（全双工）WebSocket谈天室、协作编辑、游戏✅ 可共同利用gRPC Streaming双向（基于 HTTP/2）gRPC微服务间流式通讯✅ 原生支持SignalR双向（抽象层）多种（WebSocket/SSE/长轮询）实时应用框架✅ 内部利用选择发起：

• 简朴的服务器推送 → SSE + IAsyncEnumerable
  
• 必要客户端回传数据 → WebSocket
  
• 微服务流式调用 → gRPC Streaming
  
• 高级实时功能（如房间、群组） → SignalR
  

10.6 完备示例：实时日志推送

服务端（ASP.NET Core）：

1. await foreach (var item in GetNumbersAsync(100))
2. {
3.         Console.WriteLine(item);
4.         if (item == 10)
5.                 break; // 停止迭代，后续数据不会再生成
6. }

复制代码

客户端（欣赏器）：
[code][/code]效果：欣赏器实时体现服务器日志，无需革新页面。
10.7 要点总结

✅ 关系：IAsyncEnumerable 是 SSE 在 ASP.NET Core 中的服务端实现方式
✅ 上风：代码轻巧（无需手动拼接 SSE 格式），框架主动处理惩罚
✅ 实用场景：

• 实时数据推送（股票、天气、比分）
  
• AI 流式输出（ChatGPT 效果）
  
• 日志/监控流
  
• 进度更新
  

✅ 限定：

• 只支持单向推送（客户端 → 服务器必要额外哀求）
  
• 文本格式（二进制数据必要 Base64 编码）
  

❌ 不得当：必要双向通讯的场景（发起用 WebSocket 或 SignalR）
🎓 本章总结

焦点要点回顾

• IAsyncEnumerable 是什么：异步版本的 IEnumerable，支持流式返回异步数据
  
• 焦点语法：
  
  
  
  • 生产者：async IAsyncEnumerable + yield return + await
    
  • 消耗者：await foreach
    
  
  
• 上风：
  
  
  
  • 相应快：边加载边处理惩罚，无需期待全部完成
    
  • 内存低：同时在内存中的只有当前批次
    
  • 可取消：支持 CancellationToken
    
  
  
• 实用场景：分页 API、大数据集查询、实时数据流、日志流
  
• 性能优化：ValueTask、预加载战略、批量处理惩罚
  
• 常见陷阱：多次遍历会重新实行、忘记 await foreach、在 yield 之前加载全部数据
  

与其他技能的对比

技能实用场景上风劣势Task小数据量、必要多次遍历简朴、可多次遍历内存占用高、相应慢IAsyncEnumerable大数据量、流式处理惩罚内存低、相应快不能多次遍历、不知道总数Channel生产者-消耗者模式解耦、背压控制代码复杂IObservable (Rx)复杂变乱流强大的组合利用符学习曲线陡峭SSE (Server-Sent Events)HTTP 实时推送基于 HTTP、主动重连单向推送、文本格式最佳实践清单

• ✅ 对于 API 分页、数据库大数据集，优先思量 IAsyncEnumerable
  
• ✅ 始终支持 CancellationToken（利用 [EnumeratorCancellation]）
  
• ✅ 库代码中利用 ConfigureAwait(false)
  
• ✅ 必要多次遍历时，先 ToListAsync()
  
• ✅ 在 yield 之前处理惩罚非常，克制流克制
  
• ✅ 利用 System.Linq.Async 举行 LINQ 利用
  
• ❌ 不要在 yield 之前加载全部数据（失去流式意义）
  
• ❌ 不要忘记 await foreach
  

📚 保举阅读

• 官方文档
  
  
  
  • IAsyncEnumerable Interface
    
  • System.Linq.Async on NuGet