Как устроены атомики?
Начнем с того, что такое атомики и зачем нужны. Atomic* – семейство классов из java.util.concurrent . Они предоставляют набор атомарных операций для соответствующих типов. Например с помощью методов getAndIncrement / incrementAndGet класса AtomicInteger можно делать неатомарный в обычных условиях инкремент ( i++ ).
Условно можно разделить подходы реализации большинства atomic-методов на две группы: compare-and-set и set-and-get.
Методы категории compare-and-set принимают старое значение и новое. Если переданное старое значение совпало с текущим, устанавливается новое. Обычно делегируют вызов в методы класса Unsafe , которые заменяются нативными реализациями виртуальной машины. Виртуальная машина в большинстве случаев использует атомарную операцию процессора compare-and-swap (CAS). Поэтому атомики обычно более эффективны чем стандартная дорогостоящая блокировка.
В случае set-and-get старое значение неизвестно. Поэтому нужен небольшой трюк: программа сначала считывает текущее значение, а затем записывает новое, тоже с помощью CAS, потому что запись могла успеть поменяться даже за этот шаг. Эта попытка чтения+записи повторяется в цикле, пока старое значение не совпадет и переменная не будет успешно записана.
Этот трюк называется double-checked или optimistic locking, и может быть использован и в пользовательском коде с любым способом синхронизации. Оптимистичность заключается в том, что мы надеемся что состояния гонки нет, прибегая к синхронизации только если гонка всё же случилась. Реализация оптимистичной блокировки может быть дана как отдельная задача.
Атомарные классы пакета util.concurrent
Пакет java.util.concurrent.atomic содержит девять классов для выполнения атомарных операций. Операция называется атомарной, если её можно безопасно выполнять при параллельных вычислениях в нескольких потоках, не используя при этом ни блокировок, ни синхронизацию synchronized. Прежде, чем перейти к рассмотрению атомарных классов, рассмотрим выполнение наипростейших операций инкремента и декремента целочисленных значений.
С точки зрения программиста операции инкремента (i++, ++i) и декремента (i—, —i) выглядят наглядно и компактно. Но, с точки зрения JVM (виртуальной машины Java) данные операции не являются атомарными, поскольку требуют выполнения нескольких действительно атомарных операции: чтение текущего значения, выполнение инкремента/декремента и запись полученного результата. При работе в многопоточной среде операции инкремента и декремента могут стать источником ошибок. Т.е. в многопоточной среде простые с виду операции инкремента и декремента требуют использование синхронизации и блокировки. Но блокировки содержат массу недостатков, и для простейших операций инкремента/декремента являются тяжеловесными. Выполнение блокировки связано со средствами операционной системы и несёт в себе опасность приостановки с невозможностью дальнейшего возобновления потока, а также опасность взаимоблокировки или инверсии приоритетов (priority inversion). Кроме этого, появляются дополнительные расходы на переключение потоков. Но можно ли обойтись без блокировок? В ряде случаев можно!
Блокировка подразумевает пессимистический подход, разрешая только одному потоку выполнять определенный код, связанный с изменением значения некоторой «общей» переменной. Таким образом, никакой другой поток не имеет доступа к определенным переменным. Но можно использовать и оптимистический подход. В этом случае блокировки не происходит, и если поток обнаруживает, что значение переменной изменилось другим потоком, то он повторяет операцию снова, но уже с новым значением переменной. Так работают атомарные классы.
Описание атомарного класса AtomicLong
Рассмотрим принцип действия механизма оптимистической блокировки на примере атомарного класса AtomicLong, исходный код которого представлен ниже. В этом классе переменная value объявлена с модификатором volatile, т.е. её значение могут поменять разные потоки одновременно. Модификатор volatile гарантирует выполнение отношения happens-before, что ведет к тому, что измененное значение этой переменной увидят все потоки.
Каждый атомарный класс включает метод compareAndSet, представляющий механизм оптимистичной блокировки и позволяющий изменить значение value только в том случае, если оно равно ожидаемому значению (т.е. current). Если значение value было изменено в другом потоке, то оно не будет равно ожидаемому значению. Следовательно, метод compareAndSet вернет значение false, что приведет к новой итерации цикла while в методе getAndAdd. Таким образом, в очередном цикле в переменную current будет считано обновленное значение value, после чего будет выполнено сложение и новая попытка записи получившегося значения (т.е. next). Переменные current и next — локальные, и, следовательно, у каждого потока свои экземпляры этих переменных.
private volatile long value; public final long get() < return value; >public final long getAndAdd(long delta) < while (true) < long current = get(); long next = current + delta; if (compareAndSet(current, next)) return current; >>
Метод compareAndSet реализует механизм оптимистической блокировки. Знакомые с набором команд процессоров специалисты знают, что ряд архитектур имеют инструкцию Compare-And-Swap (CAS), которая является реализацией этой самой операции. Таким образом, на уровне инструкций процессора имеется поддержка необходимой атомарной операции. На архитектурах, где инструкция не поддерживается, операции реализованы иными низкоуровневыми средствами.
Основная выгода от атомарных (CAS) операций появляется только при условии, когда переключать контекст процессора с потока на поток становится менее выгодно, чем немного покрутиться в цикле while, выполняя метод boolean compareAndSwap(oldValue, newValue). Если время, потраченное в этом цикле, превышает 1 квант потока, то, с точки зрения производительности, может быть невыгодно использовать атомарные переменные.
Список атомарных классов
Атомарные классы пакета java.util.concurrent.atomic можно разделить на 4 группы :
| • AtomicBoolean • AtomicInteger • AtomicLong • AtomicReference |
Atomic-классы для boolean, integer, long и ссылок на объекты. Классы этой группы содержат метод compareAndSet, принимающий 2 аргумента : предполагаемое текущее и новое значения. Метод устанавливает объекту новое значение, если текущее равно предполагаемому, и возвращает true. Если текущее значение изменилось, то метод вернет false и новое значение не будет установлено. Кроме этого, классы имеют метод getAndSet, который безусловно устанавливает новое значение и возвращает старое. Классы AtomicInteger и AtomicLong имеют также методы инкремента/декремента/добавления нового значения. |
| • AtomicIntegerArray • AtomicLongArray • AtomicReferenceArray |
Atomic-классы для массивов integer, long и ссылок на объекты. Элементы массивов могут быть изменены атомарно. |
| • AtomicIntegerFieldUpdater • AtomicLongFieldUpdater • AtomicReferenceFieldUpdater |
Atomic-классы для обновления полей по их именам с использованием reflection. Смещения полей для CAS операций определяется в конструкторе и кэшируются. Сильного падения производительности из-за reflection не наблюдается. |
| • AtomicStampedReference • AtomicMarkableReference |
Atomic-классы для реализации некоторых алгоритмов, (точнее сказать, уход от проблем при реализации алгоритмов). Класс AtomicStampedReference получает в качестве параметров ссылку на объект и int значение. Класс AtomicMarkableReference получает в качестве параметров ссылку на объект и битовый флаг (true/false). |
Полная документация по атомарным классам на английском языке представлена на оффициальном сайте Oracle. Наиболее часто используемые классы (не трудно догадаться) сосредоточены в первой группе.
Производительность атомарных классов
Согласно множеству источников неблокирующие алгоритмы в большинстве случаев более масштабируемы и намного производительнее, чем блокировки. Это связано с тем, что операции CAS реализованы на уровне машинных инструкций, а блокировки тяжеловесны и используют приостановку и возобновление потоков, переключение контекста и т.д. Тем не менее, блокировки демонстрируют лучший результат только при очень «высокой конкуренции», что в реальной жизни встречается не так часто.
Основной недостаток неблокирующих алгоритмов связан со сложностью их реализации по сравнению с блокировками. Особенно это касается ситуаций, когда необходимо контролировать состояние не одного поля, а нескольких.
Пример неблокирующего генератора последовательности
Рассмотрим генерирующий последовательность [1, 2, 4, 8, 16, . ] класс SequenceGenerator, функционирующий в многопоточной среде.
Листинг класса SequenceGenerator для генерирования последовательности
Для работы в многопоточной среде без блокировок используем атомарную ссылку AtomicReference, которая обеспечит хранение целочисленного значения типа java.math.BigInteger. Метод next возвращает текущее значение; переменная next вычисляет следующее значение. Метод compareAndSet атомарного класса element обеспечивает сохранение нового значения, если текущее не изменилось. Таким образом, метод next возвращает текущее значение и увеличивает его в 2 раза.
import java.math.BigInteger; import java.util.concurrent.atomic.AtomicReference; public class SequenceGenerator < private static BigInteger MULTIPLIER; private AtomicReferenceelement; public SequenceGenerator() < if (MULTIPLIER == null) MULTIPLIER = BigInteger.valueOf(2); element = new AtomicReference( BigInteger.ONE); > public BigInteger next() < BigInteger value; BigInteger next; do < value = element.get(); next = value.multiply(MULTIPLIER); >while (!element.compareAndSet(value, next)); return value; > >
Листинг последовательности Sequence
Для тестирования генератора последовательности SequenceGenerator используем класс Sequence, реализующий интерфейс Runnable. В качестве параметра конструктор класса получает идентификатор потока id, размер последовательности count и генератор последовательности sg. В методе run в цикле с незначительными задержками формируется последовательность чисел sequence. После завершения цикла значения последовательности «выводятся» в консоль методом printSequence.
import java.math.BigInteger; import java.util.ArrayList; import java.util.List; class Sequence implements Runnable < Thread thread; int id; int count; SequenceGenerator sg; Listsequence; sequence = new ArrayList(); boolean printed = false; Sequence(final int id, final int count, SequenceGenerator sg) < this.count = count; this.id = id; this.sg = sg; thread = new Thread(this); System.out.println("Создан поток " + id); thread.start(); >@Override public void run() < try < for (int i = 0; i < count; i++) < sequence.add(sg.next()); Thread.sleep((long) ( (Math.random()*2 + 1)*30)); >> catch (InterruptedException e) < System.out.println("Поток " + id + " прерван"); >System.out.print("Поток " + id + " завершён"); printSequence(); > public void printSequence() < if (printed) return; String tmp = "["; for (int i = 0; i < sequence.size(); i++) < if (i >0) tmp += ", "; String nb = String.valueOf(sequence.get(i)); while (nb.length() < 9) nb = " " + nb; tmp += nb; >tmp += "]"; System.out.println("Последовательность потока " + id + " : " + tmp); printed = true; > >
Листинг примера SequenceGeneratorExample
В примере SequenceGeneratorExample сначала создается генератор последовательности SequenceGenerator. После этого в цикле формируется массив из десяти Sequence, которые в паралелльных потоках по три раза обращаются к генератору последовательсности.
public class SequenceGeneratorExample < public static void main(String[] args) < SequenceGenerator sg = new SequenceGenerator(); Listsequences = new ArrayList(); for (int i = 0; i < 10; i++) < Sequence seq = new Sequence(i + 1, 3, sg); sequences.add(seq); >System.out.println("\nРасчет последовательностей\n"); int summa; // Ожидания завершения потоков do < summa = 0; for (int i = 0; i < sequences.size(); i++) < if (!sequences.get(i).thread.isAlive()) < sequences.get(i).printSequence(); summa++; >> try < Thread.sleep(100); >catch (InterruptedException e) <> > while (summa < sequences.size()) ; System.out.println("\n\nРабота потоков завершена"); System.exit(0); >>
Результаты выполнения примера
При выполнении примера в консоль будет выведена следующая информация :
Создан поток 0 Создан поток 1 Создан поток 2 Создан поток 3 Создан поток 4 Создан поток 5 Создан поток 6 Создан поток 7 Создан поток 8 Создан поток 9 Расчет последовательностей Поток 7 завершён Последовательность потока 7 : [ 256, 4096, 524288] Поток 5 завершён Поток 4 завершён Поток 1 завершён Последовательность потока 1 : [ 2, 1024, 2097152] Последовательность потока 4 : [ 16, 8192, 8388608] Последовательность потока 5 : [ 64, 2048, 32768] Поток 9 завершён Поток 3 завершён Поток 6 завершён Последовательность потока 3 : [ 8, 131072, 134217728] Последовательность потока 6 : [ 32, 16384, 268435456] Последовательность потока 9 : [ 512, 262144, 16777216] Поток 0 завершён Поток 2 завершён Поток 8 завершён Последовательность потока 0 : [ 1, 65536, 67108864] Последовательность потока 2 : [ 4, 1048576, 33554432] Последовательность потока 8 : [ 128, 4194304, 536870912] Работа потоков завершена
Каждый поток в цикле сформировал целочисленный массив из 3-х значений при обращении к «атомарному» генератору последовательности. Как видно из результатов выполнения примера, значения не пересекаются.
Скачать примеры
Рассмотренный на странице пример использования атомарного класса в виде проекта Eclipse можно скачать здесь (7.41 Кб).
Java Blog
Чтобы разобрать использование атомарных операций в Java рассмотрим следующий пример:
public class Counter < int counter; public void increment() < counter++; >>
В случае однопоточной среды это работает отлично; однако, как только мы разрешаем запись более чем одному потоку, мы начинаем получать противоречивые результаты.
Причина этому в операции приращения (counter++), которая может выглядеть как атомарная операция, но на самом деле представляет собой комбинацию трех операций: получение значения, увеличение и обратная запись обновленного значения.
Если два потока попытаются получить и обновить значение одновременно, это может привести к потере обновлений.
Один из способов управления доступом к объекту — использование блокировок. Этого можно достичь, используя ключевое слово synchronized в сигнатуре метода приращения. Ключевое слово synchronized гарантирует, что только один поток может входить в метод одновременно:
public class SynchronizedCounterWithLock < private volatile int counter; public synchronized void increment() < counter++; >>
Кроме того, нам нужно добавить ключевое слово volatile, чтобы обеспечить надлежащую видимость ссылок среди потоков.
Использование блокировок решает проблему. Однако производительность страдает в этом случае.
Когда несколько потоков пытаются получить блокировку, один из них выигрывает, а остальные потоки либо блокируются, либо приостанавливаются.
Процесс приостановки и последующего возобновления потока очень дорог и влияет на общую эффективность системы.
В небольшой программе, такой как счетчик, время, затрачиваемое на переключение контекста, может быть намного больше, чем фактическое выполнение кода, что значительно снижает общую эффективность.
Атомарные операции
Существует направление исследований, посвященное созданию неблокирующих алгоритмов для конкурентных сред. Эти алгоритмы используют низкоуровневые атомарные машинные инструкции, такие как сравнение и замена (CAS, compare-and-swap), для обеспечения целостности данных.
Типичная операция CAS работает с тремя операндами:
- Место в памяти для работы (M)
- Существующее ожидаемое значение (A) переменной
- Новое значение (B), которое необходимо установить
Операция CAS атомарно обновляет значение в M до B, но только если существующее значение в M совпадает с A, в противном случае никаких действий не предпринимается.
В обоих случаях возвращается существующее значение в M. Это объединяет три шага — получение значения, сравнение значения и обновление значения — в одну операцию на машинном уровне.
Когда несколько потоков пытаются обновить одно и то же значение через CAS, один из них выигрывает и обновляет значение. Однако, в отличие от блокировок, никакой другой поток не приостанавливается; вместо этого им просто сообщают, что им не удалось обновить значение. Затем потоки могут перейти к дальнейшей работе, и переключение контекста полностью исключено.
Еще одно последствие состоит в том, что основная логика программы становится более сложной. Это потому, что мы должны обработать сценарий, когда операция CAS не удалась. Мы можем повторять его снова и снова, пока он не увенчается успехом, или мы можем ничего не делать и двигаться дальше в зависимости от варианта использования.
AtomicInteger
Рассмотрим методы приращения и уменьшения в классе AtomicInteger.
Первая группа методов обновляет значение и возвращает то, которое было до обновления:
public final int getAndIncrement() < for (;;) < int current = get(); int next = current + 1; if (compareAndSet(current, next)) return current; >> public final int getAndDecrement() < for (;;) < int current = get(); int next = current - 1; if (compareAndSet(current, next)) return current; >> public final int getAndAdd(int delta) < for (;;) < int current = get(); int next = current + delta; if (compareAndSet(current, next)) return current; >>
Вторая группа методов обновляет значение и возвращает то, которое стало после обновления:
public final int incrementAndGet() < for (;;) < int current = get(); int next = current + 1; if (compareAndSet(current, next)) return next; >> public final int decrementAndGet() < for (;;) < int current = get(); int next = current - 1; if (compareAndSet(current, next)) return next; >> public final int addAndGet(int delta) < for (;;) < int current = get(); int next = current + delta; if (compareAndSet(current, next)) return next; >>
Как видно в обоих случаях используется функция compareAndSet, представляющая операцию CAS (compare-and-swap), которая использует unsafe пакет:
public final boolean compareAndSet(int expect, int update)
- Пакет java.util.concurrent.atomic
- Spring Resource: предостережения относительно FileSystemResource
- Модульное (юнит) тестирование в Spring
Атомарные операции в Java
Давай разберем данный пример, который поможет понять работу атомарных операций:
public class Counter < int count; public void increment() < count++; >>
Когда у нас один поток, все работает классно, но если мы добавляем многопоточку, то получаем неправильные результаты, а все из-за того, что операция инкремента составляет не одну операцию, а три: запрос на получение текущего значения count , потом увелечение ее на 1 и запись снова в count .
И когда два потока захотят увеличить переменную, скорее всего, ты потеряешь данные. То есть оба потока получают 100, в результате оба запишут 101 вместо ожидаемого значения 102.
И как же это решить? Нужно использовать блокировки. Ключевое слово synchronized помогает решить данную проблему, использование этого слова дает вам гарантию, что один поток будет обращаться к методу единовременно.
public class SynchronizedCounterWithLock < private volatile int count; public synchronized void increment() < count++; >>
Плюс надо добавлять ключевое слово volatile , которое обеспечивает корректную видимость ссылок среди потоков. Мы разбирали его работу выше.
Но все же есть минусы. Самый большой — это производительность, в тот момент времени, когда много потоков пытаются получить блокировку и один получает возможность для записи, остальные потоки будут или заблокированы, или приостановлены до момента освобождения потока.
Все эти процессы, блокировка, переход в другой статус — очень дороги для производительности системы.
Атомарные операции
Алгоритм использует низкоуровневые машинные инструкции, такие как сравнение и замена (CAS, compare-and-swap, что обеспечивает целостность данных и по ним уже существует большое количество исследований).
Типичная операция CAS работает с тремя операндами:
- Место в памяти для работы (M)
- Существующее ожидаемое значение (A) переменной
- Новое значение (B), которое необходимо установить
CAS атомарно обновляет M до B, но только если значение M совпадает с A, в противном случае никаких действий предприниматься не будет.
В первом и втором случае вернут значение М. Это позволяет объединить три шага, а именно — получение значения, сравнение значения и его обновление. И это все превращается в одну операцию на машинном уровне.
В тот момент времени, когда многопоточное приложение обращается к переменной и пытается обновить его и применяется CAS, то один из потоков получит его и сможет обновить его. Но в отличии от блокировок, другие потоки просто получат ошибки о том, что им не удалось обновить значение. Потом они перейдут к дальнейшей работе, а переключение полностью исключено при таком типе работе.
При этом логика становится труднее из-за того, что мы должны обработать ситуацию, когда операция CAS не отработала успешно. Мы просто смоделируем код таким образом, чтобы он не двигался дальше, пока операция не произойдет успешно.
Знакомство с атомарными типами
Ты столкнулся с ситуацией, когда тебе нужно настроить синхронизацию для самой простой переменной типа int ?
Первый способ, который мы уже разобрали – это использование volatile + synchronized . Но есть еще специальные классы Atomic*.
Если у нас используется CAS, то операции работают быстрее по сравнению с первым способом. И в дополнение у нас есть специальные и очень удобные методы для добавления значения и операции инкремента и декремента.
AtomicBoolean , AtomicInteger , AtomicLong , AtomicIntegerArray , AtomicLongArray —классы в которых операции атомарны. Ниже мы разберем работу с ними.
AtomicInteger
Класс AtomicInteger предоставляет операции с значением int , которые могут быть прочитаны и записаны атомарно, в дополнение содержит расширенные атомарные операции.
У него есть методы get и set , которые работают, как чтение и запись по переменным.
То есть “происходит до (happens-before)” с любым последующим получением той же переменной, о которой мы говорили ранее. У атомарного метода compareAndSet также есть эти особенности согласованности памяти.
Все операции, которые возвращают новое значение, выполняются атомарно:
| int addAndGet (int delta) | Добавляет определенное значение к текущему значению. |
| boolean compareAndSet (ожидаемое int, обновление int) | Устанавливает значение для данного обновленного значения, если текущее значение совпадает с ожидаемым значением. |
| int decrementAndGet () | Уменьшает на единицу текущее значение. |
| int getAndAdd (int delta) | Добавляет данное значение к текущему значению. |
| int getAndDecrement () | Уменьшает на единицу текущее значение. |
| int getAndIncrement () | Увеличивает на единицу текущее значение. |
| int getAndSet (int newValue) | Устанавливает заданное значение и возвращает старое значение. |
| int incrementAndGet () | Увеличивает на единицу текущее значение. |
| lazySet (int newValue) | В конце-концов устанавливается на заданное значение. |
| boolean weakCompareAndSet (ожидаемое, обновление int) | Устанавливает значение для данного обновленного значения, если текущее значение совпадает с ожидаемым значением. |
ExecutorService executor = Executors.newFixedThreadPool(5); IntStream.range(0, 50).forEach(i -> executor.submit(atomicInteger::incrementAndGet)); executor.shutdown(); executor.awaitTermination(Long.MAX_VALUE, TimeUnit.HOURS); System.out.println(atomicInteger.get()); // выведет 50