Глава 17 ПАРАЛЛЕЛЬНОЕ ВЫПОЛНЕНИЕ

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

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

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

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

Задачи

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

Например, задача LiftOff выводит обратный отсчет перед стартом:

//: concurrency/LiftOff.java

// Реализация интерфейса Runnable.

public class LiftOff implements Runnable {

protected int countDown = 10; // Значение по умолчанию

private static int taskCount = 0;

private final int id = taskCount++;

public LiftOff() {}

public LiftOff(int countDown) {

this.countDown = countDown;

}

public String status() {

return "#" + id + "(" +

(countDown > 0 ? countDown : "Liftoff!") + "), ";

}

public void run() {

while(countDown-- > 0) {

System.out.print(status());

Thread.yield();

}

}

}

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

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

Вызов статического метода Thread.yield() в run() обращен к планировщику потоков (часть потокового механизма Java, обеспечивающая переключение процессора между потоками). Фактически он означает, что очередная важная часть цикла была выполнена и теперь можно на время переключиться на другую задачу. Вызов yield() не обязателен, но в данном примере он обеспечивает более интересные результаты: вы с большей вероятностью увидите, что программный поток прерывает и возобновляет свою работу.

В следующем примере метод run() не выделяется в отдельный программный поток, а просто вызывается напрямую в main() (впрочем, поток все же используется — тот, который всегда создается для main()):

//: concurrency/MainThread.java

public class MainThread {

public static void main(String[] args) {

LiftOff launch = new LiftOff();

launch.run();

}

}

<spoiler text="Output:">

#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),

</spoiler> Класс, реализующий Runnable, должен содержать метод run(), но ничего особенного в этом методе нет — он не обладает никакими особыми потоковыми возможностями. Чтобы использовать многопоточное выполнение, необходимо явно связать задачу с потоком.

Класс Thread

Традиционный способ преобразования объекта Runnable в задачу заключается в передаче его конструктору Thread. Следующий пример показывает, как организовать выполнение LiftOff с использованием Thread:

//: concurrency/BasicThreads.java

// Простейший вариант использования класса Thread..

public class BasicThreads {

public static void main(String[] args) {

Thread t = new Thread(new LiftOff());

t.start();

System.out.println("Waiting for LiftOff");

}

}

<spoiler text="Output:"> (90% match)

Waiting for LiftOff

#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),

</spoiler> Конструктору Thread передается только объект Runnable. Метод start() выполняет необходимую инициализацию потока, после чего вызов метода run() интерфейса Runnable запускает задачу на выполнение в новом потоке.

Из выходных данных видно, что вызов start() быстро возвращает управление (сообщение «Waiting for LiftOff» появляется до завершения отсчета). В сущности, мы вызываем LiftOff.run(), а этот метод еще не завершил свое выполнение; но, поскольку LiftOff.run() выполняется в другом потоке, в потоке main() в это время можно выполнять другие операции. (Данная возможность не ограничивается потоком main() — любой поток может запустить другой поток.) Получается, что программа выполняет два метода сразу — main() и LiftOff.run().

В программе можно легко породить дополнительные потоки для выполнения дополнительных задач:

//: concurrency/MoreBasicThreads.java

// Добавление новых потоков.

public class MoreBasicThreads {

public static void main(String[] args) {

for(int i = 0; i < 5; i++)

new Thread(new LiftOff()).start();

System.out.println("Waiting for LiftOff");

}

}

<spoiler text="Output:"> (Sample)

Waiting for LiftOff

#0(9), #1(9), #2(9), #3(9), #4(9), #0(8), #1(8), #2(8), #3(8), #4(8), #0(7), #1(7),

#2(7), #3(7), #4(7), #0(6), #1(6), #2(6), #3(6), #4(6), #0(5), #1(5), #2(5), #3(5),

#4(5), #0(4), #1(4), #2(4), #3(4), #4(4), #0(3), #1(3), #2(3), #3(3), #4(3), #0(2),

#1(2), #2(2), #3(2), #4(2), #0(1), #1(1), #2(1), #3(1), #4(1), #0(Liftoff!),

#1(Liftoff!), #2(Liftoff!), #3(Liftoff!), #4(Liftoff!)

</spoiler> Из выходных данных видно, что задачи выполняются одновременно друг с другом, с поочередной активизацией и выгрузкой потоков. Переключение осуществляется автоматически планировщиком потоков. Если на компьютере установлено несколько процессоров, планировщик потоков автоматически распределяет потоки между разными процессорами.

