c# - 与巨大的数据流异步

Question

我们使用 IEnumerables 从数据库返回庞大的数据集:

public IEnumerable<Data> Read(...)
{
    using(var connection = new SqlConnection(...))
    {
        // ...
        while(reader.Read())
        {
            // ...
            yield return item;
        }
    }
}

现在我们想使用异步方法来做同样的事情。然而，没有用于异步的 IEnumerables，因此我们必须将数据收集到列表中，直到加载整个数据集:

public async Task<List<Data>> ReadAsync(...)
{
    var result = new List<Data>();
    using(var connection = new SqlConnection(...))
    {
        // ...
        while(await reader.ReadAsync().ConfigureAwait(false))
        {
            // ...
            result.Add(item);
        }
    }
    return result;
}

这将消耗服务器上的大量资源，因为返回前所有数据都必须在列表中。 IEnumerables 处理大数据流的最佳且易于使用的异步替代方案是什么？我想避免在处理时将所有数据存储在内存中。

Answer 1

最简单的选择是使用 TPL Dataflow .您需要做的就是配置 ActionBlock负责处理(如果您愿意，可以并行处理)并异步“发送”项目一个接一个。
我还建议设置一个 BoundedCapacity，当处理速度无法处理时，它会限制读取器从数据库中读取数据。

var block = new ActionBlock<Data>(
    data => ProcessDataAsync(data),
    new ExecutionDataflowBlockOptions
    {
        BoundedCapacity = 1000,
        MaxDegreeOfParallelism = Environment.ProcessorCount
    });

using(var connection = new SqlConnection(...))
{
    // ...
    while(await reader.ReadAsync().ConfigureAwait(false))
    {
        // ...
       await block.SendAsync(item);
    }
}

您还可以使用 Reactive Extensions ，但这是一个比您可能需要的更复杂、更强大的框架。