что такое барьер в канкаренси

Русские Блоги

Java Concurrency CyclicBarrier (барьер синхронизации)

CyclicBarrier (синхронный барьер)

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

Использование барьера синхронизации. Существует два способа использования барьера синхронизации, как видно из метода построения CyclicBarrier.

Конструктор метод 1:
Передавайте количество потоков, чтобы заблокировать стороны и объект Runnable, роль этого объекта используется для решения следующих сложных бизнес-сценариев: когда первый поток должен достичь барьера Перед обработкой задачи эта задача может быть записана в барьерном действии.

Конструктор метод 2:
Если объект runnable равен нулю, ему просто нужно просто дождаться, пока другие потоки достигнут барьера синхронизации.

Далее проиллюстрируем использование двух барьеров синхронизации

Метод строительства 2:
(1) создайте объект CyclicBarrier и передайте количество потоков, которые должны быть заблокированы в барьере синхронизации;

(2) Вызовите метод cb.await () в каждом дочернем потоке, который нужно заблокировать в барьере синхронизации;

Выходные результаты:

Выходной результат может быть не 1 2, но также может быть 2 1

Конструктор 1:

(1) Создайте объект CyclicBarrier, передайте количество потоков, которые должны быть заблокированы в барьере синхронизации, и барьер объект действия;

(2) Написать класс, который реализует интерфейс Runnable для реализации бизнес-логики до того, как первый поток достигнет барьера синхронизации;

(3) Вызвать метод cb.await () в каждом дочернем потоке, который нужно заблокировать в барьере синхронизации;

Выходные результаты:

Лично считаю, что порядок 1 и 2 является неопределенным, но первым должен быть выведен «случай-перед CyclicBarrier»

Поскольку метод cb.await () выбрасывает InterruptedException и BrokenBarrierException, для перехвата этих двух исключений следует использовать метод try-catch в дочернем потоке;

Счетчик потока в CyclicBarrier можно сбросить с помощью метода reset ();

После того, как все потоки достигнут барьера синхронизации, не все потоки начинают выполняться «одновременно», но время запуска каждого потока минимизируется;

Используйте сценарий:
Сценарий, в котором необходимы многопоточные результаты вычислений для окончательного объединения этих результатов;

Разница между CountDwonLatch и CyclicBarrier

1. CountDwonLatch блокирует один поток, CyclicBarrier блокирует несколько потоков;

2. Счетчик CountDwonLatch не может быть изменен при создании объекта, но CyclicBarrier может использовать метод reset () для сброса счетчика;

3. Функция CyclicBarrier более распространена. Например, метод int getWaitingNumber () возвращает количество потоков, заблокированных в барьере синхронизации. Логический метод isBroken () возвращает информацию о том, были ли прерваны какие-либо заблокированные потоки.

Источник

Синхронизаторы пакета concurrent

Объекты синхронизации Synchroniser из пакета java.util.concurrent включают :

Объект синхронизации Semaphore

Как было отмечено выше, Semaphore ограничивает доступ к определенному участку кода, иначе говоря, к общему ресурсу, в качестве которого могут выступать программые/аппаратные ресурсы или файловая система.

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

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

Конструкторы Semaphore

Класс Semaphore имеет два приведенных ниже конструктора :

Параметр permits определяет исходное значение счетчика разрешений, т.е. количество потоков исполнения, которым может быть одновременно предоставлен доступ к общему ресурсу. По умолчанию ожидающим потокам предоставляется разрешение в неопределенном порядке. Если же использовать второй конструктор и параметру справедливости fair присвоить значение true, то разрешения будут предоставляться ожидающим потокам исполнения в том порядке, в каком они его запрашивали.

Метод получения разрешения acquire

Чтобы получить у семафора разрешение необходимо вызвать у него один из перегруженных методов acquire :

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

Освобождение ресурса

Чтобы освободить разрешение у семафора следует вызвать у него один из перегруженных методов release :

В первом методе освобождается одно разрешение, а во втором — количество разрешений, обозначенное параметром number.

Пример использования Semaphore

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

Общий ресурс CONTROL_PLACES, символизирующий контролеров и используемый всеми потоками, выделен оператором synchronized. С помощью семафора осуществляется контроль доступа только одному потоку.

Результат выполнения примера с семафором

Обратите внимание, что «всадник 3» завершил проверку, после чего «всадник 6» вошел в блокируемый участок кода. А вот «всадник 7» успел раньше вывести сообщение о входе в блокируемый участок кода, чем «всадник 1» сообщил о его «покидании».