При разных запусках программы будут получены разные результаты, поскольку работа планировщика потоков недетерминирована. Более того, вы наверняка увидите значительные различия в результатах работы данной программы-примера, запуская ее на различных версиях пакета JDK. К примеру, предыдущие версии JVM не слишком часто выполняли квантование времени, соответственно, поток 1 мог первым закончить свой цикл, затем все свои итерации произвел бы поток 2, и т. д. Фактически то же самое получилось бы, если бы вызывалась процедура, выполняющая все циклы одновременно, за тем исключением, что запуск совокупности потоков требует больших издержек. Более поздние версии JDK обеспечивают более качественное квантование, и каждый поток регулярно получает свою долю внимания. Как правило, Sun не упоминает о подобных изменениях, так что рассчитывать на определенные «правила поведения» потоков не стоит. Лучше всего при написании кода с потоками занять максимально консервативную позицию.

Когда метод main() создает объекты-потоки Thread, он не сохраняет на них ссылки. Обычный объект, «забытый» таким образом, стал бы легкой добычей сборщика мусора, но только не объект-поток Thread. Каждый поток (Thread) са­мостоятельно «регистрирует» себя, то есть на самом деле ссылка на него где-то существует, и сборщик мусора не вправе удалить его объект.

Исполнители

Исполнители (executors), появившиеся в библиотеке java.util.concurrent в Java SE5, упрощают многозадачное программирование за счет автоматизации управления объектами Thread. Они создают дополнительную логическую прослойку между клиентом и выполнением задачи; задача выполняется не напрямую клиентом, а промежуточным объектом. Исполнители позволяют управлять выполнением асинхронных задач без явного управления жизненным циклом потоков. Именно такой способ запуска задач рекомендуется использовать в JavaSE5/6.

Вместо явного создания объектов Thread в MoreBasicThreads.java мы можем воспользоваться исполнителем. Объект LiftOff умеет выполнять определенную операцию и предоставляет единственный метод для выполнения. Объект ExecutorService умеет создавать необходимый контекст для выполнения объектов Runnable. В следующем примере класс CachedThreadPool создает один поток для каждой задачи. Обратите внимание: объект ExecutorService создается статическим методом класса Executors, определяющим разновидность исполнителя:

//: concurrency/CachedThreadPool.java

import java.util.concurrent.*;

public class CachedThreadPool {

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(new LiftOff());

exec.shutdown();

}

}

<spoiler text="Output:"> (Sample)

#0(9), #0(8), #1(9), #2(9), #3(9), #4(9), #0(7), #1(8), #2(8), #3(8), #4(8),

#0(6), #1(7), #2(7), #3(7), #4(7), #0(5), #1(6), #2(6), #3(6), #4(6), #0(4),

#1(5), #2(5), #3(5), #4(5), #0(3), #1(4), #2(4), #3(4), #4(4), #0(2), #1(3),

#2(3), #3(3), #4(3), #0(1), #1(2), #2(2), #3(2), #4(2), #0(Liftoff!), #1(1),

#2(1), #3(1), #4(1), #1(Liftoff!), #2(Liftoff!), #3(Liftoff!), #4(Liftoff!),

</spoiler> Очень часто для создания и управления всеми задачами в системе достаточно одного исполнителя.

Вызов shutdown() предотвращает передачу Executor новых задач. Текущий поток (в данном случае тот, в котором выполняется main()) продолжает выполняться со всеми задачами, переданными до вызова shutdown(). Работа программы прекращается после завершения всех задач в Executor.

CachedThreadPool в этом примере легко заменяется другим типом Executor. Например, в потоковом пуле фиксированного размера (FixedThreadPool) используется ограниченный набор потоков для выполнения переданных задач:

//: concurrency/FixedThreadPool.java

import java.util.concurrent.*;

public class FixedThreadPool {

public static void main(String[] args) {

// В аргументе конструктора передается количество потоков:

ExecutorService exec = Executors.newFixedThreadPool(5);

for(int i = 0; i < 5; i++)

exec.execute(new LiftOff());

exec.shutdown();

}

}

<spoiler text="Output:"> (Sample)

#0(9), #0(8), #1(9), #2(9), #3(9), #4(9), #0(7), #1(8), #2(8), #3(8), #4(8),

#0(6), #1(7), #2(7), #3(7), #4(7), #0(5), #1(6), #2(6), #3(6), #4(6), #0(4),

#1(5), #2(5), #3(5), #4(5), #0(3), #1(4), #2(4), #3(4), #4(4), #0(2), #1(3),

#2(3), #3(3), #4(3), #0(1), #1(2), #2(2), #3(2), #4(2), #0(Liftoff!), #1(1),

#2(1), #3(1), #4(1), #1(Liftoff!), #2(Liftoff!), #3(Liftoff!), #4(Liftoff!),

</spoiler> С FixedThreadPool дорогостоящая операция создания потоков выполняется только один раз, в самом начале, поэтому количество потоков остается фиксированным. Это способствует экономии времени, поскольку вам не приходится нести затраты, связанные с созданием потока, для каждой отдельной задачи. В системах, управляемых событиями, обеспечивается максимальная скорость выполнения обработчиков событий, так как они могут просто получить поток из пула. Перерасход ресурсов в такой схеме исключен, так какFixedThreadPool использует ограниченное количество объектов Thread.

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

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

