Как использовать ожидание в параллельном foreach?

Так что я провел большую часть ночи, пытаясь понять это.

Мне посчастливилось вчера познакомиться с parallel.foreach, и он работает так, как я хочу, за исключением одной детали.

У меня есть следующее:

        Parallel.ForEach(data, (d) =>
        {
            try
            {
                MyMethod(d, measurements);
            }
            catch (Exception e)
            {
              // logg
            }

        });

В методе «MyMethod» у меня есть много логики, которая выполняется, и большая часть ее в порядке, но я делаю вызовы API, где я извлекаю данные, и я использую асинхронную задачу для этого, чтобы иметь возможность использовать «ожидание», чтобы код подождите, пока эта конкретная часть не будет выполнена, а затем двигайтесь дальше:

    private async void MyMethod(PimData pimData, IEnumerable<ProductMeasurements> measurements)
    {
        try
        {
           // alot of logic but most relevant part 

            await Task.WhenAll(ExecuteMeasurmentAndChartLogic(pimData.ProductNumber, entity));
            await Task.WhenAll(resourceImportManager.HandleEntityImageFiles(pimData.ProductType + pimData.ProductSize,SwepImageType.Png, ResourceFileTypes.ThreeD, entity, LinkTypeId.ProductResource));

            await Task.WhenAll(resourceImportManager.HandleEntityImageFiles(pimData.ProductSketch, SwepImageType.Png, ResourceFileTypes.Sketch, entity, LinkTypeId.ProductResource));

        }
        catch (Exception e)
        {
            // logg
        }
    }

Проблемы:

1 Для начала цикл завершается до завершения всего кода

2 Вторая проблема заключается в том, что я получаю сообщение «Задача была отменена» во многих вызовах API.

3 И в-третьих, как упоминалось выше, код не ждет полного выполнения каждого метода.

Я не могу заставить его выполнить все в методе ExecuteMeasurmentAndChartLogic(), прежде чем перейти к следующему шагу.

Это дает мне следующие проблемы (больше проблем):

В этом методе я создаю элемент и добавляю его в БД, и этот элемент нуждается в дополнительной информации, которую я получаю из вызова API, который выполняется внутри ExecuteMeasurmentAndChartLogic(), но проблема в том, что несколько элементов создаются и должны ждать остальные данные, которые мне не нужны.

ПРИМЕЧАНИЕ. Я знаю, что создание элемента и добавление в базу данных до того, как будут получены все данные, не является лучшей практикой, но я интегрируюсь с PIM, и этот процесс является деликатным

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

Уточнить:

Работает несколько предметов

Каждый элемент обрабатывает ВСЮ логику, которую ему нужно обработать, прежде чем перейти к следующей части кода. Обычно это делается с помощью await.

В приведенном выше коде метод resourceImportManager() выполняется до завершения ExecuteMeasurmentAndChartLogic(). чего я не хочу.

Вместо Parallel.ForEach я использовал:

    Task task1 = Task.Factory.StartNew(() => MyMethod(data, measurements));
    Task.WaitAll(task1);

но это не сильно помогло

Довольно новичок в этом и не смог понять, где я делаю это неправильно.

РЕДАКТИРОВАТЬ: Обновлены проблемы с этим

РЕДАКТИРОВАТЬ: так выглядит ExecuteMeasurmentAndChartLogic():

    public async Task ExecuteMeasurmentAndChartLogic(string productNumber, Entity entity)
    {
        try
        {
            GrafGeneratorManager grafManager = new GrafGeneratorManager();
            var graphMeasurmentList = await MeasurmentHandler.GetMeasurments(productNumber);

            if (graphMeasurmentList.Count == 0) return;

            var chart = await grafManager.GenerateChart(500, 950, SystemColors.Window, ChartColorPalette.EarthTones,
                "legend", graphMeasurmentList);

            await AddChartsAndAddToXpc(chart, entity, productNumber);
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
        }

    }

РЕДАКТИРОВАТЬ: Предыстория этого: я звоню в API, чтобы получить много данных. Для каждого элемента в этих данных мне нужно сделать вызов API и получить данные, которые я применяю к элементу.

После прочтения комментариев, которые также заставили меня задуматься по-другому. Возможно, я могу перебрать все свои элементы и выполнить для них небольшую логику, добавить URL-адрес в список задач и создать отдельную задачу, которая выполняет это один за другим.

буду держать это в курсе


