Книга «Конкурентность в C#. Асинхронное, параллельное и многопоточное программирование. 2-е межд. изд.»14.04.2020 18:33

Привет, Хаброжители! Если вы побаиваетесь конкурентного и многопоточного программирования, эта книга написана для вас. Стивен Клири предоставляет в ваше распоряжение 85 рецептов работы с .NET и C# 8.0, необходимых для параллельной обработки и асинхронного программирования. Конкурентность уже стала общепринятым методом разработки хорошо масштабируемых приложений, но параллельное программирование остается непростой задачей. Подробные примеры и комментарии к коду позволят разобраться в том, как современные инструменты повышают уровень абстракции и упрощают конкурентное программирование. Вы научитесь использовать async и await для асинхронных операций, расширять возможности кода за счет использования асинхронных потоков, исследовать потенциал параллельного программирования с библиотекой TPL Dataflow, создавать конвейеры потоков данных с библиотекой TPL Dataflow, задействовать функциональность System.Reactive на базе LINQ, использовать потоково-безопасные и неизменяемые коллекции, проводить модульное тестирование конкурентного кода, брать под контроль пул потоков, реализовывать корректную кооперативную отмену, анализировать сценарии на предмет объединения конкурентных методов, пользоваться всеми возможностями асинхронно-совместимого объектно-ориентированного программирования, распознавать и создавать адаптеры для кода, в котором используются старые стили асинхронного программирования.

Основы параллельного программирования

4.1. Параллельная обработка данных

Задача

Имеется коллекция данных. Требуется выполнить одну и ту же операцию с каждым элементом данных. Эта операция является ограниченной по вычислениям и может занять некоторое время.

Решение

Тип Parallel содержит метод ForEach, разработанный специально для этой задачи. Следующий пример получает коллекцию матриц и поворачивает эти матрицы:

void RotateMatrices(IEnumerable matrices, float degrees)
{
   Parallel.ForEach(matrices, matrix => matrix.Rotate(degrees));
}

Возможны ситуации, в которых преждевременно требуется прервать цикл (например, при обнаружении недействительного значения). Следующий пример обращает каждую матрицу, но при обнаружении недействительной матрицы цикл будет прерван:

void InvertMatrices(IEnumerable matrices)
{
   Parallel.ForEach(matrices, (matrix, state) =>
   {
      if (!matrix.IsInvertible)
        state.Stop();
      else
        matrix.Invert();
   });
}

Этот код использует ParallelLoopState.Stop для остановки цикла и предотвращения любых дальнейших вызовов тела цикла. Учтите, что цикл является параллельным, поэтому могут уже выполняться другие вызовы тела цикла, включая вызовы для элементов, следующих после текущего. В приведенном примере кода если третья матрица не является обратимой, то цикл прерывается и новые матрицы обрабатываться не будут, но может оказаться, что уже обрабатываются другие матрицы (например, четвертая и пятая).

Более распространенная ситуация встречается тогда, когда требуется отменить параллельный цикл. Это не то же, что остановка цикла; цикл останавливается изнутри и отменяется за своими пределами. Например, кнопка отмены может отменить CancellationTokenSource, отменяя параллельный цикл, как в следующем примере:

void RotateMatrices(IEnumerable matrices, float degrees,
      CancellationToken token)
{
   Parallel.ForEach(matrices,
         new ParallelOptions { CancellationToken = token },
         matrix => matrix.Rotate(degrees));
}

Следует иметь в виду, что каждая параллельная задача может выполняться в другом потоке, поэтому любое совместное состояние должно быть защищено. Следующий пример обращает каждую матрицу и подсчитывает количество матриц, которые обратить не удалось:

// Примечание: это не самая эффективная реализация.
// Это всего лишь пример использования блокировки
// для защиты совместного состояния.
int InvertMatrices(IEnumerable matrices)
{
  object mutex = new object();
  int nonInvertibleCount = 0;
  Parallel.ForEach(matrices, matrix =>
  {
     if (matrix.IsInvertible)
    {
       matrix.Invert();
    }
    else
    {
       lock (mutex)
      {
         ++nonInvertibleCount;
      }
    }
  });
  return nonInvertibleCount;
}

Пояснение

Метод Parallel.ForEach предоставляет возможность параллельной обработки для последовательности значений. Аналогичное решение Parallel LINQ (PLINQ) предоставляет практически те же возможности в LINQ-подобном синтаксисе. Одно из различий между Parallel и PLINQ заключается в том, что PLINQ предполагает, что может использовать все ядра на компьютере, тогда как Parallel может динамически реагировать на изменения условий процессора.