//: concurrency/SingleThreadExecutor.java

import java.util.concurrent.*;

public class SingleThreadExecutor {

public static void main(String[] args) {

ExecutorService exec =

Executors.newSingleThreadExecutor();

for(int i = 0; i < 5; i++)

exec.execute(new LiftOff());

exec.shutdown();

}

}

<spoiler text="Output:">

#0(9), #0(8), #0(7), #0(6), #0(5), #0(4), #0(3), #0(2), #0(1), #0(Liftoff!),

#1(9), #1(8), #1(7), #1(6), #1(5), #1(4), #1(3), #1(2), #1(1), #1(Liftoff!),

#2(9), #2(8), #2(7), #2(6), #2(5), #2(4), #2(3), #2(2), #2(1), #2(Liftoff!),

#3(9), #3(8), #3(7), #3(6), #3(5), #3(4), #3(3), #3(2), #3(1), #3(Liftoff!),

#4(9), #4(8), #4(7), #4(6), #4(5), #4(4), #4(3), #4(2), #4(1), #4(Liftoff!),

</spoiler> Другой пример: допустим, имеется группа потоков, выполняющих операции с использованием файловой системы. Вы можете запустить эти задачи под управлением SingleThreadExecutor, чтобы в любой момент гарантированно вы­полнялось не более одной задачи. При таком подходе вам не придется возиться с синхронизацией доступа к общим ресурсам (без риска для целостности файловой системы). Возможно, в окончательной версии кода будет правильнее синхронизировать доступ к ресурсу (см. далее в этой главе), ноSingleThreadExecutor позволит быстро организовать координацию доступа при построении рабочего прототипа.

Возврат значений из задач

Интерфейс Runnable представляет отдельную задачу, которая выполняет некоторую работу, но не возвращает значения. Если вы хотите, чтобы задача возвращала значение, реализуйте интерфейс Callable вместо интерфейса Runnable. Пара­метризованный интерфейсCallable, появившийся в Java SE5, имеет параметр типа, представляющий возвращаемое значение метода call() (вместо run()), а для его вызова должен использоваться метод ExecutorService.submit(). Простой пример:

//: concurrency/CallableDemo.java

import java.util.concurrent.*;

import java.util.*;

class TaskWithResult implements Callable<String> {

private int id;

public TaskWithResult(int id) {

this.id = id;

}

public String call() {

return "result of TaskWithResult " + id;

}

}

public class CallableDemo {

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

ArrayList<Future<String>> results =

new ArrayList<Future<String>>();

for(int i = 0; i < 10; i++)

results.add(exec.submit(new TaskWithResult(i)));

for(Future<String> fs : results)

try {

// Вызов get() блокируется до завершения;:

System.out.println(fs.get());

} catch(InterruptedException e) {

System.out.println(e);

return;

} catch(ExecutionException e) {

System.out.println(e);

} finally {

exec.shutdown();

}

}

}

<spoiler text="Output:">

result of TaskWithResult 0

result of TaskWithResult 1

result of TaskWithResult 2

result of TaskWithResult 3

result of TaskWithResult 4

result of TaskWithResult 5

result of TaskWithResult 6

result of TaskWithResult 7

result of TaskWithResult 8

result of TaskWithResult 9

</spoiler> Метод submit() создает объект Future, параметризованный по типу результата, возвращаемому Callable. Вы можете обратиться к Future с запросом isDone(), чтобы узнать, завершена ли операция. После завершения задачи и появления ре­зультата производится его выборка методом get(). Если get() вызывается без предварительной проверки isDone(), вызов блокируется до появления результата. Также можно вызвать get() с интервалом тайм-аута.

Перегруженный метод Executors.callable() получает Runnable и выдает Callable. ExecutorService содержит методы для выполнения коллекций объектов Callable.

Ожидание

Другим способом управления вашими потоками является вызов метода sleep(), который переводит поток в состояние ожидания на заданное количество миллисекунд. Если в классе LiftOff заменить вызов yield() на вызов метода sleep(), будет получен следующий результат:

//: concurrency/SleepingTask.java

// Вызов sleep() для приостановки потока.

import java.util.concurrent.*;

public class SleepingTask extends LiftOff {

public void run() {

try {

while(countDown-- > 0) {

System.out.print(status());

// Старый стиль.

// Thread.sleep(l00);

// Стиль Java SE5/6:

TimeUnit.MILLISECONDS.sleep(100);

}

} catch(InterruptedException e) {

System.err.println("Interrupted");

}

}

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(new SleepingTask());

exec.shutdown();

}

}

