1. введение
В этой статье мы рассмотрим один из самых фундаментальных механизмов в Java – синхронизацию потоков.
Сначала мы обсудим некоторые основные термины и методологии, связанные с параллелизмом.
Дальнейшее чтение:
Руководство по ключевому слову Synchronized в Java
Параллелизм Java
Как запустить поток в Java
И мы разработаем простое приложение, в котором мы будем решать проблемы параллелизма с целью лучшего понимания wait() и notify().
2. Синхронизация потоков в Java
В многопоточной среде несколько потоков могут попытаться изменить один и тот же ресурс. Если потоки не управляются должным образом, это, конечно, приведет к проблемам с согласованностью.
2.1. Охраняемые блоки в Java
Один из инструментов, который мы можем использовать для координации действий нескольких потоков в Java, – это защищенные блоки. Такие блоки перед возобновлением выполнения проверяют наличие определенного условия.
Имея это в виду, мы воспользуемся:
- Object.wait() – для приостановки потока
- Object.notify() – для пробуждения потока
Это можно лучше понять из следующей диаграммы, на которой показан жизненный цикл Нить :
Обратите внимание, что существует множество способов управления этим жизненным циклом; однако в этой статье мы сосредоточимся только на wait() и notify().
3. Метод wait()
Проще говоря, когда мы вызываем wait ()– , это заставляет текущий поток ждать, пока какой-либо другой поток не вызовет notify() или notifyAll() на том же объекте.
Для этого текущий поток должен владеть объектом monitor . Согласно Javadocs , это может произойти, когда:
- мы выполнили метод synchronized экземпляра для данного объекта
- мы выполнили тело блока synchronized на данном объекте
- путем выполнения синхронизированных статических методов для объектов типа Класса
Обратите внимание, что только один активный поток может одновременно владеть монитором объекта.
Этот метод wait() поставляется с тремя перегруженными сигнатурами. Давайте взглянем на это.
3.1. подождите()
Метод wait() заставляет текущий поток бесконечно ждать, пока другой поток либо не вызовет notify() для этого объекта, либо notifyAll() .
3.2. ожидание(длительный тайм-аут)
Используя этот метод, мы можем указать тайм-аут, после которого поток будет автоматически пробужден. Поток можно разбудить до достижения тайм-аута с помощью notify() или notifyAll().
Обратите внимание, что вызов wait(0) совпадает с вызовом wait().
3.3. ожидание(длительный тайм-аут, int nanos)
Это еще одна сигнатура, обеспечивающая ту же функциональность, с той лишь разницей, что мы можем обеспечить более высокую точность.
Общий период ожидания (в наносекундах) рассчитывается как 1_000_000*тайм-аут + наносекунды.
4. notify() и notifyAll()
Метод notify() используется для пробуждения потоков, ожидающих доступа к монитору этого объекта.
Существует два способа уведомления ожидающих потоков.
4.1. уведомить()
Для всех потоков, ожидающих на мониторе этого объекта (используя любой из методов wait () ), метод notify() уведомляет любой из них о произвольном пробуждении. Выбор того, какой именно поток следует разбудить, не является детерминированным и зависит от реализации.
Поскольку notify() пробуждает один случайный поток, его можно использовать для реализации взаимоисключающей блокировки, когда потоки выполняют аналогичные задачи, но в большинстве случаев было бы более целесообразно реализовать notifyAll() .
4.2. Уведомить всех()
Этот метод просто пробуждает все потоки, ожидающие на мониторе этого объекта.
Пробужденные потоки завершатся обычным образом – как и любой другой поток.
Но прежде чем мы позволим продолжить их выполнение, всегда определите быструю проверку условия, необходимого для продолжения потока – потому что могут возникнуть некоторые ситуации, когда поток был разбужен без получения уведомления (этот сценарий обсуждается позже в примере).
5. Проблема Синхронизации Отправителя И Получателя
Теперь, когда мы понимаем основы, давайте рассмотрим простое приложение Sender – Receiver , которое будет использовать методы wait() и notify() для настройки синхронизации между ними:
- Отправитель должен отправить пакет данных Получателю
- Получатель не может обработать пакет данных до тех пор, пока Отправитель не закончит его отправку
- Аналогично, Отправитель не должен пытаться отправить другой пакет, если Получатель уже не обработал предыдущий пакет
Давайте сначала создадим Data класс, состоящий из данных пакета , которые будут отправлены от Отправителя к Получателю. Мы будем использовать wait() и notifyAll() для настройки синхронизации между ними:
public class Data {
private String packet;
// True if receiver should wait
// False if sender should wait
private boolean transfer = true;
public synchronized void send(String packet) {
while (!transfer) {
try {
wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
Log.error("Thread interrupted", e);
}
}
transfer = false;
this.packet = packet;
notifyAll();
}
public synchronized String receive() {
while (transfer) {
try {
wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
Log.error("Thread interrupted", e);
}
}
transfer = true;
notifyAll();
return packet;
}
}Давайте разберемся, что здесь происходит:
- Переменная packet обозначает данные, передаваемые по сети
У нас есть логическая переменная transfer – , которую Отправитель и Получатель будут использовать для синхронизации:
- Если эта переменная true , то Получатель должен дождаться Отправителя для отправки сообщения
- Если это false , Отправитель должен дождаться Получателя , чтобы получить сообщение
Отправитель использует send() метод для отправки данных Получателю :
- Если transfer имеет значение false, мы будем ждать, вызвав wait() в этом потоке
- Но когда это true , мы переключаем статус, устанавливаем наше сообщение и вызываем notifyAll () , чтобы разбудить другие потоки, чтобы указать, что произошло значительное событие, и они могут проверить, могут ли они продолжить выполнение
Аналогично, Receiver будет использовать метод receive() :
- Если transfer был установлен в false отправителем , то только он будет продолжен, в противном случае мы вызовем wait() в этом потоке Когда условие выполнено, мы переключаем статус, уведомляем все ожидающие потоки о пробуждении и возвращаем пакет данных, который был
- Получателем
5.1. Зачем заключать wait() в цикл while?
Поскольку notify() и notifyAll() случайным образом пробуждает потоки, ожидающие на мониторе этого объекта, не всегда важно, что условие выполнено. Иногда может случиться так, что поток просыпается, но условие на самом деле еще не выполнено.
Мы также можем определить проверку, чтобы спасти нас от ложных пробуждений – когда поток может проснуться от ожидания, даже не получив уведомления.
5.2. Зачем нам нужно синхронизировать методы send() и receive ()?
Мы поместили эти методы в synchronized методы, чтобы обеспечить внутренние блокировки. Если поток, вызывающий метод wait () , не обладает встроенной блокировкой, будет выдана ошибка.
Теперь мы создадим Sender и Receiver и реализуем интерфейс Runnable на обоих, чтобы их экземпляры могли выполняться потоком.
Давайте сначала посмотрим, как будет работать Отправитель :
public class Sender implements Runnable {
private Data data;
// standard constructors
public void run() {
String packets[] = {
"First packet",
"Second packet",
"Third packet",
"Fourth packet",
"End"
};
for (String packet : packets) {
data.send(packet);
// Thread.sleep() to mimic heavy server-side processing
try {
Thread.sleep(ThreadLocalRandom.current().nextInt(1000, 5000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
Log.error("Thread interrupted", e);
}
}
}
}Для этого Отправитель :
- Мы создаем несколько случайных пакетов данных, которые будут отправляться по сети в пакетах[] массиве
- Для каждого пакета мы просто вызываем send()
- Затем мы вызываем Thread.sleep() со случайным интервалом, чтобы имитировать тяжелую обработку на стороне сервера
Наконец, давайте реализуем наш Приемник :
public class Receiver implements Runnable {
private Data load;
// standard constructors
public void run() {
for(String receivedMessage = load.receive();
!"End".equals(receivedMessage);
receivedMessage = load.receive()) {
System.out.println(receivedMessage);
// ...
try {
Thread.sleep(ThreadLocalRandom.current().nextInt(1000, 5000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
Log.error("Thread interrupted", e);
}
}
}
}Здесь мы просто вызываем load.receive() в цикле, пока не получим последний “End” пакет данных.
Давайте теперь посмотрим на это приложение в действии:
public static void main(String[] args) {
Data data = new Data();
Thread sender = new Thread(new Sender(data));
Thread receiver = new Thread(new Receiver(data));
sender.start();
receiver.start();
}Мы получим следующий результат:
First packet Second packet Third packet Fourth packet
И вот мы здесь – мы получили все пакеты данных в правильном, последовательном порядке и успешно установили правильную связь между нашим отправителем и получателем.
6. Заключение
В этой статье мы обсудили некоторые основные концепции синхронизации в Java; более конкретно, мы сосредоточились на том, как мы можем использовать wait() и notify() для решения интересных проблем синхронизации. И, наконец, мы рассмотрели пример кода, в котором мы применили эти концепции на практике.
Прежде чем мы закончим здесь, стоит упомянуть, что все эти низкоуровневые API, такие как wait () , notify() и notifyAll () -это традиционные методы, которые хорошо работают, но механизм более высокого уровня часто проще и лучше – например, собственные интерфейсы Java Lock и Condition (доступны в java.util.concurrent.замки пакет).
Для получения дополнительной информации о пакете java.util.concurrent посетите нашу статью обзор java.util.concurrent , а Блокировка и Условие описаны в руководстве по java.util.concurrent.Замки, здесь .
Как всегда, полные фрагменты кода, используемые в этой статье, доступны на GitHub.