Есть ли способ объединить LINQ и Async?

Question

В основном у меня есть процедура, подобная

var results = await Task.WhenAll(
    from input in inputs
    select Task.Run(async () => await InnerMethodAsync(input))
);
.
.
.
private static async Task<Output> InnerMethodAsync(Input input)
{
    var x = await Foo(input);
    var y = await Bar(x);
    var z = await Baz(y);
    return z;
}

, и мне интересно, есть ли причудливый способ объединить это в один запрос LINQ, который похож на «асинхронный поток» (лучший способ, которым я могу описать это).

Answer 1

Довольно очевидный ответ, но у вас есть , только что использованные LINQ и async вместе - вы используете select LINQ для проецирования и запуска группы асинхронных задач, а затемawait результатов, что обеспечивает асинхронный шаблон параллелизма.

Хотя вы, вероятно, только что предоставили пример, в вашем коде есть несколько моментов, на которые следует обратить внимание (я переключился на синтаксис Lambda,но применяются те же принципы)

• Поскольку в каждой задаче перед первой await (т.е. до этой работы не было выполнено никакой операции) нет реальной причины , чтобы использовать Task.Run здесь.
• И поскольку в лямбде после вызова InnerMethodAsync не нужно выполнять никаких действий, вам не нужно упаковывать вызовы InnerMethodAsync в асинхронную лямбду (но будьте осторожныиз IDisposable)

т.е. Вы можете просто выбрать Task, возвращенный из InnerMethodAsync, и ждать их с помощью Task.WhenAll.

var tasks = inputs
    .Select(input => InnerMethodAsync(input)) // or just .Select(InnerMethodAsync);

var results = await Task.WhenAll(tasks);

Возможны более сложные шаблоныс асинхронностью и Linq, но вместо того, чтобы заново изобретать колесо, вы должны взглянуть на Reactive Extensions и библиотеку потоков данных TPL , которые имеют множество строительных блоков для сложных потоков.

Answer 2

Попробуйте использовать Microsoft Reactive Framework.Тогда вы можете сделать это:

IObservable<Output[]> query =
    from input in inputs.ToObservable()
    from x in Observable.FromAsync(() => Foo(input))
    from y in Observable.FromAsync(() => Bar(x))
    from z in Observable.FromAsync(() => Baz(y))
    select z;

Output[] results = await query.ToArray();

Простой.

Просто NuGet "System.Reactive" и добавьте using System.Reactive.Linq; в ваш код.

Answer 3

Когда вы используете LINQ, обычно есть две части: создание и итерация.

Создание:

var query = list.Select( a => a.Name);

Эти вызовы всегда синхронны.Но этот код не делает намного больше, чем создает объект, который предоставляет IEnumerable.Фактическая работа не будет выполнена позже, из-за паттерна, называемого отложенное выполнение .

Итерация:

var results = query.ToList();

Этот код принимает перечислимое значение и получает значение каждого элемента, что обычно включает вызов ваших делегатов обратного вызова (в данном случае a => a.Name).Это та часть, которая потенциально дорогая и может извлечь выгоду из асинхронности, например, если ваш обратный вызов - что-то вроде async a => await httpClient.GetByteArrayAsync(a).

Так что это итерационная часть, в которой мы заинтересованы, если мы хотим сделать этоasync.

Проблема здесь в том, что ToList() (и большинство других методов, которые вызывают итерацию, например Any() или Last()) не являются асинхронными методами, поэтому ваш делегат обратного вызова будет вызываться синхронно,и вы получите список задач вместо нужных вам данных.

Мы можем обойти это с помощью такого кода:

public static class ExtensionMethods
{
    static public async Task<List<T>> ToListAsync<T>(this IEnumerable<Task<T>> This)
    {
        var tasks = This.ToList();     //Force LINQ to iterate and create all the tasks. Tasks always start when created.
        var results = new List<T>();   //Create a list to hold the results (not the tasks)
        foreach (var item in tasks)
        {
            results.Add(await item);   //Await the result for each task and add to results list
        }
        return results;
    }
}

С помощью этого метода расширения, мы можем переписать ваш код:

var results = await inputs.Select( async i => await InnerMethodAsync(i) ).ToListAsync();

^ Это должно дать вам асинхронное поведение, которое вы ищете, и избежать создания задач пула потоков, как в вашем примере.

Примечание:Если вы используете LINQ-to-entity, дорогая часть (получение данных) вам не доступна.Для LINQ-to-entity вы можете использовать ToListAsync () , который поставляется вместе с EF Framework.

Попробуйте и посмотрите временные характеристики в моей демонстрации на DotNetFiddle .