<spoiler text="Output:">

#0(9), #1(9), #2(9), #3(9), #4(9), #0(8), #1(8), #2(8), #3(8), #4(8), #0(7),

#1(7), #2(7), #3(7), #4(7), #0(6), #1(6), #2(6), #3(6), #4(6), #0(5), #1(5),

#2(5), #3(5), #4(5), #0(4), #1(4), #2(4), #3(4), #4(4), #0(3), #1(3), #2(3),

#3(3), #4(3), #0(2), #1(2), #2(2), #3(2), #4(2), #0(1), #1(1), #2(1), #3(1),

#4(1), #0(Liftoff!), #1(Liftoff!), #2(Liftoff!), #3(Liftoff!), #4(Liftoff!),

</spoiler> Вызов метода sleep() может привести к исключению InterruptedException; перехват этого исключения продемонстрирован в run(). Поскольку исключения не распространяются по потокам обратно в main(), вы должны локально обработать любые исключения, возникающие внутри задачи.

В Java SE5 появилась новая версия sleep(), оформленная в виде метода класса TimeUnit; она продемонстрирована в приведенном примере. Она делает программу более наглядной, поскольку вы можете указать единицы измерения продолжительности задержки. КлассTimeUnit также может использоваться для выполнения преобразований, как будет показано далее в этой главе.

На некоторых платформах задачи выполняются в порядке «идеального распределения» — от 0 до 4, затем снова от 4 до 0. Это вполне логично, поскольку после каждой команды вывода задача переходит в состояние ожидания, что позволяет планировщику потоков переключиться на другой поток. Тем не менее такое поведение зависит от базовой реализации потокового механизма, поэтому полагаться на него нельзя. Если вам потребуется управлять порядком выполнения задач, используйте средства синхронизации (см. далее) или же вообще откажитесь от использования потоков и напишите собственные функции синхронизации, которые передают управление друг другу в нужном порядке.

Приоритет

Приоритет (priority) потока сообщает планировщику информацию об относительной важности потока. Хотя порядок обращения процессора к существующему набору потоков и не детерминирован, если существует несколько приос­тановленных потоков, одновременно ожидающих запуска, планировщик сначала запустит поток с большим приоритетом. Впрочем, это не значит, что потоки с младшими приоритетами не выполняются вовсе (то есть тупиковых ситуаций из-за приоритетов не возникает). Потоки с более низкими приоритетами просто запускаются чуть реже.

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

Следующий пример демонстрирует использование приоритетов. Приоритет существующего потока читается методом getPriority() и задается методом setPriority():

//: concurrency/SimplePriorities.java

// Использование приоритетов потоков.

import java.util.concurrent.*;

public class SimplePriorities implements Runnable {

private int countDown = 5;

private volatile double d; // Без оптимизации

private int priority;

public SimplePriorities(int priority) {

this.priority = priority;

}

public String toString() {

return Thread.currentThread() + ": " + countDown;

}

public void run() {

Thread.currentThread().setPriority(priority);

while(true) {

// Высокозатратная, прерываемая операция:

for(int i = 1; i < 100000; i++) {

d += (Math.PI + Math.E) / (double)i;

if(i % 1000 == 0)

Thread.yield();

}

System.out.println(this);

if(--countDown == 0) return;

}

}

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();

for(int i = 0; i < 5; i++)

exec.execute(

new SimplePriorities(Thread.MIN_PRIORITY));

exec.execute(

new SimplePriorities(Thread.MAX_PRIORITY));

exec.shutdown();

}

}

<spoiler text="Output:"> (70% match)

Thread[pool-1-thread-6,10,main]: 5

Thread[pool-1-thread-6,10,main]: 4

Thread[pool-1-thread-6,10,main]: 3

Thread[pool-1-thread-6,10,main]: 2

Thread[pool-1-thread-6,10,main]: 1

Thread[pool-1-thread-3,1,main]: 5

Thread[pool-1-thread-2,1,main]: 5

Thread[pool-1-thread-1,1,main]: 5

Thread[pool-1-thread-5,1,main]: 5

Thread[pool-1-thread-4,1,main]: 5

</spoiler> В этой версии метод toString() переопределяется и использует метод Thread. toString(), который выводит имя потока (его можно задать в конструкторе, но здесь имена автоматически генерируются в виде pool-1-thread-1, pool-1-thread-2 и т. д.), приоритет и группу, к которой принадлежит поток. Переопределенная версия toString() также выводит обратный отсчет, выполняемый задачей. Обратите внимание: для получения ссылки на объект Thread, управляющий задачей, внутри самой задачи, следует вызвать методThread.currentThread().

Мы видим, что приоритет последнего потока имеет наивысший уровень, а все остальные потоки находятся на низшем уровне. Учтите, что приоритет задается в начале выполнения run(); задавать его в конструкторе бессмысленно, потому что Executor в этот момент еще не начал выполнять задачу.

