Межпотоковое взаимодействие в Java: wait, notify, notifyAll

Взаимное исключение обеспечивает безопасное совместное состояние, но не позволяет одному потоку сигнализировать другому об изменении состояния. Именно для этого предназначено трио wait, notify и notifyAll из java.lang.Object. Это низкоуровневый примитив координации, предоставляемый Java — все механизмы более высокого уровня (блокирующие очереди, защёлки, семафоры, Condition) построены на той же идее: поток ожидает внутри монитора, пока другой поток не скажет ему проснуться.

Современный код редко вызывает wait/notify напрямую. Вместо них используются BlockingQueue, CountDownLatch или Condition. Но нужно знать базовый механизм, потому что (а) именно его используют эти классы под капотом, (б) он применяется в каждой читаемой вами библиотеке, и (в) когда что-то идёт не так с высокоуровневым кодом, диагностика нередко ведёт вплоть до пропущенного notify.

Трио методов

Определены в java.lang.Object, поэтому они есть у каждого объекта:

void wait() throws InterruptedException;
void wait(long timeoutMillis) throws InterruptedException;
void notify();
void notifyAll();

Жёсткое правило: вызывать эти методы можно только при владении монитором объекта, на котором они вызываются. То есть вы должны находиться внутри блока synchronized (obj) { ... } (или synchronized-метода, блокирующего тот же obj). Вызов obj.wait() без владения монитором obj немедленно выбрасывает IllegalMonitorStateException.

synchronized (lock) {
  lock.wait();                                  // ok — we hold lock
  lock.notify();                                // ok — same
}
lock.wait();                                    // IllegalMonitorStateException

Именно это правило делает API работоспособным: ожидание и уведомление гарантированно происходят при удержании блокировки, поэтому состояние, о котором они говорят, остаётся согласованным.

Что на самом деле делает `wait()`

wait() — это не «сон». Он атомарно выполняет три действия:

Освобождает монитор объекта, на котором был вызван.
Паркует текущий поток в множестве ожидания этого монитора.
При пробуждении (через notify/notifyAll/interrupt/таймаут) повторно захватывает монитор перед возвратом.

Часть «атомарно освобождает и паркует» делает wait безопасным: notify, пришедший в промежутке между «мы решили ждать» и «мы фактически начали ждать», иначе был бы потерян. С wait этого промежутка не существует.

После возврата из wait() вы снова находитесь внутри блока synchronized с удержанной блокировкой — именно поэтому код после wait() может безопасно читать общее состояние.

Что делают `notify()` и `notifyAll()`

notify() выбирает один поток (какой именно — определяет JVM, как правило не FIFO) из множества ожидания и переводит его из состояния WAITING/TIMED_WAITING в BLOCKED. Уведомлённый поток всё ещё ожидает монитора; уведомитель по-прежнему его удерживает. Уведомлённый поток может повторно захватить монитор только после того, как уведомитель выйдет из блока synchronized.

notifyAll() пробуждает каждый поток из множества ожидания таким же образом. Все переходят в состояние BLOCKED; все выстраиваются в очередь за блокировкой; повторно захватывают её по одному по мере высвобождения.

notify быстрее (просыпается один поток), но опасен: если пробудить не тот поток (тот, чьё условие фактически не выполнено), он возвращается к wait() и ничего полезного не происходит. notifyAll безопаснее (хоть кто-то из ожидающих сможет продвинуться), но дороже. По умолчанию используйте notifyAll; переключайтесь на notify только когда можете доказать, что все ожидающие взаимозаменяемы.

Обязательный паттерн цикла `while`

Самое важное правило при использовании wait:

Всегда вызывайте wait() внутри цикла while, повторно проверяющего условие.

synchronized (lock) {
  while (!conditionHolds()) {
    lock.wait();
  }
  // now condition holds AND we own the lock
}

Три причины использовать цикл, а не if:

Ложные пробуждения. JVM разрешено будить wait без какой-либо причины. Цикл их перехватывает.
notifyAll будит более одного потока. Когда все они борются за блокировку, победитель может не иметь ничего полезного делать — кто-то другой уже потребил ресурс. Цикл отправляет его обратно в wait.
Другое состояние может измениться. Между notify и моментом повторного захвата блокировки кто-то другой с этой блокировкой мог отменить то, чего вы ждали. Цикл перепроверяет.