Объект синхронизации CountDownLatch

Объект синхронизации потоков CountDownLatch представляет собой «защелку с обратным отсчетом» : несколько потоков, выполняя определенный код, блокируются до тех пор, пока не будут выполнены заданные условия. Количество условий определяются счетчиком. Как только счетчик обнулится, т.е. будут выполнены все условия, самоблокировки выполняемых потоков снимаются, и они продолжают выполнение кода.

Таким образом, CountDownLatch также, как и Semaphore, работает со счетчиком, обнуление которого снимает самоблокировки выполняемых потоков. Конструктор CountDownLatch :

Параметр number определяет количество событий, которые должны произойти до того момента, когда будет снята самоблокировка.

Метод самоблокировки await

CountDownLatch имеет два перегруженных метода await для самоблокировки :

В первом методе ожидание длится до тех пор, пока счетчик CountDownLatch не достигнет нуля. А во втором методе ожидание длится только в течение определенного периода времени, определяемого параметром ожидание wait. Время ожидания указывается в единицах unit объекта перечисления TimeUnit, определяюший временно́е разбиение. Существуют следующие значения данного перечисления :

Метод уменьшения счетчика countDown

Чтобы уменьшить счетчик объекта CountDownLatch следует вызвать метод countDown :

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

Пример использования CountDownLatch

В примере несколько всадников должны подъехать к барьеру. Как только все всадники выстроятся перед барьером, будут даны команды «На старт», «Внимание», «Марш». После этого барьер опустится и начнутся скачки. Объект синхронизации CountDownLatch выполняет роль счетчика количества всадников и команд.

При «выходе на старт» поток вызывает методы countDown, уменьшая значение счетчика на 1, и await, блокируя самого себя в ожидании обнуления счетчика «защелки». Как только все потоки выстроятся на «старте» с интервалом в 1 сек. подаются команды. Каждая команда сопровождается уменьшением счетчика. После обнуления счетчика «защелки» потоки продолжают выполнение дальнейшего кода.

Результат выполнения примера

Объект синхронизации CyclicBarrier

Объект синхронизации CyclicBarrier представляет собой барьерную синхронизацию, используемую, как правило, в распределённых вычислениях. Особенно эффективно использование барьеров при циклических расчетах. При барьерной синхронизации алгоритм расчета делят на несколько потоков. С помощью барьера организуют точку сбора частичных результатов вычислений, в которой подводится итог этапа вычислений.

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

Класс CyclicBarrier имеет 2 конструктора :

В первом конструкторе задается количество потоков, которые должны достигнуть барьера, чтобы после этого одновременно продолжить выполнение кода. Во втором конструкторе дополнительно задается реализующий интерфейс Runnable класс, который должен быть запущен после прихода к барьеру всех потоков. Поток запускать самостоятельно НЕ НУЖНО. CyclicBarrier это делает автоматически.

Для указания потоку о достижении барьера нужно вызвать один из перегруженных методов await :

Назначение параметров wait и unit у второго метода описано выше (см. CountDownLatch).

Циклический барьер CyclicBarrier похож на CountDownLatch. Главное различие между ними связано с тем, что «защелку» нельзя использовать повторно после того, как её счётчик обнулится, а барьер можно использовать (в цикле). С точки зрения API циклический барьер CyclicBarrier имеет только метод самоблокировки await и не имеет метода декрементации счетчика, а также позволяет подключить и автоматически запускать дополнительный потоковый класс при достижении барьера всех исполняемых потоков.

Пример использования CyclicBarrier

В примере организуется переправа. Паром может вместить только 3 автомобиля. Количество автомобилей 9. Роль парома выполняет объект синхронизации FerryBarrier, которому в качестве второго параметра передается реализующий интерфейс Runnable класс FerryBoat. Как только 3 потока достигнут барьера автоматически будет запущен FerryBoat, после завершения работы которого потоки продолжают свою работу.

Обратите внимание, что потоки подходят к барьеру с интервалом в 400 ms. Время задержки у барьера/переправы (после того, как собралось необходимое количество потоков), составляет 500 ms, если не считать время вывода сообщений в консоль. За это время к барьеру успевает подойти еще один поток. Что мы и видим при выводе сообщений в консоль.

Результат выполнения примера

Варьируя временем на переправе и временем прихода автомобилей на переправу, можно либо заставить паром простаивать, либо будут простаивать автомобили на переправе.

Объект синхронизации Exchanger