В метод run() были добавлены 100 000 достаточно затратных операций с плавающей запятой, включая суммирование и деление с числом двойной точности double. Переменная d была отмечена как volatile, чтобы компилятор не применял оптимизацию. Без этих вычислений вы не увидите эффекта установки различных приоритетов (попробуйте закомментировать цикл for с вычислениями). В процессе вычислений мы видим, что планировщик уделяет больше внимания потоку с приоритетом MAX_PRIORITY (по крайней мере, таково было поведение программы на машине под управлением Windows ХР). Несмотря даже на то, что вывод на консоль также является «дорогостоящей» операцией, с ним вы не увидите влияние уровней приоритетов, поскольку вывод на консоль не прерывается (иначе экран был бы заполнен несуразицей), в то время как математические вычисления, приведенные выше, прерывать допустимо. Вычисления выполняются достаточно долго, соответственно, механизм планирования потоков вмешивается в процесс и чередует потоки, проявляя при этом внимание к более приоритетным. Тем не менее для обеспечения переключения контекста в программе периодически выполняются команды yield().

В пакете JDK предусмотрено 10 уровней приоритетов, однако это не слишком хорошо согласуется с большинством операционных систем. К примеру, в Windows имеется 7 классов приоритетов, таким образом, их соотношение неочевидно (хотя в операционной системе Sun Solaris имеется 231 уровней). Переносимость обеспечивается только использованием универсальных констант МАХ_РRIORITY, NORM_PRIORITY и MIN_PRI0RITY.

Передача управления

Если вы знаете, что в текущей итерации run() сделано все необходимое, вы можете подсказать механизму планирования потоков, что процессором теперь может воспользоваться другой поток. Эта подсказка (не более чем рекомендация; нет никакой гарантии, что планировщик потоков «прислушается» к ней) воплощается в форме вызова метода yield(). Вызывая yield(), вы сообщаете системе, что в ней могут выполняться другие потоки того же приоритета.

В примере LiftOff метод yield() обеспечивает равномерное распределение вычислительных ресурсов между задачами LiftOff. Попробуйте закомментировать вызов Thread.yield() в Lift0ff.run() и проследите за различиями. И все же, в общем случае не стоит полагаться наyield() как на серьезное средство настройки вашего приложения.

Потоки-демоны

Демоном называется поток, предоставляющий некоторый сервис, работая в фоновом режиме во время выполнения программы, но при этом не является ее неотъемлемой частью. Таким образом, когда все потоки не-демоны заканчивают свою деятельность, программа завершается. И наоборот, если существуют работающие потоки не-демоны, программа продолжает выполнение. Существует, например, поток не-демон, выполняющий метод main().

//: concurrency/SimpleDaemons.java

// Потоки-демоны не препятствуют завершению работы программы.

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

public class SimpleDaemons implements Runnable {

public void run() {

try {

while(true) {

TimeUnit.MILLISECONDS.sleep(100);

print(Thread.currentThread() + " " + this);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

}

public static void main(String[] args) throws Exception {

for(int i = 0; i < 10; i++) {

Thread daemon = new Thread(new SimpleDaemons());

daemon.setDaemon(true); // Необходимо вызвать перед start()

daemon.start();

}

print("All daemons started");

TimeUnit.MILLISECONDS.sleep(175);

}

}

<spoiler text="Output:"> (Sample)

All daemons started

Thread[Thread-0,5,main] SimpleDaemons@530daa

Thread[Thread-1,5,main] SimpleDaemons@a62fc3

Thread[Thread-2,5,main] SimpleDaemons@89ae9e

Thread[Thread-3,5,main] SimpleDaemons@1270b73

Thread[Thread-4,5,main] SimpleDaemons@60aeb0

Thread[Thread-5,5,main] SimpleDaemons@16caf43

Thread[Thread-6,5,main] SimpleDaemons@66848c

Thread[Thread-7,5,main] SimpleDaemons@8813f2

Thread[Thread-8,5,main] SimpleDaemons@1d58aae

Thread[Thread-9,5,main] SimpleDaemons@83cc67

...

</spoiler> Чтобы назначить поток демоном, следует перед его запуском вызвать метод setDaemon().

После того как main() завершит свою работу, ничто не препятствует завершению программы, поскольку в процессе не работают другие потоки, кроме демонов. Чтобы результаты запуска всех потоков-демонов были более наглядными, поток main() на некоторое время погружается в «сон». Без этого вы увидели бы только часть результатов при создании демонов. (Поэкспериментируйте с вызовом sleep() для интервалов разной продолжительности.)

В примере SimpleDaemons.java используется явное создание объектов Thread для установки их «демонского» флага. Вы также можете настроить атрибуты (демон, приоритет, имя) потоков, созданных исполнителем; для этого следует написать пользовательскую реализациюThreadFactory:

//: net/mindview/util/DaemonThreadFactory.java

package net.mindview.util;

import java.util.concurrent.*;

public class DaemonThreadFactory implements ThreadFactory {

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

t.setDaemon(true);

return t;

}

}

Единственное отличие от обычной реализации ThreadFactory заключается в том, что в данном случае атрибут демона задается равным true. Теперь новый объект DaemonThreadFactory передается в аргументе Executors.newCachedThreadPool():

//: concurrency/DaemonFromFactory.java

// Использование ThreadFactory для создания демонов.

import java.util.concurrent.*;

import net.mindview.util.*;

import static net.mindview.util.Print.*;

public class DaemonFromFactory implements Runnable {

public void run() {

try {

while(true) {

TimeUnit.MILLISECONDS.sleep(100);

print(Thread.currentThread() + " " + this);

}

} catch(InterruptedException e) {

print("Interrupted");

}

}

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool(

new DaemonThreadFactory());

for(int i = 0; i < 10; i++)

exec.execute(new DaemonFromFactory());

print("All daemons started");

TimeUnit.MILLISECONDS.sleep(500); // Задержка

}

} /* (Execute to see output) *///:~

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

