我如何使`await ...`使用`yield return`(即在迭代器方法中)？

Question

我有现有的代码,看起来类似于:

IEnumerable<SomeClass> GetStuff()
{
    using (SqlConnection conn = new SqlConnection(connectionString))
    using (SqlCommand cmd = new SqlCommand(sql, conn)
    {
        conn.Open();
        SqlDataReader reader = cmd.ExecuteReader();
        while (reader.Read())
        {
            SomeClass someClass = f(reader); // create instance based on returned row
            yield return someClass;
        }
    } 
}

看来我可以通过使用获益reader.ReadAsync().但是,如果我只修改一行:

        while (await reader.ReadAsync())

编译器通知我await只能在标async有的方法中使用,并建议我修改方法签名为:

async Task<IEnumerable<SomeClass>> GetStuff()

但是,这样做会导致GetStuff()无法使用,因为:

body GetStuff()不能是迭代器块,因为Task<IEnumerable<SomeClass>>它不是迭代器接口类型.

我确信我错过了异步编程模型的关键概念.

问题:

• 我可以ReadAsync()在我的迭代器中使用吗？怎么样？
• 我怎样才能以不同的方式思考异步范式,以便了解它在这种情况下的工作原理？

Answer 1

你问的问题实际上并没有多大意义.IEnumerable<T>是一个同步接口,并且返回Task<IEnumerable<T>>对你没什么帮助,因为一些线程必须阻止等待每个项目,无论如何.

您实际想要返回的是一些异步替代方案IEnumerable<T>:类似于IObservable<T>来自TPL Dataflow的数据流块,或者IAsyncEnumerable<T>计划添加到C#8.0/.Net Core 3.0.(与此同时,有一些 库包含它.)

使用TPL Dataflow,一种方法是:

ISourceBlock<SomeClass> GetStuff() {
    var block = new BufferBlock<SomeClass>();

    Task.Run(async () =>
    {
        using (SqlConnection conn = new SqlConnection(connectionString))
        using (SqlCommand cmd = new SqlCommand(sql, conn))
        {
            await conn.OpenAsync();
            SqlDataReader reader = await cmd.ExecuteReaderAsync();
            while (await reader.ReadAsync())
            {
                SomeClass someClass;
                // Create an instance of SomeClass based on row returned.
                block.Post(someClass);
            }
            block.Complete();
        } 
    });

    return block;
}

您可能希望将错误处理添加到上面的代码中,但除此之外,它应该可以工作,并且它将完全异步.

然后,您的其余代码将异步使用返回块中的项,可能使用ActionBlock.

Answer 2

不,您当前不能使用迭代器块的异步.正如斯维克所说,你需要做类似IAsyncEnumerable的事情.

如果你有返回值,Task<IEnumerable<SomeClass>>则意味着该函数返回一个Task对象,一旦完成,它将为你提供一个完全形成的IEnumerable(在这个枚举中没有任何空间用于任务异步).一旦任务对象完成,调用者应该能够同步迭代它在枚举中返回的所有项.

这是一个返回的解决方案Task<IEnumerable<SomeClass>>.通过这样做,你可以获得异步的很大一部分好处:

async Task<IEnumerable<SomeClass>> GetStuff()
{
    using (SqlConnection conn = new SqlConnection(""))
    {
        using (SqlCommand cmd = new SqlCommand("", conn))
        {
            await conn.OpenAsync();
            SqlDataReader reader = await cmd.ExecuteReaderAsync();
            return ReadItems(reader).ToArray();
        }
    }
}

IEnumerable<SomeClass> ReadItems(SqlDataReader reader)
{
    while (reader.Read())
    {
        // Create an instance of SomeClass based on row returned.
        SomeClass someClass = null;
        yield return someClass;
    }
}

......以及一个示例用法:

async void Caller()
{
    // Calls get-stuff, which returns immediately with a Task
    Task<IEnumerable<SomeClass>> itemsAsync = GetStuff();
    // Wait for the task to complete so we can get the items
    IEnumerable<SomeClass> items = await itemsAsync;
    // Iterate synchronously through the items which are all already present
    foreach (SomeClass item in items)
    {
        Console.WriteLine(item);
    }
}

这里有迭代器部分和异步部分,它们允许您同时使用async和yield语法.该GetStuff函数异步获取数据,ReadItems然后同步将数据读入可枚举的数据.

记下ToArray()电话.这样的事情是必要的,因为枚举器函数是懒惰地执行的,所以你的异步函数可能会在读取所有数据之前处理连接和命令.这是因为using块覆盖了Task执行的持续时间,但您将after在任务完成时迭代它.

该解决方案并不能使用ReadAsync,但它确实使用OpenAsync和ExecuteReaderAsync,这可能让你最受益的.根据我的经验,ExecuteReader将占用大部分时间并且最大的好处是异步.当我读到第一行时,SqlDataReader已经拥有所有其他行并且ReadAsync只是同步返回.如果您也是这种情况,那么转移到基于推送的系统IObservable<T>(这将需要对调用函数进行重大修改)将无法获得显着的好处.

为了便于说明,请考虑针对同一问题的替代方法:

IEnumerable<Task<SomeClass>> GetStuff()
{
    using (SqlConnection conn = new SqlConnection(""))
    {
        using (SqlCommand cmd = new SqlCommand("", conn))
        {
            conn.Open();
            SqlDataReader reader = cmd.ExecuteReader();
            while (true)
                yield return ReadItem(reader);
        }
    }
}

async Task<SomeClass> ReadItem(SqlDataReader reader)
{
    if (await reader.ReadAsync())
    {
        // Create an instance of SomeClass based on row returned.
        SomeClass someClass = null;
        return someClass;
    }
    else
        return null; // Mark end of sequence
}

......以及一个示例用法:

async void Caller()
{
    // Synchronously get a list of Tasks
    IEnumerable<Task<SomeClass>> items = GetStuff();
    // Iterate through the Tasks
    foreach (Task<SomeClass> itemAsync in items)
    {
        // Wait for the task to complete. We need to wait for 
        // it to complete before we can know if it's the end of
        // the sequence
        SomeClass item = await itemAsync;
        // End of sequence?
        if (item == null) 
            break;
        Console.WriteLine(item);
    }
}

在这种情况下,GetStuff立即返回枚举,其中可枚举中的每个项都是一个SomeClass在完成时呈现对象的任务.这种方法有一些缺陷.首先,可枚举同步返回,所以在它返回时我们实际上不知道结果中有多少行,这就是为什么我把它变成无限序列的原因.这是完全合法的,但它有一些副作用.我需要使用null在无限的任务序列中发出有用数据的结束信号.其次,你必须要小心你如何迭代它.您需要向前迭代它,并且需要在迭代到下一行之前等待每一行.在完成所有任务后,您还必须仅处理迭代器,以便GC在使用完成之前不会收集连接.由于这些原因,这不是一个安全的解决方案,我必须强调,我将其包括在内以帮助回答您的第二个问题.

Answer 3

从 C# 8 开始，这可以通过IAsyncEnumerable来完成

修改后的代码：

async IAsyncEnumerable<SomeClass> GetStuff()
{
    using (SqlConnection conn = new SqlConnection(connectionString))
    using (SqlCommand cmd = new SqlCommand(sql, conn)
    {
        conn.Open();
        SqlDataReader reader = cmd.ExecuteReader();
        while (reader.Read())
        {
            SomeClass someClass = f(reader); // create instance based on returned row
            yield return someClass;
        }
    } 
}

像这样消费它：

await foreach (var stuff in GetStuff())
    ...