使用异步 lambda 的并行 foreach

问题描述 投票：0回答：10

我想并行处理一个集合，但我在实现它时遇到了困难，因此我希望得到一些帮助。

如果我想在并行循环的 lambda 内调用 C# 中标记为异步的方法，就会出现麻烦。例如：

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

问题发生在计数为 0 时，因为创建的所有线程实际上只是后台线程，并且 

Parallel.ForEach
调用不会等待完成。如果我删除 async 关键字，该方法将如下所示：

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, item =>
{
  // some pre stuff
  var responseTask = await GetData(item);
  responseTask.Wait();
  var response = responseTask.Result;
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

它可以工作，但它完全禁用了等待的聪明才智，我必须做一些手动异常处理..（为了简洁而删除）。

如何实现在 lambda 中使用 wait 关键字的 

Parallel.ForEach
循环？可以吗？

Parallel.ForEach 方法的原型采用 

Action<T>
作为参数，但我希望它等待我的异步 lambda。

c# asynchronous async-await task-parallel-library parallel.foreach
10个回答
322
投票

如果你只想要简单的并行性，你可以这样做：

var bag = new ConcurrentBag<object>();
var tasks = myCollection.Select(async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
});
await Task.WhenAll(tasks);
var count = bag.Count;

如果您需要更复杂的东西，请查看 Stephen Toub 的 

ForEachAsync
帖子。

147
投票

您可以使用 

AsyncEnumerator NuGet Package
中的 ParallelForEachAsync 扩展方法：

using Dasync.Collections;

var bag = new ConcurrentBag<object>();
await myCollection.ParallelForEachAsync(async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
}, maxDegreeOfParallelism: 10);
var count = bag.Count;

免责声明：我是 AsyncEnumerator 库的作者，该库是开源的并在 MIT 下获得许可，我发布此消息只是为了帮助社区。

127
投票

新的 .NET 6 API 之一是 Parallel.ForEachAsync，这是一种安排异步工作的方法，允许您控制并行度：

var urls = new [] 
{
    "https://dotnet.microsoft.com",
    "https://www.microsoft.com",
    "https://stackoverflow.com"
};

var client = new HttpClient();

var options = new ParallelOptions { MaxDegreeOfParallelism = 2 };
await Parallel.ForEachAsync(urls, options, async (url, token) =>
{
    var targetPath = Path.Combine(Path.GetTempPath(), "http_cache", url);

    var response = await client.GetAsync(url);

    if (response.IsSuccessStatusCode)
    {
        using var target = File.OpenWrite(targetPath);

        await response.Content.CopyToAsync(target);
    }
});

另一个例子在 Scott Hanselman 的博客中。

来源，供参考。

37
投票

通过

SemaphoreSlim
可以实现并行度控制。

var bag = new ConcurrentBag<object>();
var maxParallel = 20;
var throttler = new SemaphoreSlim(initialCount: maxParallel);
var tasks = myCollection.Select(async item =>
{
  await throttler.WaitAsync();
  try
  {
     var response = await GetData(item);
     bag.Add(response);
  }
  finally
  {
     throttler.Release();
  }
});
await Task.WhenAll(tasks);
var count = bag.Count;
16
投票

从其他答案和接受的 asnwer 引用的文章中编译的最简单的扩展方法：

public static async Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, Task> asyncAction, int maxDegreeOfParallelism)
{
    var throttler = new SemaphoreSlim(initialCount: maxDegreeOfParallelism);
    var tasks = source.Select(async item =>
    {
        await throttler.WaitAsync();
        try
        {
            await asyncAction(item).ConfigureAwait(false);
        }
        finally
        {
            throttler.Release();
        }
    });
    await Task.WhenAll(tasks);
}

更新：这是一个简单的修改，还支持评论中请求的取消令牌（未经测试）