Сделаем еще один шаг — создадим вспомогательный класс DaemonThreadPoolExecutor:

//: net/mindview/util/DaemonThreadPoolExecutor.java

package net.mindview.util;

import java.util.concurrent.*;

public class DaemonThreadPoolExecutor

extends ThreadPoolExecutor {

public DaemonThreadPoolExecutor() {

super(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,

new SynchronousQueue<Runnable>(),

new DaemonThreadFactory());

}

}

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

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

//: concurrency/Daemons.java

// Потоки, порождаемые демонами, также являются демонами

// Daemon threads spawn other daemon threads.

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

class Daemon implements Runnable {

private Thread[] t = new Thread[10];

public void run() {

for(int i = 0; i < t.length; i++) {

t[i] = new Thread(new DaemonSpawn());

t[i].start();

printnb("DaemonSpawn " + i + " started, ");

}

for(int i = 0; i < t.length; i++)

printnb("t[" + i + "].isDaemon() = " +

t[i].isDaemon() + ", ");

while(true)

Thread.yield();

}

}

class DaemonSpawn implements Runnable {

public void run() {

while(true)

Thread.yield();

}

}

public class Daemons {

public static void main(String[] args) throws Exception {

Thread d = new Thread(new Daemon());

d.setDaemon(true);

d.start();

printnb("d.isDaemon() = " + d.isDaemon() + ", ");

// Даем потокам-демонам завершить процесс запуска:

TimeUnit.SECONDS.sleep(1);

}

}

<spoiler text="Output:"> (Sample)

d.isDaemon() = true, DaemonSpawn 0 started, DaemonSpawn 1 started, DaemonSpawn 2 started,

DaemonSpawn 3 started, DaemonSpawn 4 started, DaemonSpawn 5 started, DaemonSpawn 6 started,

DaemonSpawn 7 started, DaemonSpawn 8 started, DaemonSpawn 9 started, t[0].isDaemon() = true,

t[1].isDaemon() = true, t[2].isDaemon() = true, t[3].isDaemon() = true, t[4].isDaemon() = true,

t[5].isDaemon() = true, t[6].isDaemon() = true, t[7].isDaemon() = true, t[8].isDaemon() = true,

t[9].isDaemon() = true,

</spoiler> Поток Daemon переводится в режим демона, а затем порождает группу новых потоков, которые явно не назначаются демонами, но при этом все равно оказываются ими. Затем Daemon входит в бесконечный цикл, на каждом шаге которого вызывается метод yield(), передающий управление другими процессам.

Учтите, что потоки-демоны завершают свои методы run() без выполнения секций finally:

//: concurrency/DaemonsDontRunFinally.java

// Потоки-демоны не выполняют секцию finally.

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

class ADaemon implements Runnable {

public void run() {

try {

print("Starting ADaemon");

TimeUnit.SECONDS.sleep(1);

} catch(InterruptedException e) {

print("Exiting via InterruptedException");

} finally {

print("This should always run?");

}

}

}

public class DaemonsDontRunFinally {

public static void main(String[] args) throws Exception {

Thread t = new Thread(new ADaemon());

t.setDaemon(true);

t.start();

}

}

<spoiler text="Output:">

Starting ADaemon

</spoiler> Запуск программы наглядно показывает, что секция finally не выполняется. С другой стороны, если закомментировать вызов setDaemon(), вы увидите, что секция finally была выполнена.