person ThunD3eR    schedule 28.07.2017    source источник
comment
Используйте делегаты обратного вызова после завершения обработки.   -  person Ramankingdom    schedule 28.07.2017
comment
Есть ли конкретная причина, по которой MyMethod является async void, а не async Task?   -  person GWigWam    schedule 28.07.2017
comment
@GWigWam нет особой причины, хотел изменить, спасибо, что напомнили   -  person ThunD3eR    schedule 28.07.2017
comment
Что возвращает resourceImportManager.HandleEntityImageFiles. Для начала похоже, что вы используете WaitAll и WhenAll для отдельных задач. Кроме того, асинхронная пустота — это большое нет-нет, вы, вероятно, могли бы изменить Parallel.ForEach на Task.WhenAll с помощью некоторого рефакторинга.   -  person Peter Bons    schedule 28.07.2017
comment
Кроме того, parallel.for лучше подходит для работы, связанной с ЦП, а Task.WhenAll — для работы, связанной с вводом-выводом. использование async/await в параллели для является небольшим недостатком дизайна imho   -  person Peter Bons    schedule 28.07.2017
comment
@Peter Bons resourceImportManager.HandleEntityImageFiles ничего не возвращает, я просто хочу, чтобы он выполнил все там, прежде чем двигаться дальше.   -  person ThunD3eR    schedule 28.07.2017
comment
если он ничего не возвращает, как он может быть параметром Task.WhenAll?   -  person Peter Bons    schedule 28.07.2017
comment
@Peter Bons: общедоступный асинхронный Task HandleEntityImageFiles был пустым, прежде чем я попробовал все возможные вещи, которые я мог придумать, чтобы заставить его выполнять все   -  person ThunD3eR    schedule 28.07.2017
comment
Чтобы повторить комментарии выше, либо он привязан к ЦП, и в этом случае вы должны использовать Parallel.ForEach, либо он привязан к вводу-выводу, и в этом случае вам следует использовать асинхронный режим.   -  person Tim Rogers    schedule 28.07.2017
comment
тогда вы должны использовать await HandleEntityImageFiles вместо await await Task.WhenAll(HandleEntityImageFiles), так как он возвращает одну задачу. Теперь вы можете запускать несколько задач одновременно, но тогда я ожидаю что-то вроде await Task.WhenAll(ExecuteMeasurmentAndChartLogic(pimData.ProductNumber, entity), resourceImportManager.HandleEntityImageFiles(xxx), resourceImportManager.HandleEntityImageFiles(yyy);   -  person Peter Bons    schedule 28.07.2017
comment
Но я предлагаю вам прочитать еще несколько руководств, прежде чем создавать такую ​​логику, поскольку вы смешиваете слишком много вещей, и это не лучшее место для объяснения всех этих вещей.   -  person Peter Bons    schedule 28.07.2017
comment
@Peter Bons, который я использовал await перед использованием Task.WhenAll, дал мне тот же результат. Что касается дополнительного чтения, я согласен, и у меня есть, но мне нужно сделать оптимизацию, и я пытаюсь учиться по ходу дела.   -  person ThunD3eR    schedule 28.07.2017
comment
Рассмотрим поток данных TPL.   -  person Stephen Cleary    schedule 28.07.2017
comment
@Peter Bons, не могли бы вы поделиться мыслями о моем решении ниже. Спасибо!   -  person ThunD3eR    schedule 28.07.2017


Ответы (1)


Не используйте Parralel.ForEach вообще. Сделайте так, чтобы ваш метод возвращал Task вместо void, собирайте все задачи и ждите их, например:

Task.WaitAll(data.Select(d => MyMethod(d, someParam)).ToArray());
person Eduard Lepner    schedule 28.07.2017
comment
Пока это дает мне лучший результат, но меня беспокоит, действительно ли он выполняет логику для каждого элемента одновременно? Кажется, он идет намного медленнее, чем цикл, с которого я начал - person ThunD3eR; 28.07.2017
comment
Я вижу это так: каждый элемент получает задачу, но второй элемент ждет выполнения первого элемента, а это не то, что я хочу, возвращает меня к обычному циклу foreach. - person ThunD3eR; 28.07.2017
comment
Код, который я предоставил, является асинхронным и параллельным. Он может выполняться в одном потоке, если конфигурация пула потоков нарушена или недостаточно потоков для захвата. Или, может быть, MyMethod имеет узкое место и фактически обращается к одному и тому же блокирующему ресурсу. Однако, если этот метод действителен, мой ответ также должен быть верным. - person Eduard Lepner; 28.07.2017
comment
Нет блокирующего ресурса. Я отлаживаю это, и после нескольких элементов он останавливается и ждет выполнения элементов. это для замедления - person ThunD3eR; 28.07.2017
comment
@Ra3IDeN Тогда ваш фактический метод не предназначен для параллельной работы, и вам нужно исправить это. - person Servy; 28.07.2017
comment
@Servy Я думаю, что да, проверьте мой ответ ниже. - person ThunD3eR; 28.07.2017
comment
@Ra3IDeN Этот ответ уменьшает объем работы, которую можно выполнять параллельно, для основной выполняемой работы, и неправильно распараллеливает работу, которую нельзя выполнять параллельно, в результате небезопасный код. - person Servy; 28.07.2017
comment
Наконец решил принять этот ответ, поскольку он отвечает на заданный вопрос. Однако я не буду использовать это решение, так как мне нужна многопоточность, а это блокирует каждый поток. - person ThunD3eR; 08.08.2017
comment
Как насчет Task.WaitAll(data.Select(d => Task.Run(() => MyMethod(d, someParam)).ToArray())); для сохранения параллелизма? Сам не проверял, просто идея. - person epicTurk; 17.05.2019