首页 文章

与异步lambda并行的foreach

提问于
浏览 次
63

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

如果我想在并行循环的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时会出现问题，因为创建的所有线程实际上只是后台线程， 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中使用await关键字实现 Parallel.ForEach 循环？可能吗？

Parallel.ForEach方法的原型采用 Action<T> 作为参数，但我希望它等待我的异步lambda .

c# async-await task-parallel-library parallel.foreach

4 回答

  • 1

    您可以使用AsyncEnumerator NuGet Package中的 ParallelForEachAsync 扩展方法：

    using System.Collections.Async;

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;
    回复于
  • 105

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

    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's ForEachAsync post .

    回复于
  • 23

    我已经为此创建了一个扩展方法，它使用了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

    我的ParallelForEach异步的轻量级实现 .

    特征：

    • 限制（最大并行度） .

    • 异常处理（完成时将抛出聚合异常） .

    • 内存高效（无需存储任务列表） .

    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)
                {
                    if (exceptions.Count > 0)
                    {
                        tcs.SetException(new AggregateException(exceptions));
                    }
                    else
                    {
                        tcs.SetResult(null);
                    }
                }
            });
        }

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

    用法示例：

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

相关问题