如果将IEnumerable与async/await一起使用(使用Dapper从SQL Server中流式传输数据)，会发生什么情况? [英] What happens with returning IEnumerable if used with async/await (streaming data from SQL Server with Dapper)?

问题描述

我正在使用Dapper从SQL Server中的一个非常大的数据集中流式传输数据.返回IEnumerable并调用Query()效果很好，但是当我切换到QueryAsync()时，该程序似乎尝试从SQL Server读取所有数据，而不是流式传输.

I am using Dapper to stream data from a very large set in SQL Server. It works fine with returning IEnumerable and calling Query(), but when I switch to QueryAsync(), it seems that the program tries to read all of the data from SQL Server instead of streaming.

根据此问题，它应该可以与buffered: false正常工作，我在做，但是问题没有说明async/await.

According to this question, it should work fine with buffered: false, which I am doing, but the question says nothing about async/await.

现在根据此问题，要完成我想做的事情并不容易QueryAsync().

Now according to this question, it's not straightforward to do what I want with QueryAsync().

我是否正确理解在为async/await切换上下文时会枚举可枚举的对象?

Do I understand correctly that enumerables are iterated when the context is switched for async/await?

另一个问题是，当新的C#8异步流可用时，是否可以做到这一点?

Another question if this is something that will be possible to do when the new C#8 async streaming is available?

推荐答案

更新2020年3月

.NET Core 3.0(和3.1)现已推出，完全支持异步流. Microsoft.Bcl.AsyncInterfaces 将其支持添加到.NET Standard 2.0和.NET Framework 4.6.1+，尽管出于理智考虑应使用4.7.2.如 .NET标准实施支持上的文档所述，

.NET Core 3.0 (and 3.1) have come out now, with full support for async streams. The Microsoft.Bcl.AsyncInterfaces adds support for them to .NET Standard 2.0 and .NET Framework 4.6.1+, although 4.7.2 should be used for sanity reasons. As the docs on .NET Standard implementation support explain

虽然NuGet认为.NET Framework 4.6.1支持.NET Standard 1.5至2.0，但是使用从.NET Framework 4.6.1项目中的版本构建的.NET Standard库存在一些问题.

While NuGet considers .NET Framework 4.6.1 as supporting .NET Standard 1.5 through 2.0, there are several issues with consuming .NET Standard libraries that were built for those versions from .NET Framework 4.6.1 projects.

对于需要使用此类库的.NET Framework项目，我们建议您将项目升级到目标.NET Framework 4.7.2或更高版本.

For .NET Framework projects that need to use such libraries, we recommend that you upgrade the project to target .NET Framework 4.7.2 or higher.

原始答案

如果您检查源代码 ，您会发现自己的怀疑几乎是正确的.如果buffered为false，则QueryAsync将同步流 .

If you check the source code, you'll see that your suspicion is almost correct. When buffered is false, QueryAsync will stream synchronously.

if (command.Buffered)
{
    var buffer = new List<T>();
    var convertToType = Nullable.GetUnderlyingType(effectiveType) ?? effectiveType;
    while (await reader.ReadAsync(cancel).ConfigureAwait(false))
    {
        object val = func(reader);
        if (val == null || val is T)
        {
            buffer.Add((T)val);
        }
        else
        {
            buffer.Add((T)Convert.ChangeType(val, convertToType, CultureInfo.InvariantCulture));
        }
    }
    while (await reader.NextResultAsync(cancel).ConfigureAwait(false)) { /* ignore subsequent result sets */ }
    command.OnCompleted();
    return buffer;
}
else
{
    // can't use ReadAsync / cancellation; but this will have to do
    wasClosed = false; // don't close if handing back an open reader; rely on the command-behavior
    var deferred = ExecuteReaderSync<T>(reader, func, command.Parameters);
    reader = null; // to prevent it being disposed before the caller gets to see it
    return deferred;
}

正如注释所解释的，当返回类型应为IEnumerable时，则不能使用ReadAsync.这就是为什么必须引入C#8的异步枚举的原因.

As the comment explains, it's not possible to use ReadAsync when the return type is expected to be IEnumerable. That's why C# 8's async enumerables had to be introduced.

ExecuteReaderSync的代码是:

The code for ExecuteReaderSync is :

private static IEnumerable<T> ExecuteReaderSync<T>(IDataReader reader, Func<IDataReader, object> func, object parameters)
{
    using (reader)
    {
        while (reader.Read())
        {
            yield return (T)func(reader);
        }
        while (reader.NextResult()) { /* ignore subsequent result sets */ }
        (parameters as IParameterCallbacks)?.OnCompleted();
    }
}

它使用Read而不是ReadAsync.

C#8异步流将允许对其进行重写以返回IAsyncEnumerable.仅仅更改语言版本并不能解决问题.

C#8 async streams will allow rewriting this to return an IAsyncEnumerable. Simply changing the language version won't solve the problem.

考虑到异步流上的当前文档，它看起来可能像这样:

Given the current docs on async streams this could look like :

private static async IAsyncEnumerable<T> ExecuteReaderASync<T>(IDataReader reader, Func<IDataReader, object> func, object parameters)
{
    using (reader)
    {
        while (await reader.ReadAsync())
        {
            yield return (T)func(reader);
        }

        while (await reader.NextResultAsync(cancel).ConfigureAwait(false)) { /* ignore subsequent result sets */ }
         command.OnCompleted();
        (parameters as IParameterCallbacks)?.OnCompleted();
    }
}

Buuuuuut 异步流是只能在.NET Core上运行的功能之一，并且可能尚未实现.当我尝试在Sharplab.io中写一个时，是Kaboom. [connection lost, reconnecting…]

Buuuuuut async streams is one of the things that can only work on .NET Core, and probably isn't implemented yet. When I tried to write one in Sharplab.io, Kaboom. [connection lost, reconnecting…]

这篇关于如果将IEnumerable与async/await一起使用(使用Dapper从SQL Server中流式传输数据)，会发生什么情况?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！