并行运行异步方法

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 };
}

但由于这是一种昂贵的方法，我想并行执行这些调用。我原以为移动 awaits 会解决这个问题：

// 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 };
}

我什至尝试在任务中和任务周围使用 await 和 async，但它真的很混乱，我没有运气。

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

Answer 1

并行运行异步方法是否更好，或者任务是一种好方法？

是的，“最佳”方法是使用Task.WhenAll 方法。但是，您的第二种方法应该并行运行。我创建了一个.NET Fiddle，这应该有助于阐明一些观点。您的第二种方法实际上应该并行运行。我的小提琴证明了这一点！

考虑以下几点：

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

    return Task.WhenAll(first, second);
}

注意

最好使用“Async”后缀，而不是 GetThings 和 GetExpensiveThing - 我们应该分别使用 GetThingsAsync 和 GetExpensiveThingAsync - source。

Answer 2

您可以使用Task.WhenAll，它会在所有相关任务完成后返回

检查此问题here 以供参考

Answer 3

如果GetExpensiveThing 是正确 异步的（意味着它不会同步执行任何 IO 或 CPU 工作），那么调用这两种方法然后等待结果的第二种解决方案应该有效。你也可以使用Task.WhenAll。

但是，如果不是，您可以通过将每个任务发布到线程池并使用 Task.WhenAll 组合器来获得更好的结果，例如：

public Task<IList<Thing>> GetThings() =>
    Task.WhenAll(Task.Run(() => GetExpensiveThing()), Task.Run(() => GetExpensiveThing()));

（请注意，我将返回类型更改为IList 以完全避免awaits。）

您应该避免使用Result 属性。它会导致调用者线程阻塞并等待任务完成，这与使用延续的await 或Task.WhenAll 不同。

Answer 4

Task.WhenAll() 倾向于在同时触发大规模/大量任务时表现不佳 - 没有节制/节流。

如果您正在执行列表中的大量任务并希望等待最终结果，那么我建议使用 partition 并限制并行度。

我已经修改了Stephen Toub's blog 优雅的现代 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 Task.Yield(); // prevents a sync/hot thread hangup
                 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 上的 repository 中保留一些重载。对于 NetStandard，它也在 NuGet 中。

编辑 2：感谢下面 Theodor 的 cmets，我能够通过使用 await Task.Yield(); 来缓解编写不佳的异步任务阻塞并行性。