Многопоточное программирование в Java

Как только последний поток вызывает метод await, он прибывает к барьеру, выполняется действие барьера и потоки вызывают метод await второго барьера.

Как только последний поток вызывает метод await, выполняется действие второго барьера и потоки освобождаются.

Синхронизатор Phaser аналогичен синхронизатору CyclicBarrier, за исключением того, что в CyclicBarrier количество участников-потоков фиксировано, а в Phaser новые участники-потоки могут регистрироваться и отменять регистрацию динамически.

В этом примере, мы создаем синхронизатор.

И при создании экземпляра Phaser из основного потока мы передаем 1 в качестве аргумента.

Это эквивалентно вызову метода register из текущего потока.

Мы делаем это, потому что, когда мы создаем три рабочих потока, основной поток является координатором, и поэтому Phaser должен иметь четыре зарегистрированных потока.

Далее создаются три потока, выполняющих задачу, в которую передается синхронизатор.

В этой задаче поток регистрируется для синхронизатора и вызывает метод arriveAndAwaitAdvance, который аналогичен методу await CyclicBarrier и собирает потоки у барьера, включая главный поток.

Как только последний поток вызывает метод arriveAndAwaitAdvance, включая главный поток, фаза завершается.

Отмена регистрации потока для синхронизатора производится методом arriveAndDeregister.

Здесь в первом вызове метод getPhase вернет 0, так как фаза после инициализации равна нулю.

А во втором вызове метод getPhase вернет 1.

Exchanger (обменник) позволяет обменяться данными между двумя потоками в определенной точке работы обоих потоков.

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

В этом примере создаются два потока, обрабатывающие задачу.

И в этой задаче вызывается метод exchange, который берет значение поля другого потока и присваивает его полю текущего потока.

Параллельные потоки Stream

Ранее мы уже познакомились с программным интерфейсом Stream API, который значительно упрощает работу с коллекциями данных, позволяя отфильтровать, преобразовать и объединить данные без их промежуточного сохранения.

Кроме того, Stream API не модифицирует обрабатываемую коллекцию данных, а позволяет создать на ее основе новую коллекцию данных после обработки.

Напомню, что операторы Stream API делятся на две группы:

Промежуточные операторы, которые обрабатывают поступающие элементы и возвращают стрим.

И терминальные операторы, которые обрабатывают элементы и завершают работу стрима, так что терминальный оператор в цепочке может быть только один.

Обработка коллекции в Stream API не начинается до тех пор, пока не будет вызван терминальный оператор, то есть обработка происходит от терминального оператора к источнику.

И стрим нельзя использовать повторно. Как только вы вызываете любую терминальную операцию, стрим закрывается.

Стрим потоки могут выполняться параллельно, чтобы увеличить производительность обработки большого количества элементов.

Параллельные потоки стрим созданы на основе фреймворка Fork/Join и используют общий пул потоков ForkJoinPool, создаваемый с помощью статического метода ForkJoinPool.commonPool.

Размер общего пула потоков зависит от количества доступных физических ядер процессора.

Под капотом параллельный Stream API обеспечивает работу с потоконебезопасными коллекциями, разбиение коллекции элементов на части, создание потоков и объединение частей вместе.

Потокобезопасность обработки коллекций здесь достигается за счет отсутствия модификации коллекции во время ее обработки.

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

Затем на нужном этапе группы объединяются в одну для предоставления конечного результата.

Чтобы получить параллельный стрим, нужно либо вызвать метод parallelStream вместо метода stream, либо превратить обычный стрим в параллельный, вызвав промежуточный оператор parallel.

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

Если количество элементов небольшое, нужно учесть, что некоторые параллельные стрим операции, такие как reduce и collect, требуют дополнительных вычислений (операций объединения), которые не нужны при последовательном выполнении. Поэтому в случае небольшого количества элементов, последовательный стрим может выполнится быстрее, чем параллельный стрим.

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

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