Такое поведение верно, даже если из предыдущих описаний finally у вас сложилось обратное впечатление. Демоны завершаются «внезапно», при завершении последнего не-демона. Таким образом, сразу же при выходе из main() JVM немедленно прерывает работу всех демонов, не соблюдая никакие формальности. Невозможность корректного завершения демонов ограничивает возможности их применения. Обычно объекты Executor оказываются более удачным решением, потому что все задачи, находящиеся под управлением Executor, могут быть завершены одновременно.

Варианты кодирования

Во всех предшествующих примерах все классы задач реализовали интерфейс Runnable. В очень простых случаях можно использовать альтернативное решение с прямым наследованием от Thread:

//: concurrency/SimpleThread.java

// Прямое наследование от класса Thread..

public class SimpleThread extends Thread {

private int countDown = 5;

private static int threadCount = 0;

public SimpleThread() {

// Store the thread name:

super(Integer.toString(++threadCount));

start();

}

public String toString() {

return "#" + getName() + "(" + countDown + "), ";

}

public void run() {

while(true) {

System.out.print(this);

if(--countDown == 0)

return;

}

}

public static void main(String[] args) {

for(int i = 0; i < 5; i++)

new SimpleThread();

}

}

<spoiler text="Output:">

#1(5), #1(4), #1(3), #1(2), #1(1), #2(5), #2(4), #2(3), #2(2), #2(1), #3(5),

#3(4), #3(3), #3(2), #3(1), #4(5), #4(4), #4(3), #4(2), #4(1), #5(5), #5(4),

#5(3), #5(2), #5(1),

</spoiler> Чтобы задать объектам Thread имена, вы вызываете соответствующий конструктор Thread. Имя читается в методе toString() при помощи getName().

Также иногда встречается идиома самоуправляемой реализации Runnable:

//: concurrency/SelfManaged.java

// Реализация Runnable. содержащая собственный объект Thread.

public class SelfManaged implements Runnable {

private int countDown = 5;

private Thread t = new Thread(this);

public SelfManaged() { t.start(); }

public String toString() {

return Thread.currentThread().getName() +

"(" + countDown + "), ";

}

public void run() {

while(true) {

System.out.print(this);

if(--countDown == 0)

return;

}

}

public static void main(String[] args) {

for(int i = 0; i < 5; i++)

new SelfManaged();

}

}

<spoiler text="Output:">

Thread-0(5), Thread-0(4), Thread-0(3), Thread-0(2), Thread-0(1), Thread-1(5), Thread-1(4),

Thread-1(3), Thread-1(2), Thread-1(1), Thread-2(5), Thread-2(4), Thread-2(3), Thread-2(2),

Thread-2(1), Thread-3(5), Thread-3(4), Thread-3(3), Thread-3(2), Thread-3(1), Thread-4(5),

Thread-4(4), Thread-4(3), Thread-4(2), Thread-4(1),

</spoiler> В целом происходящее не так уж сильно отличается от наследования от Thread, разве что синтаксис получается чуть более громоздким. Однако реализация интерфейса позволяет наследовать от другого класса, тогда как в варианте с Thread это невозможно.

Обратите внимание на вызов start() в конструкторе. Приведенный пример очень прост, поэтому, скорее всего, в нем такое решение безопасно, но вы должны знать, что запуск потоков в конструкторе может создать изрядные проблемы — до завершения конструктора может быть запущена на выполнение другая задача, которая обратится к объекту в нестабильном состоянии. Это еще одна причина, по которой использование Executor предпочтительнее явного создания объектов Thread.

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

//: concurrency/ThreadVariations.java

// Создание потоков с использованием внутренних классов..

import java.util.concurrent.*;

import static net.mindview.util.Print.*;

// Using a named inner class:

class InnerThread1 {

private int countDown = 5;

private Inner inner;

private class Inner extends Thread {

Inner(String name) {

super(name);

start();

}

public void run() {

try {

while(true) {

print(this);

if(--countDown == 0) return;

sleep(10);

}

} catch(InterruptedException e) {

print("interrupted");

}

}

public String toString() {

return getName() + ": " + countDown;

}

}

public InnerThread1(String name) {

inner = new Inner(name);

}

}

// Используем безымянный внутренний класс::

