当前位置：首页 > C# > 正文

C#异步流的批量处理优化（深入理解 IAsyncEnumerable 与高效数据处理）

主机测评网
C#
2025-12-08
398

在现代 C# 开发中，异步流（Asynchronous Streams）已成为处理大量实时或延迟加载数据的重要工具。借助 IAsyncEnumerable<T> 接口，开发者可以优雅地实现按需、逐项获取数据的能力，同时保持程序响应性和资源效率。

然而，在实际项目中，我们常常需要对这些异步流进行批量处理（Batch Processing），以提升性能、减少数据库或 API 调用次数、优化内存使用等。本文将带你从零开始，深入理解如何在 C# 中高效地对 IAsyncEnumerable<T> 进行批量处理，并提供可直接复用的优化方案。

什么是 C# 异步流？

自 C# 8.0 起，.NET 引入了 IAsyncEnumerable<T> 接口，它允许你以异步方式逐个返回元素。这与传统的 IEnumerable<T> 不同，后者是同步遍历的。

一个简单的异步流示例如下：

public async IAsyncEnumerable<int> GetNumbersAsync(int count){    for (int i = 0; i < count; i++)    {        await Task.Delay(100); // 模拟异步操作        yield return i;    }}

上述方法会每隔 100 毫秒返回一个整数，调用者可通过 await foreach 来消费：

await foreach (var number in GetNumbersAsync(5)){    Console.WriteLine(number);}

为什么需要批量处理？

虽然逐项处理很灵活，但在某些场景下效率较低。例如：

向数据库批量插入记录（比单条插入快数十倍）
调用支持批量参数的 REST API（减少 HTTP 请求次数）
避免频繁分配小对象，提升 GC 性能

C#异步流的批量处理优化（深入理解 IAsyncEnumerable 与高效数据处理） C#异步流批量处理优化异步编程第1张

实现异步流的批量处理

我们可以编写一个通用的扩展方法，将 IAsyncEnumerable<T> 按指定大小分批处理。以下是推荐的实现方式：

public static async IAsyncEnumerable<IList<T>> BatchAsync<T>(    this IAsyncEnumerable<T> source,    int batchSize){    if (batchSize <= 0)        throw new ArgumentException("Batch size must be greater than zero.", nameof(batchSize));    var batch = new List<T>(batchSize);    await foreach (var item in source.ConfigureAwait(false))    {        batch.Add(item);        if (batch.Count == batchSize)        {            yield return batch;            batch = new List<T>(batchSize);        }    }    if (batch.Count > 0)        yield return batch;}

这个扩展方法的关键点包括：

使用 ConfigureAwait(false) 避免不必要的上下文切换（提升性能）
预分配 List<T> 容量，减少内存重分配
最后一批不足 batchSize 的数据也会被返回

使用示例：批量写入数据库

假设你有一个从 API 获取用户数据的异步流，现在要每 100 条批量插入数据库：

var usersStream = FetchUsersFromApiAsync(); // 返回 IAsyncEnumerable<User>await foreach (var batch in usersStream.BatchAsync(100)){    await dbContext.Users.AddRangeAsync(batch);    await dbContext.SaveChangesAsync();}

这种方式显著减少了数据库事务数量，提升了整体吞吐量。