并行运行异步方法

Question

我有一个异步方法,GetExpensiveThing()它执行一些昂贵的I/O工作.这就是我使用它的方式:

// Serial execution
public async Task<List<Thing>> GetThings()
{
    var first = await GetExpensiveThing();
    var second = await GetExpensiveThing();
    return new List<Thing>() { first, second };
}

但由于这是一种昂贵的方法,我想并行执行这些调用.我本以为移动等待会解决这个问题:

// Serial execution
public async Task<List<Thing>> GetThings()
{
    var first = GetExpensiveThing();
    var second = GetExpensiveThing();
    return new List<Thing>() { await first, await second };
}

这不起作用,所以我将它们包装在一些任务中,这有效:

// Parallel execution
public async Task<List<Thing>> GetThings()
{
    var first = Task.Run(() =>
    {
        return GetExpensiveThing();
    });

    var second = Task.Run(() =>
    {
        return GetExpensiveThing();
    });

    return new List<Thing>() { first.Result, second.Result };
}

我甚至尝试过在任务中和周围等待和异步,但它真的让人感到困惑,我没有运气.

是否有更好的并行运行异步方法,或者任务是一个好方法？

Answer 1

是否有更好的并行运行异步方法,或者任务是一个好方法？

是的,"最佳"方法是利用该Task.WhenAll方法.但是,您的第二种方法应该并行运行.我已经创建了一个.NET Fiddle,这应该有助于解决问题.你的第二种方法实际上应该并行运行.我的小提琴证明了这一点!

考虑以下:

public Task<Thing[]> GetThingsAsync()
{
    var first = GetExpensiveThingAsync();
    var second = GetExpensiveThingAsync();

    return Task.WhenAll(first, second);
}

注意

最好是使用"异步"后缀,而不是GetThings和GetExpensiveThing-我们应该有GetThingsAsync和GetExpensiveThingAsync分别- 源.

Answer 2

Task.WhenAll() 在同时执行大量/大量任务时，往往会表现不佳-不会适度/节流。

如果您要在列表中执行许多任务并希望等待最终结果，那么我建议使用partition对并行度有限制的方法。

我已经修改了Stephen Toub的博客对现代LINQ的优雅处理方法：

public static Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, Task> funcBody, int maxDoP = 4)
{
    async Task AwaitPartition(IEnumerator<T> partition)
    {
        using (partition)
        {
            while (partition.MoveNext())
            { await funcBody(partition.Current); }
        }
    }

    return Task.WhenAll(
        Partitioner
            .Create(source)
            .GetPartitions(maxDoP)
            .AsParallel()
            .Select(p => AwaitPartition(p)));
}

它的工作方式很简单，采用IEnumerable-将其分解为均匀的分区，并同时针对每个分区中的每个元素触发函数/方法。任何时候每个分区中的元素不超过一个，但是n个分区中的n个任务。

扩展用法：

await myList.ParallelForEachAsync(myFunc, Environment.ProcessorCount);

编辑：如果需要更多选择，我现在在Github 的存储库中保留一些重载。NetStandard也在NuGet中。