if (!condition) wait() — самая распространённая ошибка в коде wait/notify. В тестах работает; в продакшне ломается в три часа ночи.

Классический сценарий производитель–потребитель

Канонический вариант использования wait/notify — ограниченный буфер:

class Buffer<T> {
  private final Object lock = new Object();
  private final Object[] data;
  private int count, head, tail;

  Buffer(int capacity) { data = new Object[capacity]; }

  void put(T item) throws InterruptedException {
    synchronized (lock) {
      while (count == data.length) lock.wait();             // wait for room
      data[tail] = item;
      tail = (tail + 1) % data.length;
      count++;
      lock.notifyAll();                                      // wake any consumer
    }
  }

  @SuppressWarnings("unchecked")
  T take() throws InterruptedException {
    synchronized (lock) {
      while (count == 0) lock.wait();                        // wait for an item
      T item = (T) data[head];
      data[head] = null;
      head = (head + 1) % data.length;
      count--;
      lock.notifyAll();                                      // wake any producer
      return item;
    }
  }
}

Несколько вещей, которые здесь сделаны правильно:

Одна блокировка для обоих методов (lock). Один монитор защищает всё состояние.
Оба ожидания находятся внутри циклов while.
notifyAll на обеих сторонах — потому что и производители, и потребители ждут на одном мониторе, а пробуждение лишь одного может оказаться ошибочным.
Блокировка удерживается во время wait (сам wait освобождает её внутри, а затем повторно захватывает перед возвратом).

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

Почему `notifyAll` — более безопасный вариант по умолчанию

Если заменить notifyAll на notify в буфере выше, возникнет тонкая ошибка. Два потребителя и один производитель ждут на одном мониторе. Производитель вызывает notify; JVM выбирает поток; если выбирается потребитель, когда пробуждение предназначалось для «в очереди есть место» (для потребителей неактуально), потребитель перепроверяет своё условие (очередь всё ещё возможно пуста), возвращается к wait, а производитель, которого нужно было разбудить, так и не просыпается. Очередь зависает, исключений нет.

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

Устаревшие альтернативы

В старом коде встречаются Thread.suspend() и Thread.resume(). Не используйте их. Они были помечены устаревшими ещё в Java 1.2, поскольку оставляют блокировки удержанными и нарушают инварианты. Механизм wait/notify — единственный безопасный способ заставить один поток ждать другого, используя только методы Object.

Есть также Thread.sleep — но sleep не освобождает блокировки. Поток, спящий внутри блока synchronized, блокирует все остальные потоки, желающие той же блокировки, до своего пробуждения. Используйте wait (который освобождает блокировку) для любого сценария «ждать пока что-то произойдёт»; оставьте sleep для «ждать фиксированное время, не удерживая ничего важного».

Что использовать вместо этого в продакшне

wait/notify корректны, но склонны к ошибкам. Современный код предпочитает высокоуровневые строительные блоки:

Задача	Использовать
Ограниченный производитель–потребитель	`ArrayBlockingQueue`, `LinkedBlockingQueue`
Ожидание завершения N вещей	`CountDownLatch`
Ожидание встречи всех N участников	`CyclicBarrier`, `Phaser`
Несколько условных переменных на одной блокировке	`Condition` (из `ReentrantLock.newCondition()`)
Разрешения на ресурс	`Semaphore`
Результат будущего одноразового вычисления	`CompletableFuture`

В каждом из них правильный цикл while, правильная семантика notifyAll/signalAll и правильная обработка прерываний встроены изначально. Мы познакомимся со всеми ними в этой части книги.

Разобранный пример: производитель–потребитель с `wait` и `notifyAll`

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

java— editable, runs on the server

import java.util.concurrent.atomic.AtomicInteger;

