И операционная система сама позаботится о планировании этих потоков на процессорах по мере их доступности.

Таким образом вы получаете многопоточное выполнение.

Платформа Java обеспечивает поддержку многопоточности с помощью пакета java.util.concurrent.

В многопоточном программировании существуют две основные единицы исполнения — это процессы и потоки.

И многопоточное программирование на Java в основном касается потоков.

Чем отличается поток от процесса?

Процесс имеет автономную среду исполнения.

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

Процессы часто ассоциируются с приложением.

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

Для облегчения взаимодействия между процессами большинство операционных систем поддерживают Inter Process Communication (IPC).

IPC используется не только для связи между процессами в одной и той же системе, но и процессов в разных системах.

Java поддерживает IPC с помощью сокетов, библиотек RMI и CORBA.

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

Приложение Java может создавать дополнительные процессы с помощью объекта ProcessBuilder.

Потоки существуют в процессе — каждый процесс имеет хотя бы один поток.

Потоки используют общие ресурсы процесса, включая память и открытые файлы.

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

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

Каждый поток имеет свой собственный кеш памяти.

Если поток читает общие данные, он сохраняет эти данные в своем собственном кеше памяти.

Несколько потоков создаются в приложении для обеспечения параллельной или скорее независимой обработки или асинхронного поведения.

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

При этом ускорение программы с помощью многопоточных вычислений на нескольких процессорах ограничено размером последовательной части программы. Это так называемый закон Амдала.

Этот закон гласит следующее — В случае, когда задача разделяется на несколько частей, суммарное время её выполнения на параллельной системе не может быть меньше времени выполнения самого длинного фрагмента.

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

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

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

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

Проблема видимости и доступа может привести к сбою в работе — программа перестанет реагировать и войдет в ступор или взаимную блокировку из-за одновременного доступа к данным, или может быть сбой безопасности — программа создаст неверные данные.

Как решаются эти проблемы мы обсудим позже.

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

Но с точки зрения программиста, вы начинаете с одного потока, называемого основным потоком.

Этот поток имеет возможность создавать дополнительные потоки.

Вопрос в том, как мы можем создать, запустить и выполнить поток?

В Java каждый поток представлен экземпляром класса Thread.

Создать поток, или экземпляр Thread, можно двумя способами.

Первый способ, это сначала создать объект Runnable.

Интерфейс Runnable определяет один метод run, предназначенный для того, чтобы содержать код, выполняемый в потоке.

После создания, объект Runnable передается конструктору класса Thread.

И поток запускается методом start.

Второй способ, это создать подкласс класса Thread.

Сам класс Thread реализует интерфейс Runnable, и при этом его метод run пустой.

Поэтому нужно создать подкласс класса Thread и предоставить собственную реализацию метода run.

Таким образом, первая ключевая операция — это создание потоков.

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

Затем после создания потока, он фактически не начинает выполнение.

Поэтому, следующее, что вам нужно сделать, это вызвать метод start.

Теперь, ваша основная программа сама по себе является потоком.

И у нас есть основной поток, который создает и запускает другой поток.

В другом потоке выполняется свой код.

Теперь основной поток после запуска другого потока может выполнить свой код.

В этом случае у нас параллельно выполняются два куска кода на двух разных ядрах.

Класс Thread содержит метод join.

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

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

В этом случае, планировщик потоков планирует потоки, что не гарантирует порядок выполнения потоков.

В идеальном мире все потоки всех программ работают на отдельных процессорах.

Но в реальности, потоки должны разделяться между одним или несколькими процессорами.

Либо JVM, либо операционная система базовой платформы определяют, как распределять ресурс процессора среди потоков — задача, известная как планирование потоков.

Эта часть JVM или операционной системы, которая выполняет планирование потоков, является планировщиком потоков.

Java не заставляет виртуальную машину планировать потоки определенным образом, поэтому планирование потоков зависит от конкретной платформы.

Предположим, у нас есть два потока t1 и t2.

Несмотря на то, что мы запустили потоки последовательно, планировщик потоков не запускает и не завершает их в указанном порядке.

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

А если поток t1 должен использовать вычисления потока t2, что нам делать?

Решить эту проблему мы можем с помощью метода join ().

Этот код запустит второй поток t2, только после завершения первого потока t1, так как метод join приостанавливает выполнение главного потока до тех пор, пока не завершится поток t1.

Если поток прерывается, бросается исключение InterruptedException.

Теперь, предположим, что мы передали в метод run класса MyClass основной поток и применили к нему метод join.

Тогда первый поток будет ждать, когда завершится основной поток, а основной поток будет ждать, когда завершится первый поток.

Возникнет дедлок deadlock или взаимная блокировка потоков.

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

Вызов этого метода ставит выполнение текущего потока на паузу, при этом нужно указать количество миллисекунд паузы.

Здесь также нужно обрабатывать исключение InterruptedException.

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

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

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

Например, при закрытии приложения, которое может использовать несколько потоков, и они могут быть не завершены в момент закрытия приложения.

Как запросить задачу, выполняемую в отдельном потоке, закончиться раньше?

Как заставить задачу реагировать на такой запрос?

В этом примере создается задача, которая печатает числа от 0 до 9 в консоли.

После печати числа, задача должна подождать 1 секунду перед печатью следующего числа.

Задача выполняется в отдельном потоке, отличном от основного потока приложения.

После запуска задачи основной поток должен подождать 3 секунды и затем завершить работу.

При завершении работы приложение должно запросить завершение выполняемой задачи.

Перед тем, как полностью закрыть приложение, приложение должно максимально ждать 1 сек для завершения задачи.