public static async Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, CancellationToken, Task> asyncAction, int maxDegreeOfParallelism, CancellationToken cancellationToken)
{
    var throttler = new SemaphoreSlim(initialCount: maxDegreeOfParallelism);
    var tasks = source.Select(async item =>
    {
        await throttler.WaitAsync(cancellationToken);
        if (cancellationToken.IsCancellationRequested) return;

        try
        {
            await asyncAction(item, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            throttler.Release();
        }
    });
    await Task.WhenAll(tasks);
}
5
投票

我的 ParallelForEach 异步轻量级实现。

特点：

  1. 节流（最大并行度）。
  2. 异常处理（完成时会抛出聚合异常）。
  3. 内存高效（无需存储任务列表）。

public static class AsyncEx
{
    public static async Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, Task> asyncAction, int maxDegreeOfParallelism = 10)
    {
        var semaphoreSlim = new SemaphoreSlim(maxDegreeOfParallelism);
        var tcs = new TaskCompletionSource<object>();
        var exceptions = new ConcurrentBag<Exception>();
        bool addingCompleted = false;

        foreach (T item in source)
        {
            await semaphoreSlim.WaitAsync();
            asyncAction(item).ContinueWith(t =>
            {
                semaphoreSlim.Release();

                if (t.Exception != null)
                {
                    exceptions.Add(t.Exception);
                }

                if (Volatile.Read(ref addingCompleted) && semaphoreSlim.CurrentCount == maxDegreeOfParallelism)
                {
                    tcs.TrySetResult(null);
                }
            });
        }

        Volatile.Write(ref addingCompleted, true);
        await tcs.Task;
        if (exceptions.Count > 0)
        {
            throw new AggregateException(exceptions);
        }
    }
}

使用示例：

await Enumerable.Range(1, 10000).ParallelForEachAsync(async (i) =>
{
    var data = await GetData(i);
}, maxDegreeOfParallelism: 100);
1
投票

我为此创建了一个扩展方法，它利用 SemaphoreSlim 并允许设置最大并行度

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxDegreeOfParallelism">Optional, An integer that represents the maximum degree of parallelism,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxDegreeOfParallelism = null)
    {
        if (maxDegreeOfParallelism.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxDegreeOfParallelism.Value, maxDegreeOfParallelism.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

用法示例：

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
1
投票

在接受的答案中，不需要 ConcurrentBag。 这是没有它的实现：

var tasks = myCollection.Select(GetData).ToList();
await Task.WhenAll(tasks);
var results = tasks.Select(t => t.Result);

任何“//一些前置内容”和“//一些后置内容”都可以进入 GetData 实现（或调用 GetData 的另一个方法）

除了更短之外，没有使用“async void”lambda，这是一种反模式。

0
投票

以下内容设置为与 

IAsyncEnumerable
一起使用，但可以修改为使用 
IEnumerable
，只需更改类型并删除 
foreach
上的“等待”即可。它比创建无数并行任务然后等待所有任务更适合大量数据。

    public static async Task ForEachAsyncConcurrent<T>(this IAsyncEnumerable<T> enumerable, Func<T, Task> action, int maxDegreeOfParallelism, int? boundedCapacity = null)
    {
        ActionBlock<T> block = new ActionBlock<T>(
           action, 
           new ExecutionDataflowBlockOptions 
           { 
             MaxDegreeOfParallelism = maxDegreeOfParallelism, 
             BoundedCapacity = boundedCapacity ?? maxDegreeOfParallelism * 3 
           });

        await foreach (T item in enumerable)
        {
           await block.SendAsync(item).ConfigureAwait(false);
        }

        block.Complete();
        await block.Completion;
    }
-2
投票

对于更简单的解决方案（不确定是否是最佳解决方案），您可以简单地将 

Parallel.ForEach
嵌套在 
Task
中 - 如此

var options = new ParallelOptions { MaxDegreeOfParallelism = 5 }
Task.Run(() =>
{
    Parallel.ForEach(myCollection, options, item =>
    {
        DoWork(item);
    }
}


ParallelOptions
将为您开箱即用进行节流。

我在现实场景中使用它在后台运行很长时间的操作。这些操作是通过 HTTP 调用的，其设计目的是在长时间操作运行时不会阻止 HTTP 调用。

  1. 调用 HTTP 进行长后台操作。
  2. 操作在后台开始。
  3. 用户获取状态 ID，可用于使用另一个 HTTP 调用检查状态。
  4. 后台操作更新其状态。

这样，CI/CD 调用不会因为长时间的 HTTP 操作而超时，而是每 x 秒循环一次状态，而不会阻塞进程

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.