异步数据流

1. 介绍

你应该已经熟悉 async 和 await。它们对“单次”的异步操作（比如下载一个文件）很管用。但如果数据是以流方式到来，或者资源需要异步清理，该怎么办？

异步“集合”的问题：想象你需要从数据库拿上百万条记录。如果方法返回 Task<List<T>>，你就得等到所有数据都加载到内存里。这既低效又会导致延迟。如果每个元素都需要异步获取，普通的同步 IEnumerable<T> 也不够用。

异步释放资源的问题：IDisposable 和 using 很适合同步清理。但如果关闭网络连接或把缓冲区写到文件本身就是异步操作呢？在同步的 Dispose() 里不能用 await，这会导致线程阻塞或清理不正确。

为了解决这些问题，引入了 IAsyncEnumerable<T> 和 IAsyncDisposable。

2. 异步数据流

IAsyncEnumerable<T> 是 IEnumerable<T> 的异步对应。它允许异步地按项生成序列元素，不需要等所有数据都准备好。

什么时候需要？

• 逐行异步读取大文件：比如，数 GB 的日志文件。
• 从网络或数据库流式传输数据：API 的结果分批到达。
• 实现服务器端的流式 API：比如 gRPC Streaming。
• 任何数据是异步产生或到达并且需要被渐进处理的场景。

它是怎么工作的？

1. IAsyncEnumerable<T>：接口有一个方法 GetAsyncEnumerator(CancellationToken cancellationToken)。取消令牌很重要！
2. IAsyncEnumerator<T>：接口有 ValueTask<bool> MoveNextAsync()（移动到下一个）和 Current（当前元素）。它也继承自 IAsyncDisposable。
3. await foreach：可以方便地迭代 IAsyncEnumerable<T>。编译器会帮你调用 MoveNextAsync() 和 Current。重要的是，await foreach 保证在迭代结束后（即便发生异常）调用迭代器的 DisposeAsync()。

用 async yield return 创建 IAsyncEnumerable<T>

你可以在返回 IAsyncEnumerable<T> 的 async 方法里使用 yield return。这允许你写出异步生成器。方法里可以用 await 去暂停生成、等待异步操作，然后继续生成。

示例：简单的异步生成器

async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine($"生成中: {i}");
        await Task.Delay(100); // 模拟异步工作
        yield return i; 
    }
}

// 用法:
async Task ConsumeAsyncNumbers()
{
    await foreach (var number in GenerateNumbersAsync())
    {
        Console.WriteLine($"收到: {number}");
    }
}
// 调用：await ConsumeAsyncNumbers();

示例：逐行异步读取文件

async IAsyncEnumerable<string> ReadFileLinesAsync(string filePath)
{
    using var reader = new StreamReader(filePath); // 这里用 'using'（StreamReader 实现了 IAsyncDisposable）
    string? line;
    while ((line = await reader.ReadLineAsync()) != null) 
    {
        yield return line;
    }
}

// 用法:
async Task ProcessFileAsync()
{
    await File.WriteAllLinesAsync("data.txt", new[] { "行 1", "行 2", "行 3" });
    await foreach (var line in ReadFileLinesAsync("data.txt")) 
    {
        Console.WriteLine($"已处理行: {line}");
    }
}
// 调用：await ProcessFileAsync();

示例：支持取消的异步生成器（CancellationToken）

async IAsyncEnumerable<int> GetCancelableSequence(
    [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        token.ThrowIfCancellationRequested(); // 检查取消
        await Task.Delay(200, token); // Task.Delay 也支持通过 token 取消
        yield return i;
    }
}

// 用法:
async Task ConsumeAndCancel()
{
    var cts = new CancellationTokenSource(500); // 500ms 后取消
    try
    {
        await foreach (var num in GetCancelableSequence(cts.Token))
        {
            Console.WriteLine($"收到: {num}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("生成已取消！");
    }
}
// 调用：await ConsumeAndCancel();

属性 [EnumeratorCancellation] 允许把 CancellationToken 传给异步生成器。这样当调用方通过 CancellationTokenSource 请求取消时，迭代可以被取消。如果没有这个属性，token 不会自动传递到 GetAsyncEnumerator。

3. 异步资源管理

同步 IDisposable 的问题

Dispose() 在 IDisposable 中是同步的（void Dispose()）。你不能在里面使用 await。如果关闭数据库连接或把缓冲区刷新到文件是耗时且异步的，同步的 Dispose() 会阻塞线程，这对异步应用不友好。

解决方案： IAsyncDisposable

IAsyncDisposable 就是为了解决这个问题。它只有一个方法：ValueTask DisposeAsync() — 用于异步清理的接口方法。

await using

这是 using 的异步版本，针对实现了 IAsyncDisposable 的对象。

• await using 保证在资源声明的代码块结束时（或因异常退出时）调用 DisposeAsync()。
• 允许正确地以异步方式释放资源，避免阻塞。

示例：基本的 IAsyncDisposable 和 await using

class MyAsyncResource : IAsyncDisposable
{
    public MyAsyncResource() => Console.WriteLine("资源已打开。");
    
    public async ValueTask DisposeAsync()
    {
        Console.WriteLine("开始异步清理...");
        await Task.Delay(200); // 模拟异步清理
        Console.WriteLine("异步清理完成。");
    }
}

// 使用 await using
async Task UseAndDisposeResource()
{
    await using var resource = new MyAsyncResource(); 
    Console.WriteLine("正在使用资源...");
} // 这里会自动调用 resource.DisposeAsync()

// 调用：await UseAndDisposeResource();

示例：多个 await using 块

async Task UseMultipleResources()
{
    await using var res1 = new MyAsyncResource();
    await using var res2 = new MyAsyncResource();
    Console.WriteLine("正在使用两个资源...");
} // 资源按 LIFO（后进先出）顺序释放：先调用 res2.DisposeAsync()，然后 res1.DisposeAsync()。
// 调用：await UseMultipleResources();

IAsyncEnumerable<T> 与 IAsyncDisposable 的兼容性

重要的是：IAsyncEnumerator<T>（用于 await foreach）本身继承自 IAsyncDisposable。这意味着如果你的异步生成器使用了需要异步清理的资源（像上面示例里的 StreamReader），await foreach 会帮你处理清理——在迭代结束时它会调用迭代器的 DisposeAsync()。