Задача должна ответить на запрос завершения, немедленно останавливаясь.

Общее выполнение задачи занимает не менее 9 секунд.

Поэтому задача не сможет распечатать все десять чисел от 0 до 9.

Для запроса на прерывание потока, основной поток вызывает метод прерывания interrupt.

В Java один поток не может просто остановить другой поток.

Поток может только запросить остановку другого потока.

И запрос выполняется в виде вызова метода interrupt.

Вызов метода interrupt в экземпляре Thread устанавливает флаг прерывания как true.

Если этот поток заблокирован вызовом методов wait, join или sleep, то его статус прерывания будет очищен, и он выбросит исключение InterruptedException.

Таким образом, как только taskThread прерывается основным потоком, Thread.sleep (1000) отвечает на прерывание, выбрасывая исключение.

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

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

После очистки статуса прерывания, подтвердить этот статус можно самопрерыванием с помощью вызова Thread.currentThread(). interrupt ().

И без использования обработки InterruptedException, прервать цикл задачи можно, проверяя статус прерывания с помощью вызова Thread.isInterrupted ().

Что только один поток может одновременно выполнять блок кода

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

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

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

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

Это гарантирует, что изменения состояния объекта будут видны для всех потоков.

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

Освобождение блокировки происходит, даже если возврат метода был вызван неперехваченным исключением.

Другими словами, каждый объект в Java имеет ассоциированный с ним монитор.

Монитор представляет своего рода инструмент для управления доступа к объекту.

Когда выполнение кода доходит до оператора synchronized, монитор объекта захватывается владельцем, и на это время монопольный доступ к синхронизированному коду имеет только один поток, который является владельцем монитора.

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

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

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

Конструкторы не могут быть синхронизированы — использование ключевого слова synchronized для конструктора является синтаксической ошибкой.

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

Еще раз, если два нестатических метода класса объявлены как synchronized, то в каждый момент времени из разных потоков на одном объекте может быть вызван только один из них.

Поток, который вызывает метод первым, захватит монитор, и второму потоку придется ждать.

Это верно только для разных потоков.

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

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

Это верно только для вызовов методов одного экземпляра.

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

Другой способ создания синхронизированного кода — синхронизированные блоки.

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

Когда один поток заходит внутрь блока кода, помеченного словом synchronized, то Java-машина тут же блокирует монитор объекта, который указан в круглых скобках после слова synchronized.

Больше ни один поток не сможет зайти в этот блок, пока наш поток его не покинет.

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

Для нестатических методов, синхронизация метода эквивалентна синхронизации тела метода с объектом this.

Для статических методов, синхронизация метода эквивалентна синхронизации тела метода с объектом Class.

Предположим, что класс имеет два поля экземпляра: c1 и c2, которые никогда не используются вместе.

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

Вместо использования синхронизированных методов или использования блокировки this, мы создаем два объекта исключительно для обеспечения блокировок.

Таким образом синхронизация блоков может дать возможность из разных потоков на одном объекте вызывать разные синхронизированные блоки.

Среда выполнения JRE неявно обеспечивает синхронизацию при доступе к volatile-переменным, но с очень большой оговоркой: чтение volatile-переменной и запись в volatile-переменную синхронизированы, а неатомарные операции, такие как операция инкремента или декремента ― нет.

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

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

В целях повышения производительности среда выполнения JRE сохраняет локальные копии переменных для каждого потока, который на них ссылается.

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

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

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

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

Это означает, что изменения в volatile переменной всегда видны для других потоков.

Опять же речь идет только об операциях чтения и записи.

Резюмируя, объявление блока кода синхронным обеспечивает для кода атомарность и видимость.

Атомарность значит, что только один поток одновременно может выполнять код, защищенный данным объектом-монитором (блокировкой), позволяя предотвратить многочисленные потоки от столкновений друг с другом во время обновления общего состояния.

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

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

Подобное правило видимости существует и для переменных volatile.

Поток T1 может получить монитор объекта A одновременно с тем, что поток T2 получит монитор объекта B, а затем каждый поток будет ожидать монитора В и А соответственно неопределенный срок.

Одним из лучших способов предотвращения взаимоблокировки — это избегать одновременного получения более одного монитора.

Еще одно нарушение живучести, это LIVELOCK или динамическая взаимоблокировка.

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

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

Поток T1 в цикле увеличивает x, затем читает значение x и продолжает делать это, пока х меньше 2.

А поток T2 в цикле уменьшает значение x, затем читает значение x и продолжает делать это, пока х больше -2.

Возможна ситуация, при которой поток T1 получает x = 1, но прежде чем он получит шанс увеличить x и достичь x = 2, поток T2 уменьшает x, противодействуя тому, что делает T1.

И делает x = -1.

Но до того, как поток T2 получит шанс уменьшить х до -2, поток T1 может снова увеличить x до 1.

И так до бесконечности.

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

Теперь третий вид проблемы с живучестью, называется STARVATION или голодание.

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

В результате этот поток голодает.

Паттерн защищенный блок Guarded Block

Предположим у нас есть задача написать приложение Producer-Consumer.

Это приложение состоит из двух потоков — производителя, который создает данные, и потребителя, который что-то делает с этими данными.

Два потока обмениваются данными с использованием общего объекта.

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

Для создания такого приложения используется паттерн Защищенный блок.

Сначала значение empty установлено в true.

Поток потребителя вызывает синхронизированный метод take.

Объект класса блокируется.

Метод запускает цикл while.

В этом цикле вызывается метод wait.