public class WaitNotifyDemo {
  static class Buffer {
    private final Object lock = new Object();
    private final int[] data;
    private int count, head, tail;
    Buffer(int cap) { data = new int[cap]; }

void put(int v) throws InterruptedException {
      synchronized (lock) {
        while (count == data.length) {
          lock.wait();                                       // releases lock, parks
        }
        data[tail] = v;
        tail = (tail + 1) % data.length;
        count++;
        lock.notifyAll();                                    // wake any consumer
      }
    }

int take() throws InterruptedException {
      synchronized (lock) {
        while (count == 0) lock.wait();
        int v = data[head];
        head = (head + 1) % data.length;
        count--;
        lock.notifyAll();                                    // wake any producer
        return v;
      }
    }
  }

public static void main(String[] args) throws InterruptedException {
    Buffer buf = new Buffer(4);                              // small buffer => high contention
    int totalItems = 2000;
    AtomicInteger remaining = new AtomicInteger(totalItems);
    AtomicInteger consumedSum = new AtomicInteger();

// 3 consumers
    Thread[] consumers = new Thread[3];
    for (int c = 0; c < consumers.length; c++) {
      final int id = c;
      consumers[c] = new Thread(() -> {
        try {
          while (remaining.getAndDecrement() > 0) {
            int v = buf.take();
            consumedSum.addAndGet(v);
          }
          // we over-decremented at the end; restore so we don't take extra
          remaining.incrementAndGet();
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }, "consumer-" + id);
      consumers[c].start();
    }

// 2 producers, 1000 items each
    long producedSum = 0;
    Thread[] producers = new Thread[2];
    for (int p = 0; p < producers.length; p++) {
      final int base = p * 1000;
      producers[p] = new Thread(() -> {
        try {
          for (int i = 0; i < 1000; i++) buf.put(base + i);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }, "producer-" + p);
      producers[p].start();
    }
    for (int i = 0; i < 2000; i++) producedSum += i;

for (Thread p : producers) p.join();
    for (Thread c : consumers) c.join();

System.out.println("produced sum: " + producedSum);
    System.out.println("consumed sum: " + consumedSum.get());
    System.out.println("match? " + (producedSum == consumedSum.get()));

// Demonstrate the rule: wait outside synchronized throws
    Object lock = new Object();
    try {
      lock.wait(1);                                          // no monitor — boom
    } catch (IllegalMonitorStateException e) {
      System.out.println("\nwait() outside synchronized: " + e.getClass().getSimpleName());
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
}

Что стоит вынести из результата выполнения:

Суммы совпали. Каждый элемент, добавленный производителем, был взят ровно одним потребителем; ничего не было продублировано, ничего не потеряно. Это свойство корректности производитель–потребитель, достигнутое всего одним монитором и парой wait/notifyAll.
Буфер был всего 4 слота, поэтому производители постоянно заполняли его, а потребители постоянно опустошали. Циклы while позволяли им парковаться и снова парковаться по мере цикличного использования очереди. Без wait производители крутились бы на count == capacity, сжигая CPU; с ним — они спят до тех пор, пока потребитель не подаст сигнал.
notifyAll вызывался на той же блокировке, которую удерживали и производители, и потребители. Это весь механизм координации: один монитор, взаимное исключение и сигнализация, с циклом while, перехватывающим любое нерелевантное пробуждение.
Последний wait вне блока synchronized немедленно выбросил IllegalMonitorStateException. Это исполнение JVM правила: ждать/уведомлять можно только на мониторе, которым вы сейчас владеете. Если вы видите это исключение, путь кода дошёл до wait без предшествующего synchronized.
Та же форма — ограниченный буфер, взаимное исключение, сигнал при каждом изменении состояния — это то, что ArrayBlockingQueue делает внутри, только использует два объекта Condition (один для «не полон», другой для «не пуст») вместо одного большого notifyAll. Это правильный способ написать такое в продакшне; версия с wait/notifyAll — базовый механизм, на котором построены все высокоуровневые классы.

Что дальше

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

Практика

Почему `obj.wait()` всегда должен вызываться из блока `synchronized (obj)` (или `synchronized`-метода, блокирующего `obj`)?

`wait` атомарно освобождает монитор и паркует поток — JVM требует, чтобы вызывающий владел монитором для его освобождения, иначе выбрасывается `IllegalMonitorStateException`Это лишь соглашение; JVM допускает `wait` откуда угодно, но код сложнее рассуждать`wait` требует `synchronized` для атомарности вызова с окружающим `notify` — без него пробуждение могло бы гонять`synchronized` лениво выделяет очередь ожидания; вызов `wait` первым создаёт объект очереди, который нужен JVM

Трио методов

Что на самом деле делает wait()

Что делают notify() и notifyAll()

Обязательный паттерн цикла while