c# - 将 IAsyncEnumerable 与 Dapper 一起使用

Question

我们最近迁移了使用 Dapper 的 ASP.NET Core API。到 .NET Core 3.1。迁移后，我们觉得有机会使用最新的IAsyncEnumerable来自 C# 8 的功能对于我们的端点之一。

这是更改前的伪代码:

public async Task<IEnumerable<Item>> GetItems(int id)
{
    var reader = await _connection.QueryMultipleAsync(getItemsSql,
       param: new
       {
           Id = id
       });

    var idFromDb = (await reader.ReadAsync<int?>().ConfigureAwait(false)).SingleOrDefault();
    if (idFromDb == null)
    {
       return null;
    }

    var items = await reader.ReadAsync<Item>(buffered: false).ConfigureAwait(false);

    return Stream(reader, items);
} 

private IEnumerable<Item> Stream(SqlMapper.GridReader reader, IEnumerable<Item> items)
{
    using (reader)
    {
        foreach (var item in items)
        {
            yield return item;
        }
    }     
}

后 IAsyncEnumerable代码更改:

// Import Nuget pacakage: System.Linq.Async

public async Task<IAsyncEnumerable<Item>> GetItems(int id)
{
    var reader = await _connection.QueryMultipleAsync(getItemsSql,
       param: new
       {
           Id = id
       });

    var idFromDb = (await reader.ReadAsync<int?>().ConfigureAwait(false)).SingleOrDefault();
    if (idFromDb == null)
    {
        return null;
    }

    var items = await reader.ReadAsync<Item>(buffered: false).ConfigureAwait(false);

    return Stream(reader, items);
} 

private IAsyncEnumerable<Item> Stream(SqlMapper.GridReader reader, IEnumerable<Item> items)
{
    using (reader)
    {
       await foreach (var item in items.ToAsyncEnumerable())
       {
           yield return item;
       }
    }
 }

上面的方法是使用 ToAsyncEnumerable灵感来自 this post ，但我不是 100% 确定我是否在正确的地方/上下文中使用它。

题:

dapper 库只返回 IEnumerable但是我们可以使用 ToAsyncEnumerable将其转换为 IAsyncEnumerable为 async stream像上面一样？

备注 : 这个问题看起来类似于 What happens with returning IEnumerable if used with async/await (streaming data from SQL Server with Dapper)?但我认为这不能回答我的问题。

Answer 1

更新:当我第一次写这个答案时，我不知道异步迭代器。感谢 Theodor Zoulias 指出这一点。有鉴于此，可以采用更简单的方法:

using var reader = await connection.ExecuteReaderAsync(query, parameters);
var rowParser = reader.GetRowParser<T>();

while (await reader.ReadAsync()) {
    yield return rowParser(reader);
}

原答案:
这是一个 IAsyncEnumerable我写的包装器可以帮助那些想要使用 async/await 流式传输无缓冲数据并且还想要 Dapper 类型映射的强大功能的人:

public class ReaderParser<T> : IAsyncEnumerable<T> {
    public ReaderParser(SqlDataReader reader) {
        Reader = reader;
    }
    private SqlDataReader Reader { get; }
    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default) {
        return new ReaderParserEnumerator<T>(Reader);
    }
}
public class ReaderParserEnumerator<T> : IAsyncEnumerator<T> {
    public ReaderParserEnumerator(SqlDataReader reader) {
        Reader = reader;
        RowParser = reader.GetRowParser<T>();
    }
    public T Current => Reader.FieldCount == 0 ? default(T) : RowParser(Reader);
    private SqlDataReader Reader { get; }
    private Func<IDataReader, T> RowParser { get; }
    public async ValueTask DisposeAsync() {
        await Reader.DisposeAsync();
    }
    public async ValueTask<bool> MoveNextAsync() {
        return await Reader.ReadAsync();
    }
}

用法:

var reader = await command.ExecuteReaderAsync();
return new ReaderParser<T>(reader);

然后，打包 System.Linq.Async基本上添加了所有好的 IEnumerable你知道和喜欢的扩展，例如在我的用法中:

var streamData = await repo.GetDataStream();
var buffer = await streamData.Take(BATCH_SIZE).ToListAsync();