Как написать асинхронный запрос 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? Он работает так, как будто есть обратный вызов , когда результаты готовы к использованию без какой-либо задержки блокировки ввода-вывода.

Answer 1

Хотя LINQ действительно не имеет этого сам по себе, сам фреймворк ... Вы можете легко свернуть собственный асинхронный исполнитель запросов в 30 строк или так ... На самом деле, я просто бросил это вместе для вас:)

EDIT: Благодаря написанию этого, я обнаружил, почему они не реализовали его. Он не может обрабатывать анонимные типы, поскольку они локализованы локально. Таким образом, у вас нет способа определить вашу функцию обратного вызова. Это довольно важная вещь, поскольку многие элементы linq to sql создают их в предложении select. Любые из приведенных ниже предложений страдают той же судьбой, поэтому я считаю, что этот самый простой в использовании!

EDIT: Единственное решение - не использовать анонимные типы. Вы можете объявить обратный вызов, просто принимая IEnumerable (без аргументов типа), и используйте отражение для доступа к полям (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 «s (ака user316318) решения хороши для любого типа LINQ, но (как указано на Chris Moschini) не делегируя подстилающей асинхронных вызовов, что порты завершения/O рычаги Windows я.

Asynchronous DataContext пост Уэсли Баккер (в вызвано a blog post of Scott Hanselman) описывают класс для LINQ к SQL, который использует sqlCommand.BeginExecuteReader/sqlCommand.EndExecuteReader, какие рычаги порты завершения ввода/вывода Windows.

I/O completion ports обеспечивает эффективную модель потоков для обработки нескольких асинхронных запросов ввода-вывода в многопроцессорной системе.

Answer 3

Я начал простой проект github с именем Asynq, чтобы выполнить асинхронное выполнение запроса LINQ-to-SQL. Идея довольно проста, хотя «хрупкий» на данном этапе (по состоянию на 8/16/2011):

1. Пусть LINQ к SQL делать «тяжелую» работу по переводу вашей IQueryable в DbCommand через DataContext.GetCommand() ,
2. Для SQL 200 [058], отлитый от абстрактного DbCommand экземпляр, который вы получили от GetCommand(), чтобы получить SqlCommand. Если вы используете SQL CE, вам не повезло, так как SqlCeCommand не раскрывает шаблон асинхронизации для BeginExecuteReader и EndExecuteReader.
3. Использование BeginExecuteReader и EndExecuteReaderSqlCommand от использования стандартных рамок .NET асинхронного ввода/вывода, чтобы получить себе в завершение обратного вызова делегата, который вы передаете методу BeginExecuteReader в DbDataReader.
4. Теперь у нас есть DbDataReader, который мы понятия не имеем, какие столбцы он содержит, и как сопоставить эти значения с IQueryableElementType (скорее всего, будет анонимным типом в случае объединения). Несомненно, на этом этапе вы можете вручную написать собственный редактор столбцов, который материализует свои результаты обратно в ваш анонимный тип или что-то еще. Вам нужно будет написать новый по каждому типу результата запроса, в зависимости от того, как LINQ-to-SQL обрабатывает ваш IQueryable и какой код SQL он генерирует. Это довольно неприятный вариант, и я не рекомендую его, так как он не поддерживается и не всегда будет правильным. LINQ-to-SQL может изменить форму запроса в зависимости от значений параметров, которые вы передаете, например, query.Take(10).Skip(0) создает разные SQL, чем query.Take(10).Skip(10), и, возможно, другую схему результирующего набора. Лучше всего решить эту проблему материализации:
5. «Повторно реализовать» упрощающий материализатор объектов времени выполнения, который вытягивает столбцы с DbDataReader в определенном порядке в соответствии с атрибутами отображения LINQ-to-SQL для типа ElementType. Тип для IQueryable. Реализация этого правильно, вероятно, является самой сложной частью этого решения.

Как и другие обнаружили, метод DataContext.Translate() не обрабатывает анонимные типы и может отображать только DbDataReader непосредственно к правильно отнести LINQ-к-SQL прокси-объекта. Поскольку большинство запросов, которые стоит написать в LINQ, будут связаны с сложными объединениями, которые неизбежно в конечном итоге требуют анонимных типов для окончательного предложения select, довольно бессмысленно использовать этот предоставленный метод DataContext.Translate().

Есть несколько незначительных недостатков этого решения при возможности использования имеющегося зрелого LINQ-to-SQL IQueryable поставщика:

1. Вы не можете сопоставить один экземпляр объекта для нескольких свойств анонимного типа в окончательном выборе пункта из ваш IQueryable, например from x in db.Table1 select new { a = x, b = x }. LINQ-to-SQL внутренне отслеживает, какие координаты столбцов сопоставляются с какими свойствами; он не предоставляет эту информацию конечному пользователю, поэтому вы не представляете, какие столбцы в DbDataReader повторно используются и которые являются «отличными».
2. Вы не можете включать постоянные значения в свой окончательный вариант предложения - они не могут быть переведены в SQL и будут отсутствовать в DbDataReader, поэтому вам нужно будет построить пользовательскую логику, чтобы вытащить эти постоянные значения из дерева Expression , что было бы довольно хлопот и просто не оправдано.

Я уверен, что есть другие шаблоны запросов, которые могут сломаться, но это два самых больших, о которых я мог думать, что может вызвать проблемы в существующем уровне доступа к данным LINQ-to-SQL.

Эти проблемы легко победить - просто не делайте их в своих запросах, так как ни один шаблон не дает никакой пользы от конечного результата запроса. Надеюсь, этот совет применим ко всем шаблонам запросов, которые потенциально могут вызвать проблемы материализации материализации: -P.Трудно решить проблему отсутствия доступа к информации сопоставления столбцов LINQ-to-SQL.

Более «полный» подход к решению проблемы будет состоять в том, чтобы эффективно перепрофилировать почти все LINQ-to-SQL, что является немного более трудоемким: -P. Исходя из качества, реализация провайдера LINQ-to-SQL с открытым исходным кодом была бы хорошим выбором. Причина, по которой вам нужно будет переопределить его, - это доступ ко всей информации сопоставления столбцов, используемой для материализации результатов DbDataReader, вплоть до экземпляра объекта без потери информации.

Answer 4

на основе Michael Freidgeim's answer и упомянул blog post from 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); 
     } 
    } 
}