java.util.concurrent. Часть первая: Зачем и почему?
Часть первая, в которой множеством слов раскрывается смысл существования этого API
Эта статья, хоть и не является прямым переводом, основана на статье Брайана Гетца Concurrency in JDK 5.0
Потоки позволяют задействовать все имеющиеся логические процессоры современного компьютера, будь это десктоп или сервер, и таким образом потенциально выполнять больше операций за то же время.
Однако потоки были реализованы современными операционными системами задолго до массового появления многоядерных процессоров. Зачем же нужны потоки на однопроцессорных системах?
1. Отзывчивый графический интерфейс. Длительные операции могут быть выполнены в отдельном потоке, в то время как приложение способно обслуживать события от клавиатуры и мышки.
2. Возможность более эффективного использования ресурсов: процессора, памяти, жесткого диска и сети. В то время как один из потоков простаивает, ожидая завершения операции чтения файла, второй поток может в это время устанавливать сетевое соединение с клиентом, а третий — обрабатывать какой нибудь запрос.
3. Простота модели “поток-на-запрос” и идея фоновых системных процессов. Оставляя детали распределения ресурса процессора на совесть операционной системы, мы получаем возможность сконцентрировать программу на обработке самого запроса. Мы также можем выделить определённые задачи (например, сборку мусора или системный отладчик) в отдельные потоки, и таким образом “незаметно” добавить новые свойства программе.
Первые два пункта в основном понятны, третий распишу более подробно.
В императивном программировании алгоритмы, придуманные и придумываемые человеком — это последовательные инструкции, преобразовывающие одни данные (входные) в другие (выходные). Обычно это как раз однопоточные алгоритмы, когда задача упрощена до одного набора входных значений за раз и алгоритм выполняется до тех пор, пока не будет получен результат его работы.
Однако алгоритм не создается для одного запроса. Если требуется обрабатывать большое количество входных данных, то возникает желание либо уменьшить общее время их обработки, либо увеличить скорость их обработки (и таким образом увеличить пропускную способность обработчика). Самый простой способ в данном случае — поставить больше однопоточных алгоритмов и заставить их работать параллельно. Возможно, придется как-то согласовывать доступ к общим ресурсам и координировать их выполнение, но это обычно значительно проще чем разработка нового алгоритма, который за раз может обработать больше чем один набор данных (такой алгоритм нужно не только разработать, но еще и доказать). Это и понимается под простотой модели “поток-на-запрос”.
Не знаю, что родилось первым, — идея поставить несколько алгоритмов на поток, или идея использовать разные алгоритмы в потоках и заставить их взаимодействовать, с целью решить более общую задачу. Скорее всего первая, но статья не об этом. Реальность такова, что обычное современное приложение проектируется как набор объектов, которые взаимодействуют между собой. Плюс, по вышеизложенным четырем причинам, — это еще и несколько потоков, которые взаимодействуют между собой.
Каждый убежденный программист при программировании многопоточного приложения использует общие для нескольких (возможно всех) потоков данные, размещенные в памяти. Одни данные играют ключевую роль в координации потоков, например условия завершения операции копирования большого файла, другие хранят общие для потоков сведения о состоянии объектов, обеспечивая разделяемый доступ к ним, например, значения электронной таблицы или список сотрудников компании, которые отображены на экране; вычисление третьих данных может быть целью выполнения множества потоков, например лента комментариев этого блога или счетчик посетителей сайта.
Корректность операций над такими данными в многопоточном окружении — жизненно-важное свойство любого приложения. В основном это требование корректности сводится к тому, что ни один поток не должен наблюдать внутренне несогласованное состояние разделяемого ресурса. Обычно это достигается требованием атомарности всех операций над общими данными — как доступа, так и изменения.
Поскольку в Java данные — это объекты, то к классам этих объектов, которые будут обрабатывать вызовы из нескольких потоков, предъявляется требование многопоточной безопасности (thread safety). Что это значит? Это. конечно же значит, что вызовы методов обьекта будут безопасными, что, конечно, правильно, но нисколько не помогает нам понять, что же такое потокобезопасность класса.
Во-первых потокобезопасный класс должен корректно себя вести в однопоточном окружении. Если класс верно запрограммирован, то есть соответствует своей спецификации, это значит, что любые вызовы публичных методов в любом порядке и любые операции над публичными полями не приведут к неверному внутреннему состоянию, не позволят наблюдать обьект в неверном состоянии, не нарушат любой из инвариантов класса, ни любое из его пред- или постусловий.
Во-вторых потокобезопасный класс должен также продолжать корректно себя вести в многопоточном окружении без дополнительной синхронизации со стороны вызывающего кода, то есть вышесказанное должно соблюдаться также независимо от того, в какой последовательности операционная система решит чередовать инструкции разных методов класса.
В результате все изменения состояния объекта разными потоками должны внешне выглядеть согласованными и иметь правильный порядок без дополнительных усилий в вызывающем коде, а операционная система получает возможность чередовать эти изменения в рамках, разрешенных разработчиком класса.
Многопоточное программирование в Java 8. Часть первая. Параллельное выполнение кода с помощью потоков
Авторизуйтесь
Многопоточное программирование в Java 8. Часть первая. Параллельное выполнение кода с помощью потоков
Добро пожаловать в первую часть руководства по параллельному программированию в Java 8. В этой части мы на простых примерах рассмотрим, как выполнять код параллельно с помощью потоков, задач и сервисов исполнителей.
Впервые Concurrency API был представлен вместе с выходом Java 5 и с тех пор постоянно развивался с каждой новой версией Java. Большую часть примеров можно реализовать на более старых версиях, однако в этой статье я собираюсь использовать лямбда-выражения. Если вы все еще не знакомы с нововведениями Java 8, рекомендую посмотреть мое руководство.
Потоки и задачи
Все современные операционные системы поддерживают параллельное выполнение кода с помощью процессов и потоков. Процесс — это экземпляр программы, который запускается независимо от остальных. Например, когда вы запускаете программу на Java, ОС создает новый процесс, который работает параллельно другим. Внутри процессов мы можем использовать потоки, тем самым выжав из процессора максимум возможностей.
Поскольку интерфейс Runnable функциональный, мы можем использовать лямбда-выражения, которые появились в Java 8. В примере мы создаем задачу, которая выводит имя текущего потока на консоль, и запускаем ее сначала в главном потоке, а затем — в отдельном.
Результат выполнения этого кода может выглядеть так:
Из-за параллельного выполнения мы не можем сказать, будет наш поток запущен до или после вывода «Done!» на экран. Эта особенность делает параллельное программирование сложной задачей в больших приложениях.
4–5 декабря, Онлайн, Беcплатно
Потоки могут быть приостановлены на некоторое время. Это весьма полезно, если мы хотим сэмулировать долго выполняющуюся задачу. Например, так:
Работать с потоками напрямую неудобно и чревато ошибками. Поэтому в 2004 году в Java 5 добавили Concurrency API. Он находится в пакете java.util.concurrent и содержит большое количество полезных классов и методов для многопоточного программирования. С тех пор Concurrency API непрерывно развивался и развивается.
Давайте теперь подробнее рассмотрим одну из самых важных частей Concurrency API — сервис исполнителей (executor services).
Исполнители
Concurrency API вводит понятие сервиса-исполнителя (ExecutorService) — высокоуровневую замену работе с потоками напрямую. Исполнители выполняют задачи асинхронно и обычно используют пул потоков, так что нам не надо создавать их вручную. Все потоки из пула будут использованы повторно после выполнения задачи, а значит, мы можем создать в приложении столько задач, сколько хотим, используя один исполнитель.
Вот как будет выглядеть наш первый пример с использованием исполнителя:
Класс Executors предоставляет удобные методы-фабрики для создания различных сервисов исполнителей. В данном случае мы использовали исполнитель с одним потоком.
Вот как я предпочитаю останавливать исполнителей:
Исполнитель пытается завершить работу, ожидая завершения запущенных задач в течение определенного времени (5 секунд). По истечении этого времени он останавливается, прерывая все незавершенные задачи.
Callable и Future
Давайте напишем задачу, которая возвращает целое число после секундной паузы:
Callable-задачи также могут быть переданы исполнителям. Но как тогда получить результат, который они возвращают? Поскольку метод submit() не ждет завершения задачи, исполнитель не может вернуть результат задачи напрямую. Вместо этого исполнитель возвращает специальный объект Future, у которого мы сможем запросить результат задачи.
Задачи жестко связаны с сервисом исполнителей, и, если вы его остановите, попытка получить результат задачи выбросит исключение:
Таймауты
Любой вызов метода future.get() блокирует поток до тех пор, пока задача не будет завершена. В наихудшем случае выполнение задачи не завершится никогда, блокируя ваше приложение. Избежать этого можно, передав таймаут:
Выполнение этого кода вызовет TimeoutException :
Вы уже, возможно, догадались, почему было выброшено это исключение: мы указали максимальное время ожидания выполнения задачи в одну секунду, в то время как ее выполнение занимает две.
InvokeAll
InvokeAny
Используем этот метод, чтобы создать несколько задач с разными строками и задержками от одной до трех секунд. Отправка этих задач исполнителю через метод invokeAny() вернет результат задачи с наименьшей задержкой. В данном случае это «task2»:
ForkJoinPool впервые появился в Java 7, и мы рассмотрим его подробнее в следующих частях нашего руководства. А теперь давайте посмотрим на исполнители с планировщиком (scheduled executors).
Исполнители с планировщиком
Мы уже знаем, как отдать задачу исполнителю и получить ее результат. Для того, чтобы периодически запускать задачу, мы можем использовать пул потоков с планировщиком.
ScheduledExecutorService способен запускать задачи один или несколько раз с заданным интервалом.
Этот пример показывает, как заставить исполнитель выполнить задачу через три секунды:
Кроме того, он принимает начальную задержку, которая определяет время до первого запуска.
Обратите внимание, что метод scheduleAtFixedRate() не берет в расчет время выполнения задачи. Так, если вы поставите задачу, которая выполняется две секунды, с интервалом в одну, пул потоков рано или поздно переполнится.
В этом примере мы ставим задачу с задержкой в одну секунду между окончанием выполнения задачи и началом следующей. Начальной задержки нет, и каждая задача выполняется две секунды. Так, задачи будут запускаться на 0, 3, 6, 9 и т. д. секунде. Как видите, метод scheduleWithFixedDelay() весьма полезен, если мы не можем заранее сказать, сколько будет выполняться задача.
Это была первая часть серии статей про многопоточное программирование. Настоятельно рекомендую разобрать вышеприведенные примеры самостоятельно. Все они доступны на GitHub. Можете смело форкать репозиторий и добавлять его в избранное.
Надеюсь, вам понравилась статья. Если у вас возникли какие-либо вопросы, вы можете задать их в твиттере.
JDK concurrent package
Модель памяти, существующая на данный момент в Java, гарантирует ожидаемый порядок выполнения многопоточного кода, при отсутствии в этом коде гонок потоков. И для того, чтобы обезопасить ваш код от гонок, придуманы различные способы синхронизации и обмена данными между ними.
Atomic
Результат выполнения будет ожидаемым:
Рассмотрим использование AtomicLong для реализации оптимистичной блокировки при расчете этой же суммы:
Как видно из результатов «ошибочных» попыток было не так уж и много:
На основе compare-and-set может также реализовываться неблокирующая read блокировка. В данном случае в atomic переменной будет храниться версия обрабатываемого объекта. Получив значение версии до вычислений мы можем сверить ее после вычисления. Обычные read-write блокировки вступают в силу, только если проверка версии провалилась.
Locks
ReentrantLock
В отличие от syncronized блокировок, ReentrantLock позволяет более гибко выбирать моменты снятия и получения блокировки т.к. использует обычные Java вызовы. Также ReentrantLock позволяет получить информацию о текущем состоянии блокировки, разрешает «ожидать» блокировку в течение определенного времени. Поддерживает правильное рекурсивное получение и освобождение блокировки для одного потока. Если вам необходимы честные блокировки (соблюдающие очередность при захвате монитора) — ReentrantLock также снабжен этим механизмом.
Несмотря на то, что syncronized и ReentrantLock блокировки очень похожи — реализация на уровне JVM отличается довольно сильно.
Не вдаваясь в подробности JMM: использовать ReentrantLock вместо предоставляемой JVM syncronized блокировки стоит только в том случае, если у вас очень часто происходит битва потоков за монитор. В случае, когда в syncronized метод _обычно_ попадает лишь один поток — производительность ReentrantLock уступает механизму блокировок JVM.
ReentrantReadWriteLock
Дополняет свойства ReentrantLock возможностью захватывать множество блокировок на чтение и блокировку на запись. Блокировка на запись может быть «опущена» до блокировки на чтение, если это необходимо.
StampedLock _jdk 1.8_
Реализовывает оптимистичные и пессимистичные блокировки на чтение-запись с возможностью их дальнейшего увеличения или уменьшения. Оптимистичная блокировка реализуется через «штамп» лока (javadoc):
Collections
ArrayBlockingQueue
Честная очередь для передачи сообщения из одного потока в другой. Поддерживает блокирующие ( put() take() ) и неблокирующие ( offer() pool() ) методы. Запрещает null значения. Емкость очереди должна быть указанна при создании.
ConcurrentHashMap
Ключ-значение структура, основанная на hash функции. Отсутствуют блокировки на чтение. При записи блокируется только часть карты (сегмент). Кол-во сегментов ограничено ближайшей к concurrencyLevel степени 2.
ConcurrentSkipListMap
Сбалансированная многопоточная ключ-значение структура (O(log n)). Поиск основан на списке с пропусками. Карта должна иметь возможность сравнивать ключи.
ConcurrentSkipListSet
ConcurrentSkipListMap без значений.
CopyOnWriteArrayList
Блокирующий на запись, не блокирующий на чтение список. Любая модификация создает новый экземпляр массива в памяти.
CopyOnWriteArraySet
CopyOnWriteArrayList без значений.
DelayQueue
PriorityBlockingQueue разрешающая получить элемент только после определенной задержки (задержка объявляется через Delayed интерфейс объекта). DelayQueue может быть использована для реализации планировщика. Емкость очереди не фиксирована.
LinkedBlockingDeque
LinkedBlockingQueue
LinkedTransferQueue
Однонаправленная `BlockingQueue`, основанная на связанности (cache-miss & cache coherence overhead). Емкость очереди не фиксирована. Данная очередь позволяет ожидать когда элемент «заберет» обработчик.
PriorityBlockingQueue
Однонаправленная `BlockingQueue`, разрешающая приоритизировать сообщения (через сравнение элементов). Запрещает null значения.
SynchronousQueue
Однонаправленная `BlockingQueue`, реализующая transfer() логику для put() методов.
Synchronization points
CountDownLatch
CyclicBarrier
Барьер ( await() ), ожидающий конкретного кол-ва вызовов await() другими потоками. Когда кол-во потоков достигнет указанного будет вызван опциональный callback и блокировка снимется. Барьер сбрасывает свое состояние в начальное при освобождении ожидающих потоков и может быть использован повторно.
Exchanger
Барьер (`exchange()`) для синхронизации двух потоков. В момент синхронизации возможна volatile передача объектов между потоками.
Phaser
Расширение `CyclicBarrier`, позволяющая регистрировать и удалять участников на каждый цикл барьера.
Semaphore
Барьер, разрешающий только указанному кол-во потоков захватить монитор. По сути расширяет функционал `Lock` возможность находиться в блоке нескольким потокам.
Executors
ExecutorService пришел на замену new Thread(runnable) чтобы упростить работу с потоками. ExecutorService помогает повторно использовать освободившиеся потоки, организовывать очереди из задач для пула потоков, подписываться на результат выполнения задачи. Вместо интерфейса Runnable пул использует интерфейс Callable (умеет возвращать результат и кидать ошибки).
Метод invokeAll отдает управление вызвавшему потоку только по завершению всех задач. Метод invokeAny возвращает результат первой успешно выполненной задачи, отменяя все последующие.
ThreadPoolExecutor
Пул потоков с возможностью указывать рабочее и максимальное кол-во потоков в пуле, очередь для задач.
ScheduledThreadPoolExecutor
Расширяет функционал ThreadPoolExecutor возможностью выполнять задачи отложенно или регулярно.
ThreadPoolExecutor
Более легкий пул потоков для «самовоспроизводящих» задач. Пул ожидает вызовов `fork()` и `join()` методов у дочерних задач в родительской.
Уровень 26. Ответы на вопросы к собеседованию по теме уровня. Часть 2. Вопросы 6-9, 11-12
6. Что такое канкаренси?
7. Какие классы из «канкаренси» ты знаешь?
8. Как устроен класс ConcurrentHashMap?
Элементы карты
Хэш-функция
ConcurrentHashMap также используется улучшенная функция хэширования.
Напомню, какой она была в HashMap из JDK 1.2:
Версия из ConcurrentHashMap JDK 1.5:
В чём необходимость усложнения хэш-функции? Таблицы в хэш-карте имеют длину, определяемую степенью двойки. Для хэш-кодов, двоичные представления которых не различаются в младшей и старшей позиции, мы будем иметь коллизии. Усложнение хэш-функции как раз решает данную проблему, уменьшая вероятность коллизий в карте.
Сегменты
Карта делится на N различных сегментов (16 по умолчанию, максимальное значение может быть 16-битным и представлять собой степень двойки). Каждый сегмент представляет собой потокобезопасную таблицу элементов карты. Увеличение количества сегментов будет способствовать тому, что операции модификации будут затрагивать различные сегменты, что уменьшит вероятность блокировок во время выполнения.
ConcurrencyLevel
Данный параметр влияет на использование картой памяти и количество сегментов в карте.
Количество сегментов будет выбрано как ближайшая степень двойки, большая чем concurrencyLevel. Занижение concurrencyLevel ведёт к тому, что более вероятны блокировки потоками сегментов карты при записи. Завышение показателя ведёт к неэффективному использованию памяти. Если лишь один поток будет изменять карту, а остальные будут производить чтение — рекомендуется использовать значение 1.
Итого
Итак, основные преимущества и особенности реализации ConcurrentHashMap :
Многопоточный пакет util.concurrent
Разработка многопоточного кода, обеспечивающего не только хорошую производительность, но и защиту данных приложения, не является тривиальной задачей. Именно для таких целей предназначен пакет java.util.concurrent. Разработчики, как правило, ограничиваются обычной синхронизацией, и поэтому с пакетом многопоточности java.util.concurrent они сталкиваются не так уж часто.
В данной статье приводится обзор пакета java.util.concurrent, разбитого на несколько функциональных групп. В описании каждой группы приводится ссылка на более подробное описание объектов с примерами на страницах сайта.
Дополнительную информацию о многопоточности можно найти в документации Oracle Java Concurrency Tutorials и в книге Doug Lea «Concurrent Programming in Java by Doug Lea», являющегося автором пакета java.util.concurrent и профессором университета штата Нью Йорк. К наиболее известным java разработкам Doug Lea следует отнести collections и util.concurrent, которые в том или ином виде отразились в существующих JDK. На домашней страничке Doug Lea размещена второе издание книги по многопоточности Concurrent Programming in Java™: Design Principles and Pattern, 2nd Edition, с которой можно познакомиться здесь.
Структура пакета java.util.concurrent
Классы и интерфейсы пакета java.util.concurrent объедининены в несколько групп по функциональному признаку, как это представлено в следующей таблице :
| Наименование | Примечание |
|---|---|
| collections | Набор более эффективно работающих в многопоточной среде коллекций нежели стандартные универсальные коллекции из java.util пакета |
| synchronizers | Объекты синхронизации, позволяющие разработчику управлять и/или ограничивать работу нескольких потоков. |
| atomic | Набор атомарных классов, позволяющих использовать принцип действия механизма оптимистической блокировки для выполнения атомарных операций. |
| Queues | Объекты создания блокирующих и неблокирующих очередей с поддержкой многопоточности. |
| Locks | Механизмы синхронизации потоков, альтернативы базовым synchronized, wait, notify, notifyAll |
| Executors | Механизмы создания пулов потоков и планирования работы асинхронных задач |
Concurrent Collections
Обычные наборы данных, реализующих интерфейсы List, Set и Map, нельзя использовать в многопоточных приложениях, если требуется синхронизация, т.е. такие коллекции недопустимы для одновременного чтения и изменения данных разными потоками. Методы обрамления Collections framework (synchronizedList, synchronizedSet, synchronizedMap), появившиеся в JDK 1.2, имеют существенный недостаток, связанный с препятствованием масштабируемости, поскольку с коллекцией одновременно может работать только один поток.
Пакет java.util.concurrent предлагает свой набор потокобезопасных классов, допускающих разными потоками одновременное чтение и внесение изменений. Итераторы классов данного пакета представляют данные на определенный момент времени и не вызывают исключение ConcurrentModificationException. Все операции по изменению коллекции (add, set, remove) приводят к созданию новой копии внутреннего массива. Этим гарантируется, что при проходе итератором по коллекции не будет ConcurrentModificationException. Следует помнить, что при копировании массива копируются только ссылки на объекты.
CopyOnWriteArrayList реализует алгоритм CopyOnWrite и является потокобезопасным аналогом ArrayList. Класс CopyOnWriteArrayList содержит изменяемую ссылку на неизменяемый массив, обеспечивая преимущества потокобезопасности без необходимости использования блокировок. Т.е. при выполнении модифицирующей операции CopyOnWriteArrayList создаёт новую копию списка и гарантирует, что её итераторы вернут состояние списка на момент создания итератора и не вызовут ConcurrentModificationException. Описание CopyOnWriteArrayList с примером представлено здесь.
ConcurrentHashMap реализует (implements) интерфейс java.util.concurrent.ConcurrentMap и отличается от HashMap и Hashtable внутренней структурой хранения пар key-value. СoncurrentHashMap использует несколько сегментов, и данный класс можно рассматривать как группу HashMap’ов. По умолчанию количество сегментов равно 16. Доступ к данным определяется по сегментам, а не по объекту. Итераторы данного класса фиксируют структуру данных на момент начала его использования. Описание ConcurrentHashMap с примером представлено здесь.
CopyOnWriteArraySet выполнен на основе CopyOnWriteArrayList с реализацией интерфейса Set. Описание CopyOnWriteArraySet с примером представлено здесь.
ConcurrentNavigableMap расширяет возможности интерфейса NavigableMap для использования в многопоточных приложениях; итераторы класса декларируются как потокобезопасные и не вызывают ConcurrentModificationException.
ConcurrentSkipListMap является аналогом коллекции TreeMap с сортировкой данных по ключу и с поддержкой многопоточности.
ConcurrentSkipListSet выполнен на основе ConcurrentSkipListMap с реализацией интерфейса Set.
Concurrent Synchronizers, объекты синхронизации
Прежде чем говорить о синхронизаторах пакета java.util.concurrent, вспомним, что такое синхронизация.
Синхронизация — это процесс, позволяющий выполнять в программе синхронно параллельные потоки. Несколько потоков могут мешать друг другу при обращении к одним и тем же объектам приложения. Для решения этой проблемы используется мьютекс, он же монитор, имеющий два состояния — объект занят и объект свободен. Монитор (мьютекс) — это высокоуровневый механизм взаимодействия и синхронизации процессов, обеспечивающий доступ к неразделяемым ресурсам. Мьютекс встроен в класс Object и, следовательно, имеется у каждого объекта.
Синхронизация в Java реализуется использованием зарезервированного слова synchronized. Можно использовать synchronized в классах, определяя синхронизированные методы или блоки. Но нельзя использовать synchronized в переменных или атрибутах в определении класса.
Таким образом, при обычной синхронизации потоков используют оператор synchronized для ограничения (блокирования) доступа к определенному методу, блоку кода или объекту без каких-либо условий. Пакет java.util.concurrent содержит пять объектов синхронизации, позволяющих накладывать определенные условия для синхронизации потоков.
Semaphore (семафор) — объект синхронизации, ограничивающий одновременный доступ к общему ресурсу нескольким потокам с помощью счетчика. При запросе разрешения и значении счетчика больше нуля доступ предоставляется, а счетчик уменьшается; в противном случае — доступ запрещается. При освобождении ресурса значение счетчика семафора увеличивается. Количество разрешений семафора определяется в конструкторе. Второй конструктор семаформа включает дополнительный параметр «справедливости», определяющий порядок предоставления разрешения ожидающим доступа потокам. Описание с примером представлено здесь.
CountDownLatch («защелка с обратным отсчетом») — объект синхронизации потоков, блокирующий один или несколько потоков до тех пор, пока не будут выполнены определенные условия. Количество условий задается счетчиком. При обнулении счетчика, т.е. при выполнении всех условий, блокировки выполняемых потоков будут сняты и они продолжат выполнение кода. Примером CountDownLatch может служить экскурсовод, собирающий группу из заданного количества туристов. Как только группа собрана она отправляется на экскурсию. Необходимо отметить, что счетчик одноразовый и не может быть инициализирован по-новому. Описание с примером представлено здесь
CyclicBarrier — объект синхронизации типа «Барьер» используется, как правило, в распределённых вычислениях. Барьерная синхронизация останавливает участника (исполняемый поток) в определенном месте в ожидании прихода остальных потоков группы. Как только все потоки достигнут барьера, барьер снимается и выполнение потоков продолжается. Циклический барьер CyclicBarrier, также, как и CountDownLatch, использует счетчик и похож на него. Отличие связано с тем, что «защелку» нельзя использовать повторно после того, как её счётчик обнулится, а барьер можно использовать (в цикле). Описание с примером представлено здесь
Exchanger — объект синхронизации, используемый для двустороннего обмена данными между двумя потоками. При обмене данными допускается null значения, что позволяет использовать класс для односторонней передачи объекта или же просто, как синхронизатор двух потоков. Обмен данными выполняется вызовом метода exchange, сопровождаемый самоблокировкой потока. Как только второй поток вызовет метод exchange, то синхронизатор Exchanger выполнит обмен данными между потоками. Описание с примером представлено здесь
Phaser — объект синхронизации типа «Барьер», но, в отличие от CyclicBarrier, может иметь несколько барьеров (фаз), и количество участников на каждой фазе может быть разным. Описание с примером представлено здесь
Атомарные классы пакета java.concurrent.atomic
Пакет java.util.concurrent.atomic включает девять атомарных классов для выполнения, так называемых, атомарных операций. Операция является атомарной, если её можно безопасно выполнять при параллельных вычислениях в нескольких потоках, не используя при этом ни блокировок, ни синхронизацию synchronized.
Атомарный класс включает метод compareAndSet, реализующий механизм оптимистичной блокировки и позволяющий изменить значение только в том случае, если оно равно ожидаемому значению. Т.е. если значение атомарного класса было изменено в другом потоке, то оно не будет равно ожидаемому значению, и метод compareAndSet не позволит изменить значение.
Ряд архитектур процессоров имеют инструкцию Compare-And-Swap (CAS), которая реализует операцию compareAndSet. Таким образом, на уровне инструкций процессора имеется поддержка необходимой атомарной операции. В архитектурах процессоров, где инструкция не поддерживается, операции реализованы иными низкоуровневыми средствами.
Подробнее об описании атомарных классов с примерами можно познакомиться здесь.
Queues
Пакет java.util.concurrent содержит классы формирования неблокирующих и блокирующих очередей для многопоточных приложений. Неблокирующие очереди «заточены» на скорость выполнения, блокирующие очереди приостанавливают потоки при работе с очередью.
Неблокирующие очереди
Потокобезопасные и неблокирующие очереди на связанных узлах (linked nodes) реализуют интерфейс Queue и его наследника Deque.
ConcurrentLinkedQueue реализуют интерфейс Queue и формирует неблокирующую и ориентированную на многопоточное исполнение очередь. Размер очереди ConcurrentLinkedQueue не имеет ограничений. Имплементация очереди использует wait-free алгоритм от Michael & Scott, адаптированный для работы с garbage collector’ом. Данный алгоритм довольно эффективен и очень быстр, т.к. построен на CAS (Compare-And-Swap). Описание с примером представлено здесь.
ConcurrentLinkedDeque реализует интерфейс Deque (Double ended queue), читается как «Deck». Данная реализация позволяет добавлять и получать элемента с обеих сторон очереди. Соответственно, класс поддерживает оба режима работы : FIFO (First In First Out) и LIFO (Last In First Out). ConcurrentLinkedDeque следует использовать в том случае, если необходимо реализовывать LIFO, поскольку за счет двунаправленности данный класс проигрывает по производительности очереди ConcurrentLinkedQueue. Описание с примером представлено здесь.
При обработке большого количества потоков данных использование неблокирующих очередей иногда может оказаться явно недостаточным : разгребающие очереди потоки перестанут справляться с наплывом данных, что может привести к «out of memory» или перегрузить IO/Net настолько, что производительность упадет в разы, пока не наступит отказ системы по таймаутам или из-за отсутствия свободных дескрипторов в системе. Самое неприятное в данном случае то, что возникающая ситуация является нестабильной, сложной для отладки. Для таких случаев нужна блокирующая очередь с возможностью задать её размер и/или условия блокировки.
Интерфейсы блокирующих очередей наряду с возможностью определения размера очереди включают методы, по-разному реагирующие на незаполнение или переполнение queue. Так, например, при добавлении элемента в переполненную очередь, один из методов вызовет IllegalStateException, другой вернет false, третий заблокирует поток, пока не появится место, четвертый же заблокирует поток на определенное время (таймаут) и вернет false, если место так и не появится.
ArrayBlockingQueue — блокирующая очередь, реализующая классический кольцевой буфер. Параметр fair в конструкторе позволяет управлять справедливостью очереди для упорядочивания работы ожидающих потоков производителей (вставляющих элементы) и потребителей (извлекающих элементы). Описание с примером представлено здесь.
LinkedBlockingQueue — блокирующая очередь на связанных узлах, реализующая «two lock queue» алгоритм : один lock добавляет элемент, второй извлекает. За счет двух lock’ов данная очередь показывает более высокую производительность по сравнению с ArrayBlockingQueue, но и расход памяти повышается. Размер очереди задается через конструктор и по умолчанию равен Integer.MAX_VALUE. Описание с примером представлено здесь.
LinkedBlockingDeque — двунаправленная блокирующая очередь на связанных узлах, реализованная как простой двунаправленный список с одним локом. Размер очереди задается через конструктор и по умолчанию равен Integer.MAX_VALUE. Описание с примером представлено здесь.
SynchronousQueue — блокирующая очередь, в которой каждая операция добавления должна ждать соответствующей операции удаления в другом потоке и наоборот. Т.е. очередь реализует принцип «один вошел, один вышел». SynchronousQueue не имеет никакой внутренней емкости, даже емкости в один элемент. Описание с примером представлено здесь.
LinkedTransferQueue — блокирующая очередь с реализацией интерфейса TransferQueue, который позволяет при добавлении элемента в очередь заблокировать вставляющий поток до тех пор, пока другой поток не заберет элемент из очереди. Блокировка может быть как с таймаутом, так и с проверкой ожидающего потока. Таким образом может быть реализован механизм передачи сообщений с поддержкой как синхронных, так и асинхронных сообщений. Подробное описание представлено здесь.
DelayQueue — специфичный вид очереди, позволяющий извлекать элементы только после некоторой задержки, определенной в каждом элементе через метод getDelay интерфейса Delayed. Подробное описание представлено здесь.
PriorityBlockingQueue — является многопоточной оберткой интерфейса PriorityQueue. При размещении элемента в очереди, его порядок определяется в соответствии с «натуральным» упорядочиванием, либо логикой Comparator’а или имплементации Comparable интерфейса. Подробное описание представлено здесь.
Описание пакета java.concurrent.locks
Пакет java.util.concurrent.locks включает классы, которые существенно отличаются от встроенной синхронизации и мониторов, и которые можно использовать для блокировки ресурсов с определенными условиями. Этот пакет дает намного большую гибкость в использовании блокировок без условий и с условием.
Lock — базовый интерфейс, предоставляющий более гибкий подход при ограничении доступа к ресурсам/блокам по сравнению с использованием synchronized. Так, при использовании нескольких блокировок, порядок их освобождения может быть произвольный. Имеется возможность перехода к альтернативному сценарию, если блокировка уже захвачена. Более подробное описание Lock с примером представлено здесь.
Condition — интерфейсное условие в сочетании с блокировкой Lock позволяет заменить методы монитора/мьютекса (wait, notify и notifyAll) объектом, управляющим ожиданием событий. Объект с условием чаще всего получается из блокировок с использованием метода lock.newCondition(). Таким образом можно получить несколько комплектов wait/notify для одного объекта. Блокировка Lock заменяет использование synchronized, а Condition — объектные методы монитора. Более подробное описание Condition с примером представлено здесь.
ReadWriteLock — интерфейс создания read/write блокировок, который реализует один единственный класс ReentrantReadWriteLock. Блокировку чтение-запись следует использовать при длительных и частых операциях чтения и редких операциях записи. Тогда при доступе к защищенному ресурсу используются разные методы блокировки, как показано ниже :
Более подробное описание интерфейса ReadWriteLock здесь.
Описание Executors
Многопоточный пакет concurrent включает средства, называемые сервисами исполнения, позволяющие управлять потоковыми задачами с возможностью получения результатов через интерфейсы Future и Callable.
Callable — расширенный аналог интерфейса Runnable, позволяющий возвращать типизированное значение. В интерфейсе используется метод call, являющийся аналогом метода run интерфейса Runnable. Более подробное описание интерфейса Callable с примером представлено здесь.
FutureTask — класс-оболочка, базирующаяся на конкретной реализации интерфейса Future. Чтобы создать реализацию класса FutureTask необходим объект Callable. FutureTask представляет удобный механизм для превращения Callable одновременно в Future и Runnable, реализуя оба интерфейса. Имплементация FutureTask может быть передана на выполнение классу, реализующему интерфейс Executor, либо запущена в отдельном потоке, как класс, реализующий интерфейс Runnable. Более подробное описание интерфейса FutureTask с примером представлено здесь.
ExecutorService — представляет собой интерфейс, имплементация которого используется для запуска потоков. Потоки можно запускать, используя методы execute и submit. Оба метода в качестве параметра принимают объекты Runnable или Callable. Метод submit возвращает значение типа Future, позволяющий получить результат выполнения потока. Метод invokeAll работает со списками задач с блокировкой потока до завершения всех задач в переданном списке или до истечения заданного времени. Метод invokeAny блокирует вызывающий поток до завершения любой из переданных задач. Реализация данного интерфейса включает метод shutdown, позволяющий завершить все принятые на исполнение задачи и блокирует поступление новых.
Интерфейс ScheduledExecutorService расширяет свойства ExecutorService для поддержки планирования потоков исполнения. В пакет concurrent включены три предопределенных класса исполнителей: ThreadPoolExecutor, ScheduledThreadPoolExecutor и ForkJoinPool. Чтобы получить реализацию данных объектов необходимо использовать класс-фабрику Executors. Описание интерфейса ExecutorService с примером представлено здесь
