将 async/await 与 DataReader 一起使用？ （没有中间缓冲区！）

我的目标很简单，我想做异步 I/O 调用（使用 async wait） - 但是：

• 不使用 DataFlow 依赖项（就像这个答案一样 https://stackoverflow.com/a/13148683/859154)
• 没有中间缓冲区（不喜欢这个答案 https://stackoverflow.com/a/13148683/859154)
• Projector 函数应作为参数发送。 (不喜欢这个答案 https://stackoverflow.com/a/13149492/859154)

Ok.

目前这是我的代码，它的工作是从数据库读取并将每一行投影到Func<>

public IEnumerable < T > GetSomeData < T > (string sql, Func < IDataRecord, T > projector)
{
    using(SqlConnection _conn = new SqlConnection(@"Data Source=..."))
    {
        using(SqlCommand _cmd = new SqlCommand(sql, _conn))
        {
            _conn.Open();
            _cmd.CommandTimeout = 100000;
            using(IDataReader rdr = _cmd.ExecuteReader())
            {
                while (rdr.Read())  yield    return projector(rdr);
            }
        }
    }
}

那么，什么是投影仪？

每个类都有一个函数，它得到一个record ( IDataRecord）并创建一个实体：

例子 ：

public class MyClass
{
    public static MyClass MyClassFactory(IDataRecord record)
    {
        return new MyClass
        {
            Name = record["Name"].ToString(),
            Datee = DateTime.Parse(record["Datee"].ToString()),
            val = decimal.Parse(record["val"].ToString())
        };
    }
    public string Name    {   get;   set;  }
    public DateTime Datee    {  get;     set;  }
    public decimal val    {  get;    set;    }
}

所以在这里 ，MyClassFactory将是Func

那么我目前如何运行它呢？

 var sql = @"SELECT TOP 1000 [NAME],[datee] ,[val]  FROM [WebERP].[dbo].[t]";
 var a = GetSomeData < MyClass > (sql, MyClass.MyClassFactory).Where(...); //notice the Func

All ok.

问题现在开始：

Adding async该方法会产生错误:( 是的，我知道 Ienumerable 是一个同步接口因此出现问题）

public async Task<IEnumerable < T >> GetSomeData < T > (string sql, Func < IDataRecord, T > projector)

不能是迭代器块，因为 '系统.线程.任务.任务>' 不是迭代器接口类型

但这里的这个人却做到了 https://stackoverflow.com/a/13149492/859154 - :

Which DOES编译。

Question

我怎样才能转换my code支持完全异步IO调用？

（在条件下：没有数据流依赖，发送投影仪函数作为参数，没有中间缓冲区）

我想做异步 I/O 调用（使用异步等待） - 但是：

• 不使用 DataFlow 依赖项（如本答案所示）
• 没有中间缓冲区（不像这个答案）
• Projector 函数应作为参数发送。 （不喜欢这个答案）

您可能想查看 Stephen Toub 的“任务、Monad 和 LINQ” http://blogs.msdn.com/b/pfxteam/archive/2013/04/03/tasks-monads-and-linq.aspx有关如何处理异步数据序列的一些好主意。

（尚）不可能合并yield and await，但我要在这里做一个口头表达者：引用的要求没有列出IEnumerable和 LINQ。因此，这是一个可能的解决方案，形状为两个协程（几乎未经测试）。

数据生成例程（对应于IEnumarable with yield):

public async Task GetSomeDataAsync<T>(
    string sql, Func<IDataRecord, T> projector, ProducerConsumerHub<T> hub)
{
    using (SqlConnection _conn = new SqlConnection(@"Data Source=..."))
    {
        using (SqlCommand _cmd = new SqlCommand(sql, _conn))
        {
            await _conn.OpenAsync();
            _cmd.CommandTimeout = 100000;
            using (var rdr = await _cmd.ExecuteReaderAsync())
            {
                while (await rdr.ReadAsync())
                    await hub.ProduceAsync(projector(rdr));
            }
        }
    }
}

数据消费者例程（对应于foreach或 LINQ 表达式）：

public async Task ConsumeSomeDataAsync(string sql)
{
    var hub = new ProducerConsumerHub<IDataRecord>();
    var producerTask = GetSomeDataAsync(sql, rdr => rdr, hub);

    while (true)
    {
        var nextItemTask = hub.ConsumeAsync();
        await Task.WhenAny(producerTask, nextItemTask);

        if (nextItemTask.IsCompleted)
        {
            // process the next data item
            Console.WriteLine(await nextItemTask);
        }

        if (producerTask.IsCompleted)
        {
            // process the end of sequence
            await producerTask;
            break;
        }
    }
}

协程执行助手（也可以实现为一对定制服务员 http://blogs.msdn.com/b/pfxteam/archive/2011/01/13/10115642.aspx):

public class ProducerConsumerHub<T>
{
    TaskCompletionSource<Empty> _consumer = new TaskCompletionSource<Empty>();
    TaskCompletionSource<T> _producer = new TaskCompletionSource<T>();

    // TODO: make thread-safe
    public async Task ProduceAsync(T data)
    {
        _producer.SetResult(data);
        await _consumer.Task;
        _consumer = new TaskCompletionSource<Empty>();
    }

    public async Task<T> ConsumeAsync()
    {
        var data = await _producer.Task;
        _producer = new TaskCompletionSource<T>();
        _consumer.SetResult(Empty.Value);
        return data;
    }

    struct Empty { public static readonly Empty Value = default(Empty); }
}

这只是一个想法。对于像这样的简单任务来说，这可能有点大材小用，并且可以在某些方面进行改进（例如线程安全、竞争条件以及在不接触序列的情况下处理序列的结尾）producerTask）。然而，它说明了异步数据检索和处理如何可能解耦。