При этом объект класса разблокируется.

Поток производитель вызывает метод put.

При этом объект класса блокируется.

Блок while метода put пропускается.

Записывается сообщение и значение empty устанавливается в false.

Вызывается метод notifyAll, который будит все потоки.

Монитор объекта перехватывается потоком потребителем.

Цикл while завершается, и значение empty устанавливается снова в true.

Сообщение потребляется.

Таким образом, паттерн защищенный блок обеспечивает координацию действий потоков.

При использовании этого паттерна не забудьте убедиться, что приложение имеет другие потоки, которые получают блокировки на объектах, на которых другие потоки ранее вызывали метод wait, и эти другие потоки вызывают методы notify или notifyAll, чтобы приложение могло избежать startvation для тех потоков, которые вызвали метод wait.

Метод notifyAll будит все ожидающие потоки, тогда как метод notify случайно выбирает один из ожидающих потоков и будит его.

Теперь вопрос.

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

Для этого нужно сделать следующее:

Не давать классу методы «setter» — методы, которые изменяют поля или объекты, на которые ссылаются поля.

Сделать все поля класса финальными и приватными.

Не допустить создание подклассов с помощью объявления класса финальным.

Не давать классу методы, изменяющие изменяемые объекты.

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

Этот класс добавляет дополнительную опцию справедливости.

Конструктор этого класса принимает необязательный параметр fairness.

Когда установлено true, при конкуренции нескольких потоков для получения блокировки, доступ предоставляется самому долго ожидающему потоку.

В противном случае блокировка не гарантирует какой-либо конкретный порядок доступа.

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

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

Как модифицировать этот код с использованием ReentrantLock.

Здесь мы создаем объект ReentrantLock, устанавливаем блокировку, выполняем операцию и производим разблокировку.

Метод newCondition класса ReentrantLock позволяет создавать несколько условий в одной блокировке для координации потоков.

В этом примере производителя-потребителя мы создаем два условия для одной блокировки.

И у нас есть два метода для записи и чтения данных.

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

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

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

Интерфейс ReadWriteLock — это усовершенствованный механизм блокировки потоков, который позволяет нескольким потокам одновременно читать определенный ресурс, но только одному потоку записывать одновременно.

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

Так как ошибки параллелизма возникают в первую очередь при одновременном считывании и записи в общий ресурс или при одновременном выполнении нескольких записей в общий ресурс.

А здесь запись ограничивается одним потоком.

Правила, по которым потоку разрешено блокировать ReadWriteLock для чтения или записи защищенного ресурса, заключаются в следующем:

Можно блокировать для чтения, если никакие потоки не заблокировали ReadWriteLock для записи, и ни один поток не запросил блокировку записи, но еще не получил ее.

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

Можно блокировать для записи, если ни один поток не читает или не записывает.

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

ReadWriteLock — это интерфейс.

Таким образом, для использования ReadWriteLock пакет java.util.concurrent. locks предоставляет реализацию ReadWriteLock — класс ReentrantReadWriteLock.

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

Надо заметить, что ReadWriteLock фактически внутренне хранит два экземпляра Lock. Один защищенный доступ для чтения и один защитный доступ для записи.

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

И для этого Java представляет класс ReentrantReadWriteLock в пакете java.util.concurrent. locks, который обеспечивает такой механизм блокировки.

В политике блокировки он позволяет блокировать чтение одновременно несколькими потоками-читателями, если нет потока-писателя, и блокировка записи является исключительной.

Но позже выяснилось, что у ReentrantReadWriteLock есть серьезные проблемы с голоданием, если они не обрабатываются должным образом.

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

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

В Java 8 добавлен новый тип блокировки StampedLock, который помимо предоставления отдельных блокировок для чтения и записи также имеет функцию оптимистической блокировки для операций чтения.

StampedLock также предоставляет метод обновления блокировки чтения до блокировки записи, которого нет в ReentrantReadWriteLock.

Методы блокировки StampedLock возвращают штамп, представленный значением long.

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

Если штамп всегда равен нулю, это означает, что не удалось получить доступ к ресурсу.

Имейте в виду, что в StampedLock каждый вызов для получения блокировки всегда возвращает новый штамп и производится блокировка, даже если тот же поток уже содержит блокировку, что может привести к deadlock.

И еще одно замечание — ReadWriteLock имеет два режима управления доступом для чтения и записи, в то время как StampedLock имеет три режима доступа.

При чтении, метод readLock получает неэксклюзивную блокировку. Этот метод возвращает штамп, который можно использовать для разблокировки или преобразования режима.

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

Этот метод не будет блокировать и возвращает штамп как ноль, если блокировка не будет немедленно доступна.

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

При оптимистическом чтении, метод tryOptimisticRead возвращает ненулевой штамп только в том случае, если в настоящее время блокировка не удерживается в режиме записи.

При этом используется метод validate для проверки правильности значений, считанных при оптимистичном чтении.

Метод validate возвращает true, если блокировка не была запрошена для записи с момента получения данного штампа.

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

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

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

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

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

Метод tryConvertToOptimisticRead освобождает блокировку и возвращает штамп для наблюдения.

Также нужно отметить, что планировщик потоков в случае StampedLock не всегда предпочитает читателей над писателями или наоборот.

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

И мы узнали об использовании блокировок различными способами.

В многопоточном программировании есть такое понятие, как критическая секция.

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

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

И мой банковский счет, скажем, содержит 500 долларов.

И у меня есть общий банковский счет с дочерью.

Моя дочь имеет собственный банковский счет, баланс которого составляет 0 долларов.

