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

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

Взаимодействие потоков

В этой статье мы рассмотрим каким образом потоки могут взаимодействовать друг с другом. В классе Object есть три метода wait(), notify(), и notifyAll(), которые позволяют потоку сообщать информацию о своем состоянии другим, заинтересованным в этой информации, потокам.

Например. Есть два потока. Первый занимается доставкой почты, второй – обрабатывает почту. Обработчик почты должен постоянно проверять, есть ли в очереди письма, которые требуется обработать. Используя механизмы ожидания (wait) и уведомления (notify), поток-обработчик почты может посмотреть состояние Входящих и если там ничего нет сказать: “Эй, я не собираюсь тратить свое время на проверку ящика каждую секунду. Я пойду посплю, а когда доставщик почты что-нибудь положит во Входящие, предупредите меня чтобы я мог вернуться к работе.”

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

Важный момент при использовании механизма wait/notify заключается в следующем: wait(), notify(), и notifyAll() должны вызываться из синхронизированного (synchronized) контекста, иначе получим исключительную ситуацию!

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

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

    @Override
    public void run() {
        while (true) {
            // Получить форму от пользователя
            // Рассчитать необходимые для создания формы шаги (операции машины)
            // Отправить рассчитанные шаги аппаратуре
        }
    }

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

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

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

Следующий псевдокод отображает эту ситуацию:

    public void userLoop() {
        while (true) {
            // Получить форму от пользователя
            // Рассчитать необходимые для создания формы шаги (операции машины)
            // Модификация общего объекта с новыми шагами машины
        }
    }
 
    public void hardwareLoop() {
        while (true) {
            // Получить шаги из общего объекта
            // Отправить шаги аппаратуре
        }
    }

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

Методы wait() и notify() являются методами класса Object. Каждый объект имеет блокировку. Таким же образом, каждый объект может иметь список потоков, которые ждут от него сигнала (уведомления). Поток попадает в этот список ожидающих при помощи вызова wait() целевого объекта. С этого момента он не выполняет ни каких инструкций пока не будет вызван метод notify(). Если один объект ждут несколько потоков, то после вызова notify(), планировщиком потоков будет выбран один поток (какой именно не известно), который приступит к исполнению. Если ожидающих потоков нет, то ничего и не произойдет. Давайте теперь посмотрим на реальный код, в котором один объект ждет уведомления от другого.

package ru.topcode.threadomaniac;
 
public class ThreadA {
    public static void main(String[] args) {
        ThreadB b = new ThreadB();
        b.start();
 
        synchronized (b) {
            try {
                System.out.println("Ждем пока поток b выполнится...");
                b.wait();
            } catch (InterruptedException e) {
            }
 
            System.out.println("Total равно: " + b.total);
        }
    }
}
 
class ThreadB extends Thread {
    int total;
 
    public void run() {
        synchronized (this) {
            for (int i = 0; i < 100; i++) {
                total += i;
            }
 
            notify();
        }
    }
}

Эта программа содержит два потока: ThreadA – основной поток (main) и ThreadB – поток, который вычисляет сумму всех чисел от 0 до 99. После того как выполнится строка b.start(), ThreadA продолжит свое выполнение и может попасть в строку System.out.println(“Total равно: ” + b.total) до того как ThreadB закончит свои расчеты. Чтобы предотвратить такое развитие событий, мы используем метод wait().

Обратите внимание на строку synchronized (b). Это необходимо для того, чтобы ThreadA получил блокировку на b. Чтобы вызвать методы wait() или notify(), поток должен иметь блокировку объекта, на котором вызываются эти методы.

Приведенный выше код уведомляет один поток, ожидающий на этом объекте. Обратите внимание на то, что если вызвать метод wait() из потока не имеющего блокировку на объект на котором вызывается wait(), то возникнет исключение IllegalMonitorStateException. Это исключение не проверяемое (not a checked exception), поэтому в приведенном выше коде мы не ловим его в явном виде. Однако, блок кода, в котором у нас вызывается wait() обрамлен конструкцией try/catch и мы перехватываем исключение InterruptedException, т.к. ожидающий поток может быть прерван таким же образом, как и спящий (sleeping).

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

Немного модифицируем наш пример:

Класс operator:

package ru.topcode.threadomaniac.fabric;
 
class Operator extends Thread {
    public void run() {
        while (true) {
            // Получить форму от пользователя
            synchronized (this) {
                // Вычислить новые шаги для машины по форме
                notify();
            }
        }
    }
}

Класс Machine:

package ru.topcode.threadomaniac.fabric;
 
public class Machine {
    Operator operator; // Предположим, что это поле инициализировано
 
    public void run() {
        while (true) {
            synchronized (operator) {
 
                try {
                    operator.wait();
                } catch (InterruptedException ie) {
                }
 
                // Отправить машинные шаги аппаратному обеспечению
            }
        }
    }
}

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

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

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

Существует вторая форма метода wait(), которая принимает количество миллисекунд в качестве значения максимального времени ожидания. Если поток не сломался (interrupted), он продолжит свое выполнение когда получит уведомление или когда истечет заданный тайм-аут.

