Должны ли мы ждать всех асинхронных вызовов при использовании Parallel.Foreach?

Question

Если у нас есть список фильтров, можно ли использовать Parallel.ForEach для выполнения асинхронных вызовов, а затем использовать их?Должны ли мы ждать окончания всех звонков?Что делать, если у нас есть 3 асинхронных вызова, и один из них завершается с ошибкой / время ожидания?

var results = new ConcurrentBag<Service>();

Parallel.ForEach(filters, async filter =>
{
   var result = await serviceClient.GetAllAsync(filter).ConfigureAwait(false);

   results.AddRange(result);
});

MyMapper.Map(results);

, где MyMapper имеет метод:

Map(IEnumerable<Service> services) {
   foreach(Service service in services) {
      //do stuff
   }
}

Answer 1

Предполагается, что библиотека Parallel обрабатывает миллионы элементов параллельно.Вы не получите никакого улучшения, используя Parallel.ForEach там.Вместо этого вы можете использовать Task.WhenAll .

var tasks = new List<Task>();

foreach(var filter in filters)
{
   tasks.Add(serviceClient.GetAllAsync(filter));
}

results.AddRange(await Task.WhenAll(tasks).ConfigureAwait(false));

Вы можете использовать LinQ.Я не так, идея более понятна.С этим кодом вы можете иметь некоторую степень параллелизма, а await поможет вам с исключениями.

Answer 2

Я бы сделал это немного по-другому.Предполагая, что

serviceClient.GetAllAsync(filter)

делает что-то в другом потоке / задаче или выполняет запрос по некоторому URL, я не вижу необходимости использования Paralle.ForEach.

Что я хотел бы сделать, это создать список задач и вызывать их с помощью метода Task.WaitAll.Таким образом, вы можете обрабатывать любые исключения для ошибочных задач.

Фрагмент кода для этого будет выглядеть примерно так:

  var queryService = filters.Select(x => sClient.GetAllAsync()).ToArray();

        try
        {
            // Wait for all the tasks to finish.
            Task.WaitAll(queryService);
        }
        catch (AggregateException e)
        {
            //At least one of the Task instances was canceled.
            //If a task was canceled, the AggregateException contains an OperationCanceledException in its InnerExceptions collection.
        }