Класс Exchanger (обменник) предназначен для упрощения процесса обмена данными между двумя потоками исполнения. Принцип действия класса Exchanger связан с ожиданием того, что два отдельных потока должны вызвать его метод exchange. Как только это произойдет, Exchanger произведет обмен данными, предоставляемыми обоими потоками.

Обменник является обобщенным классом, он параметризируется типом объекта передачи :

Необходимо отметить, что обменник поддерживает передачу NULL значения, что дает возможность использовать его для передачи объекта в одну сторону или места синхронизации двух потоков.

Exchanger содержит перегруженный метод exchange, имеющий следующие формы :

Параметр buffer является ссылкой на обмениваемые данные. Метод возвращает данные из другого потока исполнения. Вторая форма метода позволяет определить время ожидания. Параметры wait и unit описаны выше. Метод exchange, вызванный в одном потоке, не завершится успешно до тех пор, пока он не будет вызван из второго потока исполнения.

Пример использования Exchanger

В примере использования объекта синхронизации Exchanger два почтальона из пунктов А и Б отправляются в соседние поселки В и Г доставить письма. Каждый из почтальонов должен доставить по письму в каждый из поселков. Чтобы не делать лишний круг, они встречаются в промежуточном поселке Д и обмениваются одним письмом. В результате этого каждому из почтальонов придется доставить письма только в один поселок. В примере все «шаги» почтальонов фиксируются выводом соответствующих сообщений в консоль.

Результат выполнения примера

В консоль будет выведена следующая информация :

Объект синхронизации Phaser

Phaser (фазировщик), как и CyclicBarrier, является реализацией объекта синхронизации типа «Барьер» (CyclicBarrier). В отличии от CyclicBarrier, Phaser предоставляет больше гибкости. Чтобы лучше понять Phaser, можно привести два наглядно демонстрирующих его использование примера.

В качестве первого примера можно рассмотреть несколько потоков исполнения, реализующих процесс обработки заказов из трех стадий. На первой стадии отдельные потоки исполнения проверяют сведения о клиенте, наличие товара на складе и их стоимость. На второй стадии вычисляется стоимость заказа и стоимость доставки. На заключительной стадии подтверждается оплата и определяется ориентировочное время доставки. Во втором примере несколько потоков реализуют перевозку пассажиров городским транспортом. Пассажиры ожидают транспорт на разных остановках. Транспорт, останавливаясь на остановках, одних пассажиров «сажает», других «высаживает».

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

Важные особенности Phaser :

Для создания объекта Phaser используется один из конструкторов :

Параметр parties определяет количество участников, которые должны пройти все фазы. Первый конструктор создает объект Phaser без каких-либо участников. Второй конструктор регистрирует передаваемое в конструктор количество участников. Третий и четвертый конструкторы дополнительно устанавливают родительский объект Phaser.

При создании экземпляр класса Phaser находится в нулевой фазе. В очередном состоянии (фазе) синхронизатор находится в ожидании до тех пор, пока все зарегистрированные потоки не завершат данную фазу. Потоки извещают об этом, вызывая один из методов arrive() или arriveAndAwaitAdvance().

Методы объекта синхронизации Phaser

Следующее анимационное изображение наглядно демонстрирует работу объекта синхронизации Phaser — участник регистрируется в определенной фазе синхронизатора; при переходе синхронизатора в заданное состояние (фазу) участник снимается с регистрации. Количество участников в разных фазах синхронизатора может отличаться.

Пример использования Phaser

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

Листинг класса Passenger

Конструктор класса Passenger получает значение идентификатора, номера станций посадки и назначения (высадки). При создании объекта в консоль выводится информация о пассажире (метод toString).

Как только Phaser переходит в определенную фазу, номер которой соответствует станции посадки пассажира, то поток данного Passenger стартует (run) и выводит в консоль сообщение, что пассажир вошел в вагон, т.е. находится в ожидании следующей станции/фазы (arriveAndAwaitAdvance). Если следующая станция/фаза не будет соответствовать станции назначения, то Passenger продолжит свой путь. Как только Phaser перейдет в фазу, номер которой соответствует номеру станции назначания пассажира, то цикл контроля завершится и поток продолжит работу. С задержкой в 500 ms он сообщит, что покинул вагон и отменит регистрацию в Phaser (arriveAndDeregister).

