Многопоточность
Два приложения работающие под управлением одной операционной системы — это два независимых процесса. Процесс состоит из потоков. Потоки могут выполняться параллельно друг с другом.
Многопоточность - это одновременное выполнение двух или более потоков для максимального использования центрального процессора
Применение многопоточности
Эффективное использование одного центрального процессора
Оптимальное использование нескольких центральных процессоров или их ядер
Улучшенный user experience в плане скорости ответа на запрос
Улучшенный user experience в плане справедливости распределения ресурсов
Статусы потоков
New – экземпляр потока создан, но он еще не работает.
Running — поток запущен и процессор начинает его выполнение. Suspended — запущенный поток приостанавливает свою работу, затем можно возобновить его выполнение.
Blocked — поток ожидает высвобождения ресурсов или завершение операции ввода-вывода.
Terminated — поток немедленно завершает свое выполнение.
Dead — после того, как поток завершил свое выполнение, его состояние меняется на dead, то есть он завершает свой жизненный цикл.
Для создания потоков вы можете имплементировать интерфейс Runnable или использовать подкласс Thread
Создание и запуск потоков
Существует два основных способа:
Создать класс и имплементировать интерфейс Runnable. Нам необходимо реализовать единственный метод run при имплиминтации интерфейса Runnable, который должен содержать код, выполняющийся в потоке. Объект Runnable передается конструктору класса Thread. Этот способ считается наиболее гибким и применяется для высокоуровневых API управления потоками.
public static void main(String[] args) {
System.out.println(Thread.currentThread().getName());
Runnable task = new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
System.out.println("Make some work!");
}
};
Thread thread = new Thread(task);
thread.start();
}
- Использовать подкласс Thread. Класс Thread сам имплементирует интерфейс Runnable: обратите внимание его метод run не выполняет никакой работы (необходимо переопределить поведение метода в наследнике). Можно объявить подкласс Thread, предоставляя собственную реализацию метода run. Данный способ больше подходит для “простых” приложений.
public static void main(String[] args) {
System.out.println(Thread.currentThread().getName());
ExampleThread exampleThread = new ExampleThread();
exampleThread.start();
}
public static class ExampleThread extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
System.out.println("Make some work!");
}
}
Обратите внимание, что новый поток создается только если вызвать метод start, который в свою очередь вызывает метод run - при попытке вызвать метод run напрямую, код выполняется в том же потоке. Так же при запуске JVM, создается главный поток с именем main и еще несколько служебных потоков.
Остановка потока
Теперь мы умеем создавать и запускать потоки, но как мы можем выбранный поток остановить? Вы можете обратить внимание, что класс Java Thread содержит метод stop() (помечен как deprecated, т.е. не рекомендованный к использованию). Метод stop() не дает гарантий относительно состояния, в котором поток остановили. таким образом, все объекты Java, к которым у потока был доступ во время его выполнения, останутся в неизвестном состоянии (изменения состояния объектов не будут зафиксированы и видны другим потокам), что может привести к ошибкам — например при выполнении потока создается ресурс (например подключение к базе данных), который не будет закрыт и как следствие мы получим утечку памяти. Вместо deprecated метода stop() вам следует использовать interrupt(). Если метод stop(), принудительно останавливал поток, то interrupt() только предлагает потоку остановить свое выполнение путем установки флага interrupted в true внутри потока — решение об остановке принимает сам код потока. Данный флаг отображает статус прерывания (значение по умолчанию false). Если поток прерывается другим потоком то происходит следующее:
- Если поток ожидает выполнения прерываемого метода блокирования (Thread.sleep(), Thread.join() или Object.wait()), то процесс ожидание прерывается и выбрасывает InterruptedException — после этого флаг interrupted устанавливается в значение false.
public static void main(String[] args) {Runnable task = () -> {try {Thread.sleep(10000);} catch (InterruptedException e) {System.out.println("Interrupted");}};Thread thread = new Thread(task);thread.start();thread.interrupt();}
- Если поток не ожидает выполнения прерываемого метода блокирования, флаг interrupted устанавливается в значение true — теперь код потока должен обработать переменную в реализации метода run
public static void main(String[] args) {Runnable task = () -> {while (!Thread.currentThread().isInterrupted()) {}System.out.println("Interrupted");};Thread thread = new Thread(task);thread.start(); thread.interrupt();}Также любой поток может остановиться сам — для этого необходимо вызвать static метод Thread.sleep() (самый простой способ взаимодействия с другими потоками). В операционной системе, с установленной JVM, имеется свой планировщик потоков, называемый Thread Scheduler. Данный планировщик принимает решение, какой поток и когда необходимо запускать. Метод Thread.sleep() может принимать в качестве параметра количество миллисекунд — время на которое поток попытается заснуть, возобновлением выполнения — обратите внимание, что абсолютная точность не гарантирована.public static void main(String[] args) {
Runnable task = () -> {
try {
Thread.currentThread().sleep(10000);
System.out.println("Waked up");
} catch (InterruptedException e) {
e.printStackTrace();
}
};
Thread thread = new Thread(task);
thread.start();
System.out.println("you here");
}Очередность выполнения потоков Теперь давайте представим, что в рамках решения задачи нам необходимо гарантировать, что поток не начнет свое выполнения пока не будет завершен другой поток. Для этого используется метод join() экземпляра класса Thread — он объединяет начало выполнения одного потока с завершением выполнения другого. Если метод join() вызывается на одном из потоков, то текущий Thread выполняющийся в этот момент блокируется до момента времени, пока поток, для которого вызван метод join не закончит свое выполнение. Метод join() может (не обязательно) принимать в качестве параметра количество миллисекунд — количество времени ожидания. Если в качестве значения времени ожидания указать 0, то такой поток будет «ждать вечно».
public static void main(String[] args) throws InterruptedException {Runnable task = () -> {try {Thread.sleep(10000);System.out.println("Waked up");} catch (InterruptedException e) {System.out.println("Interrupted");}};Thread thread = new Thread(task);thread.start();thread.join();System.out.println("Finished");}Также обратите внимание на статический метод Thread.yield(), который заставляет CPU переключиться на обработку других потоков системы. Метод может быть полезным, если, когда поток, для которого вызван метод yield() ожидает наступления события, а проверка наступления происходила как можно чаще.
Пул потоков
Процесс создания новых потоков и освобождение ресурсов являются дорогостоящей операцией. Мы можем изначально определить необходимое количество потоков, создать их и использовать для решения задач — в java для этого используются пулы потоков и очереди задач, из которых выбираются задачи для потоков. Пул потоков — это по сути контейнер, в котором находятся потоки, и после выполнения одной из задач они самостоятельно переходить к следующей. Вы можете использовать такой контейнер для контроля создания и управления потоками — это экономит ресурсы связанные с процессом создания новых потоков. Рассмотрим классы и интерфейсы, которые отвечают создание и управление пулом потоков (Executor Framework in Java):
Интерфейс Executor. Объекты, которые реализуют интерфейс Executor, могут выполнять runnable-задачу (Интерфейс имеет один метод void execute(Runnable command)).
ExecutorService. Интерфейс ExecutorService наследуется от интерфейса Executor и предоставляет возможности для выполнения заданий Callable.
Класс Executors. Утилитарный класс Executors создает классы, которые реализуют интерфейсы Executor и ExecutorService.
Теперь давайте посмотрим на основные реализации интерфейсов Executor и ExecutorServcie:
● ThreadPoolExecutor — пул потоков содержит фиксированное количество потоков - количество потоков определяется через конструктор.
● Executors.newCachedThreadPool() - возвращает пул потоков, если количество потоков в пуле не достаточно, то в нем будет создан новый поток.
● Executors.newSingleThreadExecutor() — пул потоков, который гарантирует, что в нем может быть только один поток.
● ScheduledThreadPoolExecutor — пул потоков используется для запуска периодических задач или задач, которые должны запуститься только раз по истечении некоторого промежутка времени.
Синхронизация
Итак, мы научились создавать и запускать потоки. Теперь давайте обсудим, что происходит когда два и более потока начинают конкурировать за ресурсы (т.е. пытаются в параллели получить доступ к одному и тому же объекту или ресурсу). При работе потоки нередко обращаются к каким-то общим ресурсам, которые определены вне потока, например, обращение к какому-то файлу или подключение к базе данных. Если одновременно несколько потоков обратятся к такому ресурсу на запись (т.е. с целью изменения состояния), то результаты выполнения программы будут сложно предсказуемыми или это может привести к ошибке выполнения.
static Integer object = Integer.valueOf(0);
public static void main(String[] args) throws InterruptedException {
Runnable task = () -> {
object = object + 1;
System.out.println(Thread.currentThread().getName());
};
Thread thread = new Thread(task);
thread.start();
System.out.println(Thread.currentThread().getName());
object = object + 1;
System.out.println(object.intValue());
}Попробуйте несколько раз запустить данный код, чтобы убедиться, что последовательность запуска потоков отличается каждый раз. Самым простым способом синхронизировать потоки (т.е. определить их поведение при работе с общим объектом) — это концепция «монитора» и ключевое слово synchronized (обратите внимание, что у любого наследника класса Object есть свой собственный «монитор» — именно поэтому нельзя синхронизировать примитивные типы). Монитор характеризуется следующей информацией:
● состоянии (locked) — признак, что монитор захвачен потоком;
● владелец (owner) — каким потоком захвачен монитор в текущий момент;
● перечень потоков, которые не смогли захватить монитор (blocked set), так как монитор захвачен другим потоком;
● перечень потоков у которых был вызван метод wait (wait set). Для синхронизации потоков используется ключевое слово synchronized (обратите внимание, что всегда синхронизируется именно объект — только у него есть монитор). Давайте посмотрим, что происходит, когда поток пытается захватить монитор объекта:
● Поток попадает в синхронизированный блок кода (synchronized блок);
● В синхронизированном объекте проверяются переменные locked и owner монитора.
● Если эти поля false и null, соответственно, они заполняются. Если переменная owner не равна потоку, который хочет захватить монитор, то поток блокируется и попадает в blocked set монитора
● Поток начал выполнять код (соответствует открывающей фигурной скобке synchronized блока)
● Поток завершил выполнение кода (соответствует закрывающейся фигурной скобке блока синхронизации)
● Переменные locked и owner монитора очищаются. Теперь давайте посмотрим как мы можем синхронизировать соответствующий объект. Во-первых, можно синхронизировать методы самого класса. В этому случае объектом синхронизации является сам объект — this:
public synchronized void doSomething(){
// .. реализация бизнес логики метода
}Во вторых, можно синхронизировать другой объект в целевом классе:
public static void main(String[] args) throws InterruptedException {
Object objectToLock = new Object();
Runnable task = () -> {
synchronized (objectToLock){
System.out.println(Thread.currentThread().getName());
}
};
Thread thread = new Thread(task);
thread.start();
synchronized (objectToLock){
for (int i = 0; i < 10; i++) {
Thread.currentThread().sleep(1000);
System.out.println("step" + i);
}
}
System.out.println(Thread.currentThread().getName());
}В третьих, в случае синхронизации статических методов мы используем монитор
самого класса:
public static void doWork(){
synchronized (MyClass.class){
// Реализация логики
}
}Примечание. Почему же тогда просто не сделать все методы синхронизированными? Проблема в том, что синхронизация и переключение между потоками ресурсоемкая операция (время выполнения увеличивается и приложение начинает работать медленнее), поэтому использовать данные механизмы нужно осторожно, только там, где между потоками существует конкуренция за ресурсы. Более того, неверное проектирование многопоточного исполнения программ, может привести к Deadlock. Deadlock (взаимная блокировка потоков) — это ошибка в многопоточном программировании, которая происходит когда несколько потоков имеют циклическую зависимость от пары синхронизированных объектов. Пусть один thread входит в монитор объекта A, а другой — объекта B. Если thread в объекте A пытается вызвать любой синхронизированный метод объекта B, а объект B в то же самое время пытается вызвать любой синхронизированный метод объекта A, то потоки будут заблокированы в процессе ожидания “навечно”.
И так мы теперь можем создать и запустить поток, а также синхронизировать потоки при использовании общих ресурсов, но в процессе работы приложения, довольно часто могут возникать ситуации, что поток ожидает события, необходимого для продолжения выполения своей работу (например ответ от web сервера на http запрос) — такой поток блокирует работу всего приложения — в этом случае логичным решением является уступить ресурс, другому потоку, который в текущий момент времени может выполнять свои задачи. У класса Object (а значит у всех не примитивных типов), есть следующие методы, которые позволяют управлять переключением потока:
● wait() — После вызова этого метода поток попадает в wait set монитора, сам же монитор освобождается (переменные locked и owner в мониторе очищаются)
● notify() — Для того чтобы потоки, которые находятся в wait set, продолжили свое выполнение, другой поток должен захватить монитор и вызвать методы notify(). После вызова метода notify() из wait set выбирается произвольный поток и переводится в blocked set. После того как этот поток выйдет из synchronized блока, нотифицированные потоки будут по одному захватывать монитор и продолжать выполнение
● notifyAll() — аналогично notify(), но все потоки из wait set переводятся в blocked set На схемы мы видим переходы потока из одного состояния в другое.
Java memory model
Без преувеличения многопоточность одна из самых сложных и комплексных тем в программировании. Пока мы познакомились, только с базовыми механизмами создания и управления потоками, но прежде чем перейти к изучению паттернов и механизмов, применяемых в многопоточном программировании, нам необходимо понять, причины не очевидного поведения таких программ. Для этого нам предстоит узнать как Java работает с памятью и как это влияет на многопоточность. Java Memory Model, JMM или модель памяти Java описывает поведение потоков в среде исполнения Java. Модель памяти — это часть семантики языка Java, которая определяет на что может и на что не должен рассчитывать разработчик при работе с потоками. Разумеется JMM посвящена большая глава в спецификации Java - сегодня мы рассмотрим только основные моменты. Первое что нам необходимо понять, как Java структурирует выделенную ей для работы память. JVM использует следующие типы памяти: Heap – это регион памяти, где хранятся объекты Java. Куча является общей для всех потоков, ее содержимое управляется сборщиком мусора. Куча содержит все объекты, созданные в вашем приложении, независимо от того, какой поток создал объект (к этому относятся и обертки примитивных типов). Stack – это область памяти, где хранятся локальные переменные и стек вызовов методов. Для каждого потока создается отдельный стек в JVM. Стек содержит все локальные переменные для каждого метода. Соответствующий поток может получить доступ только к своему стеку - локальные переменные (т.е. переменные инициализированные в методе), невидимы для других потоков, кроме потока, который их создал. Представим, что два потока выполняют один и тот же код, т.е. вызвали один и тот же метод, всё равно все локальные переменные, будут созданы своих собственных стеках. Таким образом, каждый поток имеет свою версию локальной переменной соответствующего метода.
● Все локальные переменные примитивных типов (boolean, byte, short, char, int, long, float, double) полностью хранятся в стеке потоков и не видны другим потокам.
● Локальная переменная также может быть ссылкой на объект. В этом случае ссылка (локальная переменная) хранится в стеке потоков, но сам объект хранится в куче.
● Статические переменные класса также хранятся в куче вместе с определением класса. К объектам в куче могут обращаться любые потоки, имеющие ссылку на соответствующий объект. Когда поток имеет доступ к объекту, он также может получить доступ к переменным-членам этого объекта. Если два потока вызывают метод для одного и того же объекта одновременно, они оба будут иметь доступ к переменным-членам объекта, но каждый поток будет иметь свою собственную копию локальных переменных.. Method Area – это область памяти, где хранятся информация о классах и методах JVM. Здесь также хранятся константы и статические переменные. Program Counter Register – это регистр, который указывает на следующую инструкцию, которую нужно выполнить в текущем потоке. Native Method Stack – это стек, используемый для выполнения нативного кода. Итак мы разобрали, как Java управляет памятью, теперь давайте обсудим, что может произойти, когда объекты и переменные хранятся в различных областях памяти — возникают следующие проблемы: Видимость изменений, которые произвел поток над общими переменными. Пусть два или более потока работают с общим объектом (без использования volatile-объявления или синхронизации), то изменения в этом объекта, сделанные одним из потоком, могут быть невидимы для други. Представьте: общий объект изначально хранится в Heap. Поток, выполняющийся на CPU, считывает общий объект в кэш этого же CPU. Там он вносит изменения в объект. Пока кэш CPU не был сброшен в основную память, измененная версия общего объекта не видна другим потокам. Таким образом, каждый поток может получить свою собственную копию общего объекта, каждая копия будет находиться в отдельном кэше CPU. Состояние гонки при чтении, проверке и записи общих переменных. Пусть два или более потоков совместно используют один объект и более одного потока одновременно меняют состояние этого объекта (т.е. обновляют переменные), тогда может возникнуть состояние гонки или race condition. По сути это нарушения алгоритма выполнения задач потоками - один поток начинает свое выполнение раньше, чем это необходимо. Представьте: поток X считывает переменную объекта в кэш своего процессора. Представьте также, что поток Y делает то же самое, но в кэш другого процессора. Теперь поток X прибавляет 1 к значению переменной count, и поток Y делает то же самое. Теперь переменная была увеличена дважды. Если бы эти операции были выполнены последовательно (как и было задумано), переменная была бы увеличена дважды. Тем не менее, обе операции были выполнены одновременно без использования синхронизации. Независимо от того, какой из потоков, записывает свою версию переменной в основную память, новое значение будет только на 1 больше исходного.
● Видимость изменений
● Race condition
Volatile Ключевое слово volatile указывает, что взаимодействие потоков с этой переменной должно происходить минуя кэш процессора, т. е. напрямую (т.е. запрещено копировать переменную из Heap). Когда потоки используют переменные (не volatile), они могут копировать значение этих переменных в кэш CPU для улучшения производительности. Если вы используете процессор с несколько ядрами, при этом каждый из потоков выполняется на отдельном ядре, то одна и та же переменная может находиться в разном состоянии на каждом ядре CPU. В результате мы получим несколько копий одной и той же переменной: копии в кэше каждого ядра процессора и копия переменной в Heap. При использовании не volatile переменных нельзя знать наверняка, когда JVM читает значение переменной из главной памяти и когда записывается значение в главную память. Примечание. Помните, что объявление переменной как volatile достаточно, только когда один поток изменяет переменную, а другой поток читает ее значение - Если два потока одновременно меняют состояние переменной, то volatile уже недостаточно — все равно может быть race condition. Итак, ключевое слово volatile гарантирует нам следующее:
● Если поток X пишет в volatile переменную, а затем второй поток Y читает значение, тогда все переменные, видимые потоку X перед записью в переменную volatile, также будут видны потоку Y после того как он прочитал переменную volatile.
● Если поток X читает переменную volatile, то все переменные, видимые потоку
X при чтении переменной volatile, также будут перечитаны из основной памяти.
● Применение volatile Атомарные переменные И так, теперь давайте рассмотрим ситуацию, когда два или более потоков пытаются изменить общий разделяемый ресурс: одновременно выполнять операции чтения и записи. Чтобы избежать состояния гонки, мы можем использовать synchronized-методы, synchronized-блоки (volatile нам в данном случае не подойдет). Без использования механизма синхронизации, результат может быть непредсказуемым, а значение не будет иметь никакого смысла для нашей программы. Но как было уже сказано выше, процесс синхронизации дорогостоящая операция, требующая существенных ресурсов CPU. Также этот способ блокирующий — одновременно выполняется только один поток, что сильно влияет на производительность системы в целом. Для решения этой проблемы мы можем использовать неблокирующие алгоритмы.. Эти алгоритмы называются compare and swap (CAS) и базируются на том, что современные процессоры поддерживают некоторые операции на уровне машинных инструкций. В Java нам доступны классы атомарных переменных: AtomicInteger AtomicLong, AtomicBoolean, AtomicReference. Алгоритм compare and swap работает следующим образом: есть ячейка памяти, текущее значение в ней и то значение, которое хотим записать в эту ячейку. Сначала ячейка памяти читается и сохраняется текущее значение, затем прочитанное значение сравнивается с тем, которое уже есть в ячейке памяти, и если значение прочитанное ранее совпадает с текущим, происходит запись нового значения. Следует упомянуть, что значение переменной после чтения может быть изменено другим потоком, потому что CAS не является блокирующей операцией.
● Применение атомарных переменных Неизменяемые объекты И так главная проблема с которой мы сталкиваемся, при работе с несколькими потоками это использование общих ресурсов, которые могут изменить свое состояние (т.е. изменение значений атрибутов объекта) в процессе выполнения. Пожалуй одно из самых эффективных способов решения данной проблемы это использование неизменяемых объектов (механизмы синхронизации просто не нужны). Вы уже встречались с подобными объектами — любая обертка примитивного типа неизменяемый объект. Давайте обсудим принципы, которые позволяют нам проектировать неизменяемые объекты:
● Проверьте отсутствие mutable методов — т.е. классе не должно быть ни одного публичного метода, который могут бы изменить состояние объекта.
● Все поля следует объявить как private final — это гарантирует, что если поле ссылается на примитивный тип оно никогда не измениться, если на ссылочный тип то ссылка не может быть изменена.
● Если метод вашего класса возвращает изменяемый объект, то возвращайте его копию.
● Если при вызове конструктора вы передаете в него изменяемый объект, который должен быть присвоен полю, то создавайте его копию.
● Объявите ваш класс final
● Демонстрация процесса проектирования неизменного класса ThreadLocal
Если в рамках реализации задачи, необходимо гарантировать, что каждый поток будет работать со своей уникальной переменной — воспользуйтесь классом ThreadLocal.
Класс ThreadLocal представляет хранилище тред-локальных переменных. По способу использования он похож на обычную обертку над значением, с методами get(), set() и remove() для доступа к нему, и дополнительным фабричным методом ThreadLocal.withInitial(), устанавливающим значение по-умолчанию. Отличие такой переменной от обычной в том, что ThreadLocal хранит отдельную независимую копию значения для каждого ее использующего потока — соответственно работа с такой переменной потокобезопасна. Таким образом объект класса ThreadLocal хранит внутри не одно конкретное значение, а хэш-таблицу (поток и соответствующее значение), и при использовании обращается к значению для текущего потока.