Дочь просит у меня 100 долларов.

Я перевожу деньги со своего счета на общий счет, а дочка переводит деньги с общего счета на свой счет.

Теперь мы можем смоделировать эту операцию, используя два потока.

Есть поток T1, в котором 100 долларов вычитаются из моего баланса и добавляются к общему балансу.

И есть поток T2, где 100 долларов вычитаются из общего баланса и добавляются к балансу дочери.

Теперь вопрос в том, что может пойти не так, если это будет выполнено как многопоточная программа?

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

При этом порядок чтения и записи этой общей переменной может вызвать проблемы.

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

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

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

Предположим, что у нас есть массив элементов.

И доступ к этим элементам осуществляется несколькими потоками.

И эти потоки извлекают элементы из массива и обрабатывают их.

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

На каждом этапе цикла эта переменная увеличивается на единицу.

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

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

Для устранения этой проблемы нужно изолировать эту переменную.

А обработка элементов может продолжаться параллельно.

Таким образом, нужно сделать переменную-счетчик атомарной.

Вместо того, чтобы реализовать это с помощью низкоуровневого программирования, и блокировать весь код, где используется эта переменная, мы можем воспользоваться пакетом java.util.concurrent.atomic, который определяет классы, поддерживающие атомарные операции для одиночных переменных.

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

При этом вместо синхронизации мы можем просто использовать объекты соответствующих классов.

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

Пакет java.util.concurrent.atomic также предоставляет класс AtomicReference, который позволяет обновлять ссылку на объект атомарно.

Здесь мы на основе строки создаём объект AtomicReference, который позволяет обновить ссылку на строку атомарно.

А также этот пакет содержит классы обновления, такие как AtomicReferenceFieldUpdater, которые можно использовать для операций compareAndSet для поля volatile класса.

Метод compareAndSet означает «Обновить переменную этим новым значением, но отказать, если другой поток изменил значение после моего последнего просмотра».

В этом примере сначала создается экземпляр AtomicInteger с начальным значением 123.

Затем сравнивается значение AtomicInteger с ожидаемым значением 123, и, если они равны, новое значение AtomicInteger становится 234.

Среди других атомарных классов также есть такие классы как AtomicBoolean и AtomicLong.

Для переменных типа Float и Double атомарность можно обеспечить с помощью классов AtomicInteger и AtomicLong и методов конвертации floatToIntBits, intBitstoFloat, doubleToLongBits, и longBitsToDouble.

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

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

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

И масштабируемая поддержка обновляемых переменных представлена с помощью таких классов, как DoubleAccumulator, DoubleAdder, LongAccumulator, LongAdder, в которых используются техники сокращения конкуренции, которые обеспечивают значительное увеличение пропускной способности по сравнению с атомарными переменными.

Классы LongAdder и DoubleAdder могут использоваться в качестве альтернативы AtomicLong для последовательного сложения чисел.

С точки зрения использования, применение этих классов очень похоже на использование атомарных классов.

Просто создается LongAdder и используются его методы, такие как intValue и add, чтобы получить и установить значение.

Магия происходит за кулисами.

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

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

Это уменьшает необходимость возврата и выполнения операции сравнения и замены.

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

Этот класс используется в ситуациях, когда добавлять числа приходится гораздо чаще, чем запрашивать результат.

И несложно догадаться, что, давая прирост в производительности, LongAdder требует гораздо большего количества памяти из-за того, что он хранит все слагаемые.

Классы LongAccumulator и DoubleAccumulator можно использовать для накопления результатов в соответствии с предоставленным LongBinaryOperator и DoubleBinaryOperator.

То есть вместо простого сложения класс обрабатывает входящие значения с помощью лямбды типа LongBinaryOperator или DoubleBinaryOperator, которая передаётся при инициализации.

Так же, как и Adder, Accumulator хранит весь набор переданных значений в памяти, уменьшая взаимодействие между потоками.

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

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

При передаче числа в качестве аргумента методу accumulate, этот метод вызовет функцию sum.

Вы можете установить любой объект в ThreadLocal, и этот объект будет своим для каждого потока.

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

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

При этом значения, хранящиеся в Thread Local, являются уникальными для потока, то есть каждый поток будет иметь собственную переменную Thread.

Один поток не может получить доступ и изменить переменную ThreadLocal другого потока.

ExecutorService — расширяет интерфейс Executor, добавляя функции для управления жизненным циклом задач.

ExecutorService также предоставляет метод submit, который может принимать объект Runnable, а также объекты Callable, которые позволяют задаче вернуть значение.

ScheduledExecutorService — расширяет интерфейс ExecutorService, добавляя функциональность для планирования выполнения задач.

Помимо вышеуказанных трех интерфейсов, Concurrency API также предоставляет класс Executors, который содержит фабричные методы для создания различных видов ExecutorService.

Под капотом, при использовании Executor интерфейсов, задачи Runnable сначала помещаются в очередь, а затем их выполнение распределяется по рабочим потокам пула потоков Thread Pool.

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

Существует несколько типов пулов потоков.

Это фиксированный пул потоков, который содержит определенное количество потоков.

И если поток каким-то образом завершается во время своего использования, он автоматически заменяется новым потоком.

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

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

И есть пул с одним потоком и неограниченной очередью.

Если этот единственный поток завершается из-за сбоя во время выполнения, на его место создается новый поток.

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

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

При этом если пул потоков не достиг еще своего основного размера, он создает новые потоки.

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

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

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

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

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

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

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

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

Резюмируя, Threadpool состоит из потоков, которые ищут задания для выполнения.