class InnerThread2 {

private int countDown = 5;

private Thread t;

public InnerThread2(String name) {

t = new Thread(name) {

public void run() {

try {

while(true) {

print(this);

if(--countDown == 0) return;

sleep(10);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

}

public String toString() {

return getName() + ": " + countDown;

}

};

t.start();

}

}

// Используем именованную реализацию Runnable:

class InnerRunnable1 {

private int countDown = 5;

private Inner inner;

private class Inner implements Runnable {

Thread t;

Inner(String name) {

t = new Thread(this, name);

t.start();

}

public void run() {

try {

while(true) {

print(this);

if(--countDown == 0) return;

TimeUnit.MILLISECONDS.sleep(10);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

}

public String toString() {

return t.getName() + ": " + countDown;

}

}

public InnerRunnable1(String name) {

inner = new Inner(name);

}

}

// Используем анонимную реализацию Runnable:

class InnerRunnable2 {

private int countDown = 5;

private Thread t;

public InnerRunnable2(String name) {

t = new Thread(new Runnable() {

public void run() {

try {

while(true) {

print(this);

if(--countDown == 0) return;

TimeUnit.MILLISECONDS.sleep(10);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

}

public String toString() {

return Thread.currentThread().getName() +

": " + countDown;

}

}, name);

t.start();

}

}

// Отдельный метод для выполнения кода в потоке:

class ThreadMethod {

private int countDown = 5;

private Thread t;

private String name;

public ThreadMethod(String name) { this.name = name; }

public void runTask() {

if(t == null) {

t = new Thread(name) {

public void run() {

try {

while(true) {

print(this);

if(--countDown == 0) return;

sleep(10);

}

} catch(InterruptedException e) {

print("sleep() interrupted");

}

}

public String toString() {

return getName() + ": " + countDown;

}

};

t.start();

}

}

}

public class ThreadVariations {

public static void main(String[] args) {

new InnerThread1("InnerThread1");

new InnerThread2("InnerThread2");

new InnerRunnable1("InnerRunnable1");

new InnerRunnable2("InnerRunnable2");

new ThreadMethod("ThreadMethod").runTask();

}

} /* (Execute to see output) *///:~

InnerThread1 определяет именованный внутренний класс, производный от Thread, и создает экземпляр этого класса в конструкторе. Поступать так стоит в том случае, когда у внутреннего класса есть особые возможности (новые методы), которые могут понадобиться в других методах. Однако в большинстве случаев причина создания потока — использование функциональности класса Thread, поэтому в именованном внутреннем классе особой нужды нет. InnerThread2 показывает другое решение. В конструкторе создается безымянный внутренний субкласс Thread, преобразуемый восходящим преобразованием к ссылке на Thread.t. Если другим методам класса понадобится обратиться к t, они смогут сделать это через интерфейс Thread, и им не нужно будет знать точный тип объекта.

Третий и четвертый классы примера повторяют первые два, только вместо класса Thread они используют интерфейс Runnable.

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

Присоединение к потоку

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

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

Вызов join() может быть прерван вызовом метода interrupt() для потока-инициатора, поэтому потребуется блок try-catch.

Все эти операции продемонстрированы в следующем примере:

//: concurrency/Joining.java

// Демонстрация join().

import static net.mindview.util.Print.*;

class Sleeper extends Thread {

private int duration;

public Sleeper(String name, int sleepTime) {

super(name);

duration = sleepTime;

start();

}

public void run() {

try {

sleep(duration);

} catch(InterruptedException e) {

print(getName() + " was interrupted. " +

"isInterrupted(): " + isInterrupted());

return;

}

print(getName() + " has awakened");

}

}

class Joiner extends Thread {

private Sleeper sleeper;

public Joiner(String name, Sleeper sleeper) {

super(name);

this.sleeper = sleeper;

start();

}

public void run() {

try {

sleeper.join();

} catch(InterruptedException e) {

print("Interrupted");

}

print(getName() + " join completed");

}

}

public class Joining {

public static void main(String[] args) {

Sleeper

sleepy = new Sleeper("Sleepy", 1500),

grumpy = new Sleeper("Grumpy", 1500);

Joiner

dopey = new Joiner("Dopey", sleepy),

doc = new Joiner("Doc", grumpy);

grumpy.interrupt();

}

}

<spoiler text="Output:">

Grumpy was interrupted. isInterrupted(): false

Doc join completed

Sleepy has awakened

Dopey join completed

</spoiler> Класс Sleeper — это тип потока, который приостанавливается на время, указанное в его конструкторе. В методе run() вызов метода sleep() может закончиться по истечении времени задержки, но может и прерваться. В секции catch выводится сообщение о прерывании, вместе со значением, возвращаемым методом isInterrupted(). Когда другой поток вызывает interrupt() для данного потока, устанавливается флаг, показывающий, что поток был прерван. Однако этот флаг сбрасывается при обработке исключения, поэтому внутри секции catch результатом всегда будет false. Флаг используется в других ситуациях, где поток может исследовать свое прерванное состояние в стороне от исключения.

Joiner — поток, который ожидает пробуждения потока Sleeper, вызывая для последнего метод join(). В методе main() каждому объекту Joiner сопоставляется Sleeper, и вы можете видеть в результатах работы программы, что, если Sleeper был прерван или завершился нормально, Joiner прекращает работу вместе с потоком Sleeper.

Совместное использование ресурсов

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

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