并行foreach和异步lambda

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

如果我想在并行循环的lambda中调用c#中标记为async的方法，就会出现问题。例如:

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时就会出现问题，因为创建的所有线程实际上都只是后台线程和并行线程。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;

它的工作，但它完全禁用等待聪明，我必须做一些手动异常处理..(为简洁起见，删除)。

我如何实现一个并行。ForEach循环，它在lambda?这可能吗?

平行飞船的原型。ForEach方法以Action<T>作为参数，但我希望它等待我的异步lambda。

当前回答

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

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帖子。

2013-02-28 13:30:44

其他回答

使用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;

2019-07-02 21:44:28

在接受的应答中，ConcurrentBag不是必需的。下面是一个没有它的实现:

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

任何"// some pre stuff"和"// some post stuff"都可以进入GetData实现(或另一个调用GetData的方法)

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

2021-01-18 10:30:23

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

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帖子。

2013-02-28 13:30:44

从其他答案和接受的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);
}

2021-03-15 09:56:53

下面的设置是使用IAsyncEnumerable，但可以通过更改类型和删除foreach上的“await”来修改为使用IEnumerable。它更适合于大型数据集，而不是创建无数并行任务，然后等待它们全部完成。

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

2021-02-09 03:16:26

并行foreach和异步lambda

推荐文章

最新文章

标签