如何编写异步LINQ查询？

Question

在我读了一堆LINQ相关的东西之后,我突然意识到没有文章介绍如何编写异步LINQ查询.

假设我们使用LINQ to SQL,下面的语句很清楚.但是,如果SQL数据库响应缓慢,则使用此代码块的线程将受到阻碍.

var result = from item in Products where item.Price > 3 select item.Name;
foreach (var name in result)
{
    Console.WriteLine(name);
}

似乎当前的LINQ查询规范不提供对此的支持.

有没有办法做LINQ异步编程？当结果准备好使用而没有I/O上的任何阻塞延迟时,它就像有一个回调通知.

Answer 1

虽然LINQ本身并没有这个,但框架本身确实......你可以轻松地将你自己的异步查询执行器滚动到30行左右......事实上,我只是为你扔了这个:)

编辑:通过写这篇文章,我发现了为什么他们没有实现它.它无法处理匿名类型,因为它们是本地范围的.因此,您无法定义回调函数. 这是一个非常重要的事情,因为很多linq到sql的东西在select子句中创建它们.以下任何一个建议遭受同样的命运,所以我仍然认为这个是最容易使用的!

编辑:唯一的解决方案是不使用匿名类型.您可以将回调声明为仅使用IEnumerable(无类型args),并使用反射来访问字段(ICK !!).另一种方法是将回调声明为"动态"......哦......等等......那还没有结束.:)这是动态如何使用的另一个体面的例子.有些人可能称之为滥用.

把这个放在你的实用程序库中:

public static class AsynchronousQueryExecutor
{
    public static void Call<T>(IEnumerable<T> query, Action<IEnumerable<T>> callback, Action<Exception> errorCallback)
    {
        Func<IEnumerable<T>, IEnumerable<T>> func =
            new Func<IEnumerable<T>, IEnumerable<T>>(InnerEnumerate<T>);
        IEnumerable<T> result = null;
        IAsyncResult ar = func.BeginInvoke(
                            query,
                            new AsyncCallback(delegate(IAsyncResult arr)
                            {
                                try
                                {
                                    result = ((Func<IEnumerable<T>, IEnumerable<T>>)((AsyncResult)arr).AsyncDelegate).EndInvoke(arr);
                                }
                                catch (Exception ex)
                                {
                                    if (errorCallback != null)
                                    {
                                        errorCallback(ex);
                                    }
                                    return;
                                }
                                //errors from inside here are the callbacks problem
                                //I think it would be confusing to report them
                                callback(result);
                            }),
                            null);
    }
    private static IEnumerable<T> InnerEnumerate<T>(IEnumerable<T> query)
    {
        foreach (var item in query) //the method hangs here while the query executes
        {
            yield return item;
        }
    }
}

你可以像这样使用它:

class Program
{

    public static void Main(string[] args)
    {
        //this could be your linq query
        var qry = TestSlowLoadingEnumerable();

        //We begin the call and give it our callback delegate
        //and a delegate to an error handler
        AsynchronousQueryExecutor.Call(qry, HandleResults, HandleError);

        Console.WriteLine("Call began on seperate thread, execution continued");
        Console.ReadLine();
    }

    public static void HandleResults(IEnumerable<int> results)
    {
        //the results are available in here
        foreach (var item in results)
        {
            Console.WriteLine(item);
        }
    }

    public static void HandleError(Exception ex)
    {
        Console.WriteLine("error");
    }

    //just a sample lazy loading enumerable
    public static IEnumerable<int> TestSlowLoadingEnumerable()
    {
        Thread.Sleep(5000);
        foreach (var i in new int[] { 1, 2, 3, 4, 5, 6 })
        {
            yield return i;
        }
    }

}

现在就把它放在我的博客上,非常方便.

Answer 2

TheSoftwareJedi和ulrikb(又名user316318)解决方案适用于任何LINQ类型,但(如Chris Moschini所指出的 )不会委托利用Windows I/O完成端口的底层异步调用.

Wesley Bakker的Asynchronous DataContext帖子(由Scott Hanselman的博客文章触发)描述了使用sqlCommand.BeginExecuteReader/sqlCommand.EndExecuteReader的LINQ to SQL类,它利用了Windows I/O完成端口.

I/O完成端口提供了一种有效的线程模型,用于处理多处理器系统上的多个异步I/O请求.

Answer 3

根据Michael Freidgeim的回答并提到Scott Hansellman发表的博客文章以及可以使用async/的事实await,您可以实现可重用的ExecuteAsync<T>(...)方法,该方法SqlCommand异步执行底层:

protected static async Task<IEnumerable<T>> ExecuteAsync<T>(IQueryable<T> query,
    DataContext ctx,
    CancellationToken token = default(CancellationToken))
{
    var cmd = (SqlCommand)ctx.GetCommand(query);

    if (cmd.Connection.State == ConnectionState.Closed)
        await cmd.Connection.OpenAsync(token);
    var reader = await cmd.ExecuteReaderAsync(token);

    return ctx.Translate<T>(reader);
}

然后你可以(重新)使用它:

public async Task WriteNamesToConsoleAsync(string connectionString, CancellationToken token = default(CancellationToken))
{
    using (var ctx = new DataContext(connectionString))
    {
        var query = from item in Products where item.Price > 3 select item.Name;
        var result = await ExecuteAsync(query, ctx, token);
        foreach (var name in result)
        {
            Console.WriteLine(name);
        }
    }
}