Таким образом, поток/пассажир дожидается свой фазы/станции в цикле, выделенной в коде пунктирными комментариями. Вызов метода arriveAndAwaitAdvance возвращает значение следующего номера фазы, т.е. участник будет вызван при переходе Phaser в новое состояние. Если в этом состоянии значение фазы (getPhase) будет соответствовать номеру destination, то цикл прервется, в противном случае, ожидание следующей фазы и повторное выполнение проверки условия while.

Примечание : Passenger является внутренним классом примера/класса PhaserExample, и для описания вынесен из общего кода, чтобы не загромождать листинг.

Листинг примера PhaserExample

В примере сначала создается объект синхронизации PHASER. После этого формируется массив пассажиров. При создании объекта Passenger случайным образом определяются станция посадки и станция назначения. После того, как массив пассажиров подготовлен, PHASER в цикле начинает менять свое состояние. На каждом шаге выполняется проверка «станции посадки пассажира». Если она соответствует значению фазы, то данный пассажир входит в вагон метро, т.е. регистрируется в PHASER и поток стартует. Таким образом, регистрация участников (исполнительных потоков) выполняется при нахождении PHASER в определенном состоянии/фазе. Пассажир покинет вагон при достижении метро станции назначения, т.е. при нахождении PHASER в соответствующей фазе. Но это произойдет уже в коде класса Passenger, рассмотренного выше.

Источник

Уровень 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 :

Источник

Канкаренси, BlockingQueues (java7)

— Я тебе сегодня расскажу про «канкаренси» — Concurrency.

Concurrency – это библиотека классов в Java, в которой собрали специальные классы, оптимизированные для работы из нескольких нитей. Эта тема очень интересная и обширная. Но сегодня мы просто познакомимся с ней. Эти классы собраны в пакете java.util.concurrent. Я расскажу про пару интересных классов.

Атомарные типы.

Ты уже знаешь, что даже операция count++ не является потокобезопасной (thread-safe). При увеличении переменной на 1 реально происходит три операции, в результате чего может возникнуть конфликт при одновременном изменении этой переменной.

— Ага, Элли рассказывала это немного раньше:

— Именно. Тогда в Java были добавлены типы данных, которые выполняют такие операции неразрывно – атомарно. (Атом — неделимый).

Так в Java появились типы AtomicInteger, AtomicBoolean, AtomicDouble и т.д.

Вот, допустим нам нужно сделать класс «счетчик»:

Как бы ты сделал его объекты thread-safe?

— Да, сделал бы все методы synchronized и все:

— Отличная работа. А вот, как бы он выглядел с использованием атомарных типов:

И твой и мой классы работают одинаково, но класс с AtomicInteger работает быстрее.

— Т.е. разница небольшая?

Да. Из своего опыта я могу посоветовать всегда в лоб использовать synchronized. И только когда весь код приложения уже написан и начинается процесс его оптимизации, то можно начинать переписывать его части с использованием Atomic-типов. Но в любом случае, я бы хотела, чтобы ты знал, что такие типы есть. Даже если не будешь активно ими пользоваться, ты всегда можешь увидеть код, написанный с их применением.

— Согласен, в это есть смысл.

— Кстати, а ты заметил, что атомарные типы – не immutable? AtomicInteger, в отличии от просто Integer, содержит методы по изменению своего внутреннего состояния.

— Понятно, прямо как String и StringBuffer.

— Да, что-то типа того.

Нитебезопасные (иногда называют — потокобезопасные) коллекции.

В качестве такой коллекции я хотела бы привести ConcurrentHashMap. Как сделать HashMap thread-safe?

— Сделать все его методы synchronized?

— Да, но представь теперь, что у тебя есть один такой SynchronizedHashMap, а к нему обращаются десятки нитей. И сто раз в секунду в этот map добавляется новая запись, при этом весь объект блокируется для чтения и записи.

— Да, но это же стандартный подход. Что тут можно сделать?

— Разработчики Java придумали несколько крутых штук.

Во-первых, они хранят данные в ConcurrentHashMap не одним куском, а разбивая их на порции — «корзины». И когда кто-то меняет данные в ConcurrentHashMap, то блокируется не весь объект, а только одна корзина, к которой происходит доступ. Т.е. на самом деле объект могут одновременно менять много нитей.

Во-вторых, помнишь проблему, что нельзя одновременно идти по элементам списка/мэпа и менять его? Такой код кинет исключение:

А в ConcurrentHashMap это можно:

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

— Ясно. Спасибо, Ким. Действительно, очень интересные классы. Надеюсь, когда-нибудь я тоже буду виртуозно ими владеть.

Источник