synchronized(a){ // Поток получает блокировку на 'a'
a.wait(2000); // Поток отпускает блокировку и ждет уведомления
// Максимум через 2 секунды, поток перейдет в состояние Runnable
}

Использование notifyAll()

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

При вызове метода notifyAll() все потоки получат уведомление и начнут конкурировать за право получить блокировку. Как уже было сказано ранее, объект может иметь множество ожидающих его потоков. Метод notify() воздействует только на один из них. Какой поток получит уведомление – не известно, – это зависит от реализации JVM. Поэтому, когда имеет место ожидание объекта несколькими потоками, лучший способ уведомить все потоки – использовать метод notifyAll().

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

package ru.topcode.threadomaniac.calculator;
 
public class Reader extends Thread {
    Calculator c;
 
    public Reader(Calculator calc) {
        c = calc;
    }
 
    public void run() {
        synchronized (c) {
            try {
                System.out.println("Вычисление...");
                c.wait();
            } catch (InterruptedException e) {
            }
            System.out.println("Total равно: " + c.total);
        }
    }
 
    public static void main(String[] args) {
        Calculator calculator = new Calculator();
 
        new Reader(calculator).start();
        new Reader(calculator).start();
        new Reader(calculator).start();
 
        calculator.start();
    }
}
 
class Calculator extends Thread {
    int total;
 
    public void run() {
        synchronized (this) {
            for (int i = 0; i < 100; i++) {
                total += i;
            }
            notifyAll();
        }
    }
}

Программа запускает три потока, которые ожидают получения результатов расчета, а затем запускает поток, который производит расчеты. Заметьте, что если в методе run() из класса Calculator вызвать notify() вместо notifyAll(), то будет уведомлён только один поток.

использование wait() в цикле

На самом деле, примеры машина/оператор и читатель/калькулятор имеют общую проблему. В каждом есть по крайней мере один поток, который вызывает wait() и другие потоки, которые вызывают notify() или notifyAll(). Казалось бы что все работает как надо. Но что произойдет если, например, поток Calculator запустится первым и вызовет notify() до того, как потоки-читатели начнут ожидание? К сожалению, когда читатели запускаются, они сразу же уходят в состояние ожидания. Они не делают ничего, чтобы убедиться, что событие, которого они ждут, уже произошло. Так, если Calculator уже вызвал notifyAll(), он больше не будет делать этого и читатели будут ждать вечно. Вероятно, что это не то что мы ожидали. Почти всегда, когда нужно ждать какое-то событие, хотелось бы иметь возможность проверить, произошло ли это событие или еще нет. Как правило, лучшее решение – это положить в какой-нибудь цикл условное выражение и ждать только тогда, когда событие не произошло. Например:

class Operator extends Thread {
    Machine machine;
 
    public void run() {
        while (true) {
            Shape shape = getShapeFromUser();
            MachineInstructions job =
                    calculateNewInstructionsFor(shape);
            machine.addJob(job);
        }
    }
}

Operator как и раньше, получает новые формы от пользователя и отправляет их аппаратному обеспечению в бесконечном цикле. Но теперь логика notify() перемещена в метод addJob() класса Machine:

class Machine extends Thread {
    List<MachineInstructions> jobs =
            new ArrayList<MachineInstructions>();
 
    public void addJob(MachineInstructions job) {
        synchronized (jobs) {
            jobs.add(job);
            jobs.notify();
        }
    }
 
    public void run() {
        while (true) {
            synchronized (jobs) {
                // ждать пока не появится хотя бы одно задание
                while (jobs.isEmpty()) {
                    try {
                        jobs.wait();
                    } catch (InterruptedException ie) {
                    }
                }
 
                // Если мы здесь, значит jobs не пустой 
                MachineInstructions instructions = jobs.remove(0);
 
                // Послать шаги машине
            }
        }
    }
}

Аппаратный поток хранит список задач, которые необходимо выполнить. Всякий раз, когда оператору нужно добавить новое задание в список, он вызывает метод addJob(). В это время метод run() просто продолжает выполнение цикла, проверяя наличие заданий в списке. Если заданий нет, то поток ожидает. Если он получает уведомление, то поток прекращает ожидание и перепроверяет условие цикла: пуст ли список? Вероятно, что на практике такая двойная проверка не понадобится, поскольку во время вызова notify() уже известно, что задание добавлено в список. Тем не менее, делать такую проверку – хорошая идея, т.к. по каким-то причинам может возникнуть такая ситуация, что поток получит уведомление, которое ему не предназначено.

Заметьте также, что и метод run() и метод addJob() синхронизированы на одном и том же объекте – списке задач jobs. Это необходимо по двум причинам:

  1. Т.к. мы вызываем wait() и notify() на этом экземпляре, нам нужна синхронизация чтобы не получить исключение IllegalMonitorStateException.
  2. Данные в списке jobs доступны для изменений из двух разных потоков. Синхронизация нам необходима для безопасного доступа к совместно изменяемым данным.

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

Источники

  • Kathy Sierra, Bert Bates. SCJP Sun Certified Programmer for Java 6 Exam 310-065.