Вместо запуска нового потока с объектом Runnable, рабочий поток пула потоков просто вызывает функцию run объекта Runnable.

Таким образом, поток в ThreadPool не создается с помощью Runnable, который вы предоставляете, но существующий поток пула потоков просто проверяет, готовы ли какие-либо задачи к выполнению и вызывает их напрямую.

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

Рабочие потоки опрашивают очередь, чтобы увидеть, есть ли задача для выполнения и запускают ее.

Хотя пул потоков является мощным механизмом структурирования многопоточных приложений, он не лишен риска.

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

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

Далее, размер пула потоков должен быть правильно настроен.

Потоки потребляют множество ресурсов, включая память для объекта Thread, стеки выполнения.

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

Наконец, есть накладные расходы переключения между потоками.

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

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

Это может произойти, если задача выбросит исключение.

Если объект пула потока не перехватит исключение и восполнит поток, размер пула потоков будет уменьшен на единицу.

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

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

Если поток постоянно потребляется такой задачей, он эффективно удаляется из пула.

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

Для эффективного использования пула потоков, не ставьте в очередь задачи, которые ждут синхронно результатов других задач.

Это может вызвать deadlock, когда все потоки заняты задачами, которые в свою очередь ожидают результатов от задач, поставленных в очередь, которые не могут выполняться, потому что все потоки заняты.

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

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

Для чисто вычислительных задач имеет смысл определить размер пула равным количеству процессоров системы.

Для сетевых задач размер пула можно оценить, как N* (1+WT/ST) (N-количество процессоров, WT-время ожидания ресурса, ST-время обслуживания запроса).

В этом примере создается пул из одного потока и неограниченной очереди.

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

Здесь пул потока создается отдельно при запуске приложения.

И отдельно при остановке приложения вызывается метод shutdown, который перестает принимать новые задачи, ждет выполнения ранее поставленных задач, а затем завершает работу executor.

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

Хороший способ закрыть ExecutorService, это использовать оба этих методов в сочетании с методом awaitTermination.

При таком подходе ExecutorService сначала прекратит принимать новые задачи, подождет определенный период времени для завершения всех задач.

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

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

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

Если в Runnable мы определяем метод run, то в Callable, мы определяем метод call.

Метод submit помещает задачу в очередь для выполнения потоком.

Однако он не знает, когда будет получен результат этой задачи.

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

Метод ExecutorService.submit немедленно возвращает объект Future.

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

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

Future также предоставляет метод isDone для проверки того, завершена задача или нет.

Если прерывается поток, выполняющий вычисление, метод генерирует исключение InterruptedException.

Когда вычисление задачи завершается, метод get немедленно возвращает управление.

Вы можете отменить Future, используя метод Future.cancel, который пытается отменить выполнение задачи и возвращает true, если она отменена успешно, иначе метод возвращает false.

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

Вы можете использовать метод isCancelled, чтобы проверить задача отменена или нет.

Кроме того, после отмены задачи, метод isDone всегда будет возвращать true.

Метод future.get блокирует и ждет завершения задачи.

Если, вы вызываете удаленную службу в задаче, а удаленная служба отключена, то future.get заблокирует приложение навсегда.

Чтобы это предотвратить, вы можете добавить в метод get время ожидания.

Также, как и с объектом Runnable, с помощью объекта Callable, вы можете одновременно создать сразу несколько задач и отправить их на выполнение.

Обратите внимание, что, во-первых, здесь мы создаем фиксированный пул из двух потоков.

Во-вторых, мы используем метод invokeAll, чтобы выполнить несколько задач, передавая коллекцию объектов Callable.

Метод invokeAll возвращает список объектов Future.

Метод invokeAll ожидает, пока все результаты не будут вычислены до их возврата. То есть метод invokeAll возвращает список объектов Future, для которых isDone true.

В отличие от метода submit, который возвращает сразу, метод invokeAll ожидает завершения задач.

Так как метод invokeAll возвращает коллекцию, ее можно обработать с помощью Stream API.

В отличие от метода invokeAll, метод invokeAny передает на выполнение список задач и ожидает выполнение любой из них, то есть наиболее быстрой.

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

Здесь метод schedule принимает объект Runnable, задержку и единицу задержки, и определяет выполнение задачи через 5 секунд с момента отправки.

Метод scheduleAtFixedRate принимает объект Runnable, начальную задержку, период выполнения и единицу времени, и запускает выполнение заданной задачи после указанной задержки и затем периодически выполняет ее с указанным интервалом.

Имейте в виду, что scheduleAtFixedRate не учитывает фактическую продолжительность задачи. Поэтому, если вы укажете период в одну секунду, но для выполнения задачи потребуется 2 секунды, тогда пул потоков начнет быстро перегружаться.

Поэтому для периодического запуска задачи лучше использовать метод scheduleWithFixedDelay.

Разница заключается в том, что период ожидания применяется от конца задачи до начала следующей задачи.

В этом примере задача выполняется периодически с фиксированной задержкой в одну секунду между окончанием выполнения и началом следующего выполнения.

Чтобы удалить задачу без остановки ScheduledExecutorService, можно использовать результат ScheduledFuture, возвращаемый методами интерфейса ScheduledExecutorService, и применить к этому результату метод cancel.

Для возврата результата, в метод schedule можно передать не объект Runnable, а объект Callable, и применить к результату ScheduledFuture метод get, так как ScheduledFuture расширяет уже знакомый нам интерфейс Future.

Как я уже говорил, классы ThreadPoolExecutor и ScheduledThreadPoolExecutor позволяют установить свои собственные настройки для объекта Executor и определить основной размер пула потоков, максимальный размер пула потоков, указать тип используемой очереди и другое.

