C# Academy · 课时

IAsyncEnumerable 与 await foreach

使用 IAsyncEnumerable 异步传输数据，在异步迭代器中使用 yield，并通过 await foreach 消费数据。

IAsyncEnumerable 与 await foreach 是 CoddyKit 上的免费 C# Academy 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 C# Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 C# Academy 课程共包含 4 节课。

为什么使用 IAsyncEnumerable？

IAsyncEnumerable<T>（在 C# 8 和 .NET Core 3 中引入）支持异步流式处理——项目可在可用时逐个产出，而无需先将所有内容加载到内存中。可以将其理解为逐行读取数据库游标，或流式接收 API 响应。

编写异步迭代器

声明一个返回 IAsyncEnumerable<T> 的方法，并在 async 方法中使用 yield return。await 和 yield return 关键字可以自然共存。

public async IAsyncEnumerable<int> GenerateNumbersAsync(int count)
{
    for (int i = 0; i < count; i++)
    {
        await Task.Delay(100); // simulate async work per item
        yield return i;
    }
}

使用 await foreach 进行消费

使用 await foreach 消费 IAsyncEnumerable<T>。每次迭代都会异步等待下一个项目——consumer 和 producer 会自然地保持协调节奏。

await foreach (var number in GenerateNumbersAsync(10))
{
    Console.WriteLine(number);
}

// With cancellation:
var cts = new CancellationTokenSource();
await foreach (var n in GenerateNumbersAsync(100)
    .WithCancellation(cts.Token))
{
    if (n > 50) { cts.Cancel(); break; }
    Console.WriteLine(n);
}

从数据库进行流式读取

EF Core 支持使用 AsAsyncEnumerable() 流式读取查询结果。它会逐行读取数据，而不会将完整结果集加载到内存中——这对大型表至关重要。

public async IAsyncEnumerable<Product> StreamProductsAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    await foreach (var product in _db.Products
        .Where(p => p.IsActive)
        .AsAsyncEnumerable()
        .WithCancellation(ct))
    {
        yield return product;
    }
}

流式传输 HTTP 响应

向客户端流式传输大型响应，而不缓存整个负载。服务器继续处理的同时，客户端可以逐步接收数据。

app.MapGet("/stream/products", (ProductRepository repo) =>
{
    // Returns an IAsyncEnumerable — ASP.NET Core streams it as JSON array
    return repo.StreamAllAsync();
});

// Or with custom serialization:
app.MapGet("/stream/events", async (HttpResponse response, EventStore store) =>
{
    response.ContentType = "application/json";
    await foreach (var e in store.StreamAsync())
    {
        await response.WriteAsJsonAsync(e);
        await response.Body.FlushAsync();
    }
});

EnumeratorCancellation 特性

将 [EnumeratorCancellation] 应用于异步迭代器的 CancellationToken 参数，使 consumer 端的 WithCancellation() 能够将令牌传递给 producer。

public async IAsyncEnumerable<LogEntry> TailLogAsync(
    string logFile,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    while (!ct.IsCancellationRequested)
    {
        var newLines = await ReadNewLinesAsync(logFile, ct);
        foreach (var line in newLines)
            yield return new LogEntry(line);
        await Task.Delay(500, ct);
    }
}

// Consumer:
await foreach (var entry in TailLogAsync("/var/log/app.log")
    .WithCancellation(cts.Token))
{
    Console.WriteLine(entry.Message);
}

在 IAsyncEnumerable 上使用 LINQ

System.Linq.Async NuGet 包为 IAsyncEnumerable 提供异步 LINQ 运算符（WhereAwait、SelectAwait、ToListAsync）。

// dotnet add package System.Linq.Async

var expensiveProducts = await repo.StreamAllAsync()
    .WhereAwait(async p => await pricing.IsExpensiveAsync(p.Id))
    .SelectAwait(async p => await enricher.EnrichAsync(p))
    .Take(10)
    .ToListAsync();

// Or use built-in Where/Select that don't need async:
var names = repo.StreamAllAsync()
    .Where(p => p.IsActive)
    .Select(p => p.Name);

合并多个异步流

若要并发处理多个异步流，请使用 Channel 作为汇入点——从多个 producer 写入，并由单个 consumer 读取。

async IAsyncEnumerable<T> Merge<T>(
    params IAsyncEnumerable<T>[] streams)
{
    var channel = Channel.CreateUnbounded<T>();

    var producers = streams.Select(async stream =>
    {
        await foreach (var item in stream)
            await channel.Writer.WriteAsync(item);
    });

    _ = Task.WhenAll(producers)
            .ContinueWith(_ => channel.Writer.Complete());

    await foreach (var item in channel.Reader.ReadAllAsync())
        yield return item;
}

性能：缓存与流式处理

处理大型数据集时应始终使用流式处理。先将所有内容加载到列表中（ToListAsync()）会同时将全部数据保留在内存中，而流式处理会逐个处理项目。

// BAD: loads all 1M records into memory
var allOrders = await _db.Orders.ToListAsync();
foreach (var order in allOrders)
    await ProcessAsync(order);

// GOOD: processes one record at a time
await foreach (var order in _db.Orders.AsAsyncEnumerable())
    await ProcessAsync(order);
// Memory usage: O(1) vs O(n)

异步迭代器中的错误处理

将整个 await foreach 包装在 try/catch 中，以处理 producer 产生的错误。迭代器中的错误会在当前迭代点传播给 consumer。

try
{
    await foreach (var item in RiskyStreamAsync())
    {
        await ProcessAsync(item);
    }
}
catch (HttpRequestException ex)
{
    Console.Error.WriteLine($"Stream error: {ex.Message}");
}
catch (OperationCanceledException)
{
    Console.WriteLine("Stream cancelled.");
}

快速检查

与返回 List<T> 相比，IAsyncEnumerable 的主要内存优势是什么？

回顾：IAsyncEnumerable 与 await foreach

要点：

IAsyncEnumerable<T>：逐个异步流式传输项目
在 async 方法中使用 yield return 可生成异步迭代器
await foreach：通过 .WithCancellation() 支持取消来消费流
[EnumeratorCancellation]：将 consumer 的令牌传递给迭代器
EF Core：对大型查询使用 AsAsyncEnumerable() 以提高内存效率
处理大型数据集时使用流式处理，而不是 ToListAsync()

常见问题解答

「IAsyncEnumerable 与 await foreach」课时是免费的吗？

是的 — 「IAsyncEnumerable 与 await foreach」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 C# Academy 课程的其余内容，请升级到 CoddyKit PRO。 C# Academy 课程共包含 4 节课。

「IAsyncEnumerable 与 await foreach」这节课中我会学到什么？

使用 IAsyncEnumerable 异步传输数据，在异步迭代器中使用 yield，并通过 await foreach 消费数据。你通过在浏览器中直接运行的动手代码来练习 C# Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 C# Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 C# Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「IAsyncEnumerable 与 await foreach」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 C# Academy 课中编写并运行代码吗？

能。每节 C# Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 C# Academy