Параллельный foreach с асинхронной лямбдой

139

Я хотел бы обрабатывать коллекцию параллельно, но у меня проблемы с ее реализацией, поэтому я надеюсь на некоторую помощь.

Проблема возникает, если я хочу вызвать метод, помеченный как async в 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;

Он работает, но полностью отключает умение ожидания, и мне приходится вручную обрабатывать исключения. (Удалено для краткости).

Как я могу реализовать Parallel.ForEachцикл, который использует ключевое слово await в лямбде? Является ли это возможным?

Прототип метода Parallel.ForEach принимает Action<T>параметр as, но я хочу, чтобы он дождался моей асинхронной лямбды.

clausndk
источник
1
Я предполагаю, что вы хотели удалить awaitиз await GetData(item)во втором блоке кода, так как это приведет к ошибке компиляции как есть.
Джош М.
2
Возможный дубликат Nesting ждут в Parallel.ForEach
Виталий Улантиков 02

Ответы:

189

Если вам просто нужен простой параллелизм, вы можете сделать это:

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;

Если вам нужно что-то более сложное, посмотрите пост Стивена ТубаForEachAsync .

Стивен Клири
источник
46
Наверное, нужен дроссельный механизм. Это немедленно создаст столько задач, сколько есть элементов, которые могут оказаться в 10k сетевых запросах и тому подобное.
usr
10
@usr Последний пример в статье Стивена Туба обращается к этому.
svick
@svick Я ломал голову над тем последним образцом. Мне кажется, что он просто группирует множество задач, чтобы создать для меня больше задач, но все они начинают работать в массовом порядке.
Люк Пуплетт
2
@LukePuplett Он создает dopзадачи, и каждая из них затем последовательно обрабатывает некоторое подмножество входной коллекции.
svick
4
@Afshin_Zavvar: Если вы вызываете Task.Runбез awaitрезультата, то это просто бросает работу «выстрелил и забыл» в пул потоков. Это почти всегда ошибка.
Стивен Клири
75

Вы можете использовать ParallelForEachAsyncметод расширения из пакета NuGet AsyncEnumerator :

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;
Серж Семенов
источник
1
Это твоя посылка? Я видел, как вы сейчас разместили это в нескольких местах? : D Ой, подождите .. ваше имя на упаковке: D +1
Петр Кула
18
@ppumkin, да, это мое. Я видел эту проблему снова и снова, поэтому решил решить ее самым простым способом и освободить других от борьбы :)
Серж Семенов
Спасибо .. это определенно имеет смысл и очень помогло мне!
Петр Кула
2
у вас опечатка: maxDegreeOfParallelism>maxDegreeOfParalellism
Ширан Дрор
3
Правильное написание действительно maxDegreeOfParallelism, однако есть что-то в комментарии @ ShiranDror - в вашем пакете вы по ошибке назвали переменную maxDegreeOfParalellism (и поэтому ваш цитируемый код не будет компилироваться, пока вы его не измените ..)
BornToCode
17

С SemaphoreSlimего помощью можно добиться контроля параллелизма.

var bag = new ConcurrentBag<object>();
var maxParallel = 20;
var throttler = new SemaphoreSlim(initialCount: maxParallel);
var tasks = myCollection.Select(async item =>
{
  try
  {
     await throttler.WaitAsync();
     var response = await GetData(item);
     bag.Add(response);
  }
  finally
  {
     throttler.Release();
  }
});
await Task.WhenAll(tasks);
var count = bag.Count;
Фелипе л
источник
3

Моя легкая реализация ParallelForEach async.

Особенности:

  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.SetResult(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);
nicolay.anykienko
источник
2

Я создал для этого метод расширения, который использует 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);
Джей шах
источник
«использование» не поможет. Цикл foreach будет ждать семафона бесконечно. Просто попробуйте этот простой код, который воспроизводит проблему: await Enumerable.Range (1, 4) .ForEachAsyncConcurrent (async (i) => {Console.WriteLine (i); throw new Exception ("тестовое исключение");}, maxDegreeOfParallelism: 2);
nicolay.anykienko
@ nicolay.anykienko вы правы насчет №2. Эту проблему с памятью можно решить, добавив tasksWithThrottler.RemoveAll (x => x.IsCompleted);
askids
1
Я пробовал это в своем коде, и если я maxDegreeOfParallelism не равен нулю, тупики кода. Здесь вы можете увидеть весь код для воспроизведения: stackoverflow.com/questions/58793118/…
Массимо Савацци