Parallel.ForEach реализует параллельный цикл foreach. Если вам потребуется выполнить параллельный цикл for, то класс Parallel также поддерживает метод Parallel.For. Метод Parallel.For особенно полезен при работе с несколькими массивами данных, которые получают один индекс.

Дополнительная информация

В рецепте 4.2 рассматривается параллельное агрегирование серий значений, включая суммирование и вычисление средних значений.

В рецепте 4.5 рассматриваются основы PLINQ.

В главе 10 рассматривается отмена.

4.2. Параллельное агрегирование

Задача

Требуется агрегировать результаты при завершении параллельной операции (примеры агрегирования — суммирование значений или вычисление среднего).

Решение

Для поддержки агрегирования класс Parallel использует концепцию локальных значений — переменных, существующих локально внутри параллельного цикла. Это означает, что тело цикла может просто обратиться к значению напрямую, без необходимости синхронизации. Когда цикл готов к агрегированию всех своих локальных результатов, он делает это с помощью делегата localFinally. Следует отметить, что делегату localFinally не нужно синхронизировать доступ к переменной для хранения результата. Пример параллельного суммирования:

// Примечание: это не самая эффективная реализация.
// Это всего лишь пример использования блокировки
// для защиты совместного состояния.
int ParallelSum(IEnumerable values)
{
  object mutex = new object();
  int result = 0;
  Parallel.ForEach(source: values,
        localInit: () => 0,
        body: (item, state, localValue) => localValue + item,
        localFinally: localValue =>
       {
          lock (mutex)
             result += localValue;
       });
  return result;
}

В Parallel LINQ реализована более понятная поддержка агрегирования, чем в классе Parallel:

int ParallelSum(IEnumerable values)
{
   return values.AsParallel().Sum();
}

О'кей, это был дешевый трюк, потому что в PLINQ реализована встроенная поддержка многих распространенных операторов (например, Sum). В PLINQ также предусмотрена обобщенная поддержка агрегирования с оператором Aggregate:

int ParallelSum(IEnumerable values)
{
  return values.AsParallel().Aggregate(
        seed: 0,
        func: (sum, item) => sum + item
  );
}

Пояснение

Если вы уже используете класс Parallel, следует использовать его поддержку агрегирования. В остальных случаях поддержка PLINQ, как правило, более выразительна, а код получается короче.

Дополнительная информация

В рецепте 4.5 изложены основы PLINQ.

4.3. Параллельный вызов

Задача

Имеется набор методов, которые должны вызываться параллельно. Эти методы (в основном) независимы друг от друга.

Решение

Класс Parallel содержит простой метод Invoke, спроектированный для таких сценариев. В следующем примере массив разбивается надвое, и две половины обрабатываются независимо:

void ProcessArray(double[] array)
{
   Parallel.Invoke(
         () => ProcessPartialArray(array, 0, array.Length / 2),
         () => ProcessPartialArray(array, array.Length / 2, array.Length)
   );
}

void ProcessPartialArray(double[] array, int begin, int end)
{
   // Обработка, интенсивно использующая процессор...
}

Методу Parallel.Invoke также можно передать массив делегатов, если количество вызовов неизвестно до момента выполнения:

void DoAction20Times(Action action)
{
   Action[] actions = Enumerable.Repeat(action, 20).ToArray();
   Parallel.Invoke(actions);
}

Parallel.Invoke поддерживает отмену, как и другие методы класса Parallel:

void DoAction20Times(Action action, CancellationToken token)
{
   Action[] actions = Enumerable.Repeat(action, 20).ToArray();
   Parallel.Invoke(new ParallelOptions { CancellationToken = token },
        actions);
}

Пояснение

Метод Parallel.Invoke — отличное решение для простого параллельного вызова. Отмечу, что он уже не так хорошо подходит для ситуаций, в которых требуется активизировать действие для каждого элемента входных данных (для этого лучше использовать Parallel.ForEach), или если каждое действие производит некоторый вывод (вместо этого следует использовать Parallel LINQ).

Дополнительная информация

В рецепте 4.1 рассматривается метод Parallel.ForEach, который выполняет действие для каждого элемента данных.

В рецепте 4.5 рассматривается Parallel LINQ.

Об авторе

Стивен Клири — опытный разработчик, прошел путь от ARM до Azure. Участвовал в написании открытого исходного кода библиотеки Boost C ++ и выпустил несколько собственных библиотек и утилит.

» Более подробно с книгой можно ознакомиться на сайте издательства
» Оглавление
» Отрывок

Для Хаброжителей скидка 25% по купону — Клири

По факту оплаты бумажной версии книги на e-mail высылается электронная книга.

© Habrahabr.ru