Создание ExecutorService с помощью newFixedThreadPool эквивалентно использованию ThreadPoolExecutor с очередью LinkedBlockingQueue.

Эта очередь в данном случае неограниченна и упорядочивает элементы FIFO (первый пришел-первый ушел).

Новые элементы вставляются в хвост очереди, а элементы в начале очереди обрабатываются.

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

При этом если пул потоков не достиг еще своего основного размера, он создает новые потоки.

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

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

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

Здесь мы с помощью ThreadPoolExecutor создаем пул из двух потоков с очередью SynchronousQueue.

Эта очередь с нулевой емкостью.

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

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

Такой режим может быть полезен для незамедлительной обработки задач в фоне.

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

ForkJoinTask — это базовый тип задач, выполняемых внутри ForkJoinPool.

На практике должен быть расширен один из двух его подклассов: RecursiveAction для задач void и RecursiveTask для задач, возвращающих значение.

Оба эти класса имеют абстрактный метод compute, в котором должна быть определена логика задачи.

Общий шаблон метода compute следующий — если размер задачи меньше порогового значения, задача решается без параллелизма.

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

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

После вызова метода compute задачи, она опять разделится и все повторится до тех пор, пока размер задачи не станет маленьким.

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

После возврата результатов, они объединяются.

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

Здесь, если размер массива меньше 5000 элементов, суммирование производится без параллелизма в цикле.

Если размер больше порогового значения, задача разделяется пополам.

К одной части рекурсивно применяется метод fork, а к другой рекурсивно применяется метод compute.

Затем к той части, к которой применялся метод fork, применяется метод join, ожидающий результата.

И в конце концов результаты складываются вместе.

Сам по себе вызов метода fork не создает никакого потока и не вызывает выполнение задачи.

Для этого нужно создать пул потоков ForkJoinPool и послать в его метод задачу ForkJoinTask.

Что мы и делаем в методе sumArray.

Создать пул потоков ForkJoinPool можно методом commonPool, или можно конструктором, указав уровень параллелизма или количество ядер процессора для выполнения вычислений.

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

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

Метод commonPool делает это автоматически.

Класс ForkJoinPool имеет четыре метода, принимающих задачу ForkJoinTask.

Асинхронные методы submit или execute помещают задачу в пул потоков.

Для получения результата нужно затем вызвать метод join.

Метод invoke помещает задачу в пул и ожидает результата.

Метод invokeAll позволяет отправить коллекцию задач в пул потоков.

При применении фреймворка fork/join возникает вопрос — как определить пороговое значение метода compute.

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

load factor обычно берут 8 или 16.

Хорошая практика, когда порог предполагает количество базовых шагов вычисления больше 100 и меньше 10000.

Теперь рассмотрим код, где мы рекурсивно решаем подзадачи.

Может показаться более естественным вызывать fork дважды для двух подзадач, а затем дважды вызвать join.

Или вызвать ожидающий результат метод invokeAll, а затем дважды вызвать join.

Или вызвать метод invoke, который объединяет fork и join в одном вызове.

Однако это будет менее эффективно, чем просто вызывать compute, так как вы увеличиваете накладные расходы на создание дополнительных параллельных задач, чем это нужно.

Кроме того, здесь важен порядок вызова методов.

Сначала вызвать fork, который сразу вернет.

Затем вызвать compute и получить результат.

Затем вызвать join и получить результат.

Однако при создании задачи RecursiveAction имеет смысл использовать метод invokeAll, так как нам здесь не нужен результат, а нужно просто выполнить задачи параллельно.

Самостоятельное задание

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

Создайте задачу RecursiveAction и сравните эффективность метода invokeAll и fork + compute + join.

Надо сказать, что Java 8 добавляет в интерфейс ExecutorService метод newWorkStealingPool, который фактически возвращает пул потоков ForkJoinPool.

То есть этот метод не создает никакого нового пула потоков, а отправляет нас к фреймворку fork/join.

Помимо классов RecursiveAction и RecursiveTask, Java 8 вводит класс CountedCompleter, который реализует класс ForkJoinTask.

Класс CountedCompleter обеспечивает механизм для выполнения метода после завершения всех подзадач.

Это метод onCompletion класса CountedCompleter.

Для вызова метода onCompletion, в той части метода compute, где идет вычисление подзадачи, достигшей минимального размера, мы вызываем метод tryComplete, который вызывает метод onCompletion.

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

Если необходимо вернуть результат, переопределяется метод getRawResult класса CountedCompleter.

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

Благодаря этому изменяется внутренний счетчик ожидающих задач.

Этот счетчик определяет, были ли выполнены задачи.

Если это не так, тогда счетчик уменьшается.

Когда задачи завершаются и счетчик обнуляется и вызывается метод tryComplete, событие завершения отправляется во все задачи CountedCompleter.

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

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

CompletableFuture является реализацией интерфейса Future, который использовался нами как ссылка на результат асинхронного вычисления.

Он предоставлял метод isDone, чтобы проверить, выполнено ли вычисление или нет, и метод get для получения результата вычисления при его выполнении.

Однако интерфейс Future имеет некоторые недостатки.

При его использовании результат вычисления невозможно завершить вручную.

Например, вы написали функцию для получения данных из удаленного сервиса.

Так как этот вызов занимает много времени, вы запускаете его в отдельном потоке и возвращаете Future из своей функции.

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

Сделать это с помощью Future нельзя.

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

Future предоставляет метод get, который блокирует до тех пор, пока результат не будет доступен.

И у вас нет возможности подключить функцию обратного вызова к Future и получить ее вызов автоматически, когда будет получен результат в будущем.

Далее, нельзя соединить вместе несколько объектов Future.

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

Вы не можете создать такой асинхронный рабочий поток с помощью Future.

Далее, вы не можете комбинировать несколько объектов Future вместе.

Например, у вас есть несколько различных Future, которые вы хотите запустить параллельно, а затем выполнить некоторую функцию после того, как все они завершатся.

Вы тоже не можете это сделать с помощью интерфейса Future.

Кроме того, интерфейс Future не имеет конструкции обработки исключений.

Все эти проблемы решает класс CompletedFuture, который реализует интерфейсы Future и CompletionStage и предоставляет набор методов для создания, соединения и объединения нескольких Future. Он также поддерживает обработку исключений.

Для выполнения задачи в отдельном потоке, мы применяем статический метод supplyAsync, который принимает реализацию интерфейса Supplier и выполняет эту реализацию в потоке пула потоков ForkJoinPool.

Интерфейс Supplier имеет единственный метод get, который необходимо определить.

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

Если вы хотите выполнить задачу без возврата результата, вы вызываете метод runAsync и передаете ему объект Runnable.

Также вы можете выполнить задачу в каком-либо другом пуле потоков, явно указав Executor.

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

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

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

Можно присоединить обратный вызов к CompletableFuture, используя методы thenApply, thenAccept и thenRun.

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

Этот метод принимает реализацию интерфейса Function как аргумент.

Интерфейс Function содержит метод, который принимает аргумент и возвращает результат.

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

Метод thenAccept принимает реализацию интерфейса Consumer, который имеет метод accept, принимающий аргумент и ничего не возвращающий.

Этот метод имеет доступ к результату CompletableFuture.

Метод thenRun принимает объект Runnable и не имеет доступа к результату Future.

У всех этих метод есть вариация Async, которая запускает код в отдельном потоке.

Вы также можете создать цепочку вложенных CompletableFuture.

Здесь вы сначала получаете идентификатор клиента, а затем передаете его методу thenCompose и возвращаете баланс клиента.

Теперь, если вы хотите вычислить две задачи независимо друг от друга, а затем что-то вычислить с использованием их результатов, вместо метода thenCompose нужно применить метод thenCombine.

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

Мы использовали методы thenCompose и thenCombine чтобы соединить и объединить два CompletableFutures вместе.

Теперь, если вы хотите совместить произвольное число CompletableFutures, для этого можно использовать методы allOf и anyOf.

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

ManagedBlocker

Фреймворк Fork/Join дает возможность использовать один общий пул потоков ForkJoinPool, который предварительно сконфигурирован JVM и используется для распараллеливания потоков и выполнения задач, заданных, например, с помощью CompletableFuture.supplyAsync и так далее.

Это звучит хорошо.

Однако то, что пул потоков ForkJoinPool является общим, означает совместное использование всеми компонентами, работающими в одном JVM-процессе.

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

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

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

Интерфейс ManagedBlocker предоставляет два метода.

Метод isReleasable должен возвращать true, если блокировка не требуется.

Метод block блокирует текущий поток, если это необходимо.

Эти действия выполняются любым потоком, вызывающим метод ForkJoinPool.managedBlock.

Этот метод запускает блокирующую задачу.

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

Этот метод также неоднократно вызывает методы интерфейса ManagedBlocker isReleasable и block, пока оба метода не вернут true.

Каждому вызову метода block предшествует вызов метода isReleasable, который возвращает false.

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

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

В этом примере мы блокируем рабочий поток вызовом метода Thread.sleep.

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

Для этого мы создадим класс Sleeper, расширяющий интерфейс ManagedBlocker.

Здесь при вызове метода managedBlock, будет вызван метод isReleasable, он вернет false.

Поэтому будет вызван метод block и текущий поток заснет.

При этом в пуле будет создан новый поток.

Как только текущий поток проснется, метод isReleasable начнет возвращать true, это просигнализирует пулу потоков, что один поток можно убрать.

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

Потокобезопасные коллекции

Фреймворк Collection, с его интерфейсами List, Set, и Map, обеспечивает потоковую безопасность с помощью синхронизированных оболочек.

Где используются синхронизированные блоки кода.

Однако эти потокобезопасные коллекции все еще имеют некоторые недостатки.

Например, необходимо блокировать коллекцию при ее итерации, иначе вы рискуете получить исключение ConcurrentModificationException.

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

Если итератор такой коллекции обнаруживает, что во время итерации была внесена модификация, он выдает исключение ConcurrentModificationException.

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

Поэтому появились новые классы коллекций в пакете java.util.concurrent.

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

Вместо этого предпочтительно разрешить одновременную модификацию.

Итераторы коллекций java.util.concurrent называются слабо согласованными итераторами.

Для этих классов, если элемент был удален с начала итерации и еще не возвращен методом next, он не будет возвращен вызывающему.

Если элемент был добавлен с начала итерации, он может быть возвращен или не возвращен вызывающему.

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

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

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

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

Напротив, класс пакета java.util.concurrent ConcurrentHashMap позволяет множественным операциям чтения-записи выполняться параллельно.

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

Итераторы, возвращенные ConcurrentHashMap, также слабо согласованы, что означает, что они не будут бросать ConcurrentModificationException.

Что касается очередей, очереди могут быть ограниченными или неограниченными.

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

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

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

Блокирование неограниченной очереди позволит контролировать поток и ограничить производителей.

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

Поэтому в параллельных вычислениях используется очередь BlockingQueue и ее производные классы LinkedBlockingQueue, PriorityBlockingQueue, ArrayBlockingQueue, SynchronousQueue и другие.

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

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

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

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

Когда счётчик достигает 0, все ожидающие потоки разблокируются и продолжают выполняться.

В этом примере у нас есть счетчик со значением три.

И есть два потока.

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

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

Как только счетчик обнуляется, другой поток просыпается.

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

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

В этом примере мы создаем два барьера и два потока.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

В параллельных стрим потоках лучше всего обрабатывать структуры несвязанных данных, например, ArrayList. LinkedList лучше не использовать, так как в последовательном списке все элементы связаны с предыдущими/последующими элементами. И такие данные трудно распараллелить.

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

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

Здесь мы обертываем вычисление потока в отдельную задачу и передаем ее в отдельный пул потоков.

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

Как правило, элементы передаются в поток в том же порядке, в котором они определены в источнике данных.

Сохранение порядка в параллельных потоках увеличивает издержки при выполнении.

Но если нам порядок не важен, то мы можем отключить его сохранение и тем самым увеличить производительность, использовав метод unordered.

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

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

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

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

Промежуточные операции подразделяются на stateless и stateful операции.

Stateful операции, distinct, sorted, limit, skip, должны учитывать состояние из ранее обработанных элементов при обработке новых элементов.

Для вычисления результата с учетом состояния требуется обработка всего ввода.

Например, нельзя отсортировать поток до тех пор, пока не будут видны все элементы потока.

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

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

Поэтому stateful операции замедляют выполнение параллельных стримов.

Также сами параметры или лямбда выражения стрим операций могут быть stateless и stateful.

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

Здесь мы вычисляем сумму целых чисел, отбрасывая дублирующие элементы, так как метод add класса HashSet добавляет только те элементы, которых еще нет в наборе.

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

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

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

В каждой группе операция add отбросит дубликаты и все группы просуммируются.

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

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

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

Здесь мы в параллельных потоках пытаемся модифицировать ArrayList.

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

В это время другой поток успевает туда скопировать свои данные.

Тогда при копировании данных первым потоком возникает исключение ArrayIndexOutOfBoundsException.

Если же мы здесь синхронизируем ArrayList, мы потеряем все преимущество параллелизма.

Также, лямбда-выражения в стрим операциях не должны интерферировать.

Интерференция возникает, когда источник стрима изменяется, при обработке стрима.

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

Стримы позволяют выполнять параллельные операции над различными источниками данных, включая даже потоко небезопасные коллекции, такие как ArrayList.

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

Это верно также и для последовательных стримов.

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

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

Далее вызывается промежуточная операция, которая пытается этот источник модифицировать, и возникает исключение ConcurrentModificationException.

Таким образом, при использовании параллельных стримов, требуется тестирование и анализ алгоритма выполнения.

Таким образом, Java 9 предоставляет механизм публикации-подписки, когда подписчик информирует издателя, что он готов принять заданное количество элементов, и, если эти элементы доступны, издатель выталкивает это количество элементов получателю.

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

Это двухстороннее соединение между издателем и подписчиком называется подпиской.

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

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

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

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

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

Если у издателя возникает ошибка, он сигнализирует об ошибке подписчику.

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

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

Если объект является как издателем, так и подписчиком, он называется процессором.

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

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

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

Это интерфейс Publisher — производитель элементов данных, получаемых подписчиками.

Интерфейс Subscriber — приемник элементов данных.

Интерфейс Subscription — связь между издателем и подписчиком.

И интерфейс Processor — комбинация Publisher и Subscriber для преобразования данных.

Издатель публикует поток данных для зарегистрированных подписчиков и реализует интерфейс Flow. Publisher.

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

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

Однако, если подписчик уже зарегистрирован или регистрация терпит неудачу, этот метод вызывает метод onError подписчика с объектом IllegalStateException.

При удачной подписке вызывается метод onSubscribe подписчика с новым объектом Flow.Subscription.

Подписчик подписывается на издателя для получения элементов данных и реализует интерфейс Flow.Subscriber.

Этот интерфейс объявляет метод onSubscribe и три дополнительных метода.

Метод onSubscribe вызывается для подтверждения регистрации.

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

Метод onComplete вызывается, когда издатель закрывает подписку.

Метод onError вызывается при возникновении ошибки в процессе подписки.

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

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

Подписка реализует интерфейс Flow.Subscription, который объявляет два метода.

Метод request добавляет n элементов данных к текущему невыполненному требованию для этой подписки.

Если n меньше или равно 0, метод onError подписчика вызывается с аргументом IllegalArgumentException.

В противном случае подписчик получает до n дополнительных вызовов метода onNext.

Передача в качестве аргумента значения Long.MAX_VALUE указывает неограниченное количество вызовов.

Метод cancel отменяет подписку. Но дополнительные элементы данных могут быть получены после вызова метода cancel.

Наконец, процессор представляет собой функцию преобразования данных, которая работает в потоке.

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

Класс SubmissionPublisher реализует интерфейс Flow. Publisher, асинхронно выдавая ненулевые элементы данных текущим подписчикам до тех пор, пока подписка не будет закрыта.

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

SubmissionPublisher предоставляет три конструктора для инициализации издателя.

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

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

Затем издатель в строковом потоке публикует элементы потока.

После чего издатель закрывает подписку.

Подписчик MySubscriber реализует интерфейс Subscriber.

В методе подписки onSubscribe он запрашивает элемент данных у издателя.

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