При разработке приложений, использующих параллелизм для достижения поставленных целей, вы можете столкнуться с ситуациями, в которых различные потоки могут блокировать друг друга. Если в этой ситуации приложение работает медленней, чем ожидалось, мы бы сказали, что оно отрабатывает по времени не так, как предполагалось. В данном разделе мы поближе познакомимся с проблемами, которые могут угрожать живучести многонитевого приложения.
Взаимная блокировка
Термин взаимоблокировка хорошо известен разработчикам ПО и даже большинство обычных пользователей используют его время от времени, хотя и не всегда в правильном смысле. Строго говоря, этот термин означает, что каждая из двух (или больше) потоков ждут от другой потоки, чтобы она освободила заблокированный ею ресурс, в то время как первая сам заблокировала ресурс, доступа к которому ждёт вторая:
Thread 1: locks resource A, waits for resource B
Thread 2: locks resource B, waits for resource A
Для лучшего понимания проблемы взглянем на следующий код:
public class Deadlock implements Runnable {
private static final Object resource1 = new Object();
private static final Object resource2 = new Object();
private final Random random = new Random(System.currentTimeMillis());
public static void main(String[] args) {
Thread myThread1 = new Thread(new Deadlock(), "thread-1");
Thread myThread2 = new Thread(new Deadlock(), "thread-2");
myThread1.start();
myThread2.start();
}
public void run() {
for (int i = 0; i < 10000; i++) {
boolean b = random.nextBoolean();
if (b) {
System.out.println("[" + Thread.currentThread().getName() + "] Trying to lock resource 1.");
synchronized (resource1) {
System.out.println("[" + Thread.currentThread().getName() + "] Locked resource 1.");
System.out.println("[" + Thread.currentThread().getName() + "] Trying to lock resource 2.");
synchronized (resource2) {
System.out.println("[" + Thread.currentThread().getName() + "] Locked resource 2.");
}
}
} else {
System.out.println("[" + Thread.currentThread().getName() + "] Trying to lock resource 2.");
synchronized (resource2) {
System.out.println("[" + Thread.currentThread().getName() + "] Locked resource 2.");
System.out.println("[" + Thread.currentThread().getName() + "] Trying to lock resource 1.");
synchronized (resource1) {
System.out.println("[" + Thread.currentThread().getName() + "] Locked resource 1.");
}
}
}
}
}
}
Как можно видеть из приведённого кода, две потоки стартуют и пытаются заблокировать два статических ресурса. Но для взаимоблокировки нам требуется различная последовательность для обеих потоков, поэтому мы используем экземпляр объекта Random, чтобы выбрать какой ресурс поток хочет заблокировать первым. Если логическая переменная b имеет значение true, то первым блокируется resource1, а после поток пытается получить блокировку для resource2. Если b - false, тогда поток блокирует resource2, а после пытается захватить resource1. Этой программе не требуется долго выполняться для достижения первой взаимоблокировки, т.е. программа повиснет навсегда, если мы не прервем её:
[thread-1] Trying to lock resource 1.
[thread-1] Locked resource 1.
[thread-1] Trying to lock resource 2.
[thread-1] Locked resource 2.
[thread-2] Trying to lock resource 1.
[thread-2] Locked resource 1.
[thread-1] Trying to lock resource 2.
[thread-1] Locked resource 2.
[thread-2] Trying to lock resource 2.
[thread-1] Trying to lock resource 1.
В данном запуске thread-1 завладел блокировкой resource2 и ожидает блокировки resource1, в то время как thread-2 заблокировал resource1 и ожидает resource2.
Если бы мы задали значение логической переменной b в приведенном выше коде равным истине, то не смогли бы наблюдать никакой взаимоблокировки, потому что последовательность, в которой thread-1 и thread-2 запрашивают блокировки, всегда была бы одной и той же. В этой ситуации одна из двух потоков получила бы блокировку первой и затем запрашивала бы вторую, которая по-прежнему доступна, поскольку другая поток ожидает первой блокировки.
В общем, можно выделить следующие необходимые условия возникновения взаимоблокировки:
Совместное выполнение: Существует ресурс, который может быть доступен только одной потоки в произвольный момент времени.
Удержание ресурса: Во время захвата одного ресурса, поток пытается заполучить ещё одну блокировку какого-то уникального ресурса.
Отсутствие приоритетного прерывания обслуживания: Отсутствует механизм, освобождающий ресурс, если одна поток удерживает блокировку определённый промежуток времени.
Круговое ожидание: Во время исполнения возникает совокупность потоков, в которой две (или более) потоков ждут друг от друга освобождения ресурса, который был заблокирован.
Хотя список условий и выглядит длинным, нередко хорошо отлаженные многонитевые приложения имеют проблемы взаимоблокировки. Но вы можете предотвратить их, если сумеете снять одно из условий, приведённых выше:
Совместное выполнение: это условие зачастую не может быть снято, когда ресурс должен использоваться только кем-то одним. Но это не обязательно должно стать причиной. При использовании DBMS систем возможным решением вместо использования пессимистичной блокировки по некоторой строке таблицы, которая должна быть обновлена, можно использовать технику, называемую Оптимистичной Блокировкой.
Способ избежать удержания ресурса во время ожидания другого эксклюзивного ресурса заключается в том, чтобы блокировать все необходимые ресурсы в начале алгоритма и освобождать тоже все, если невозможно их заблокировать разом. Конечно, это не всегда возможно, может быть ресурсы, требующие блокировки заранее неизвестны или такой подход просто приведёт к бесполезной трате ресурсов.
Если блокировка не может быть получена немедленно, способом обхода возможной взаимоблокировки является введений тайм-аута. Например, класс ReentrantLock из SDK обеспечивает возможность задания срока действия для блокировки.
Как мы увидели из приведённого выше примера, взаимоблокировка не возникает, если последовательность запросов не отличается у различных потоков. Это легко проконтролировать, если вы можете поместить весь блокирующий код в один метод, через который должны пройти все потоки.
В более продвинутых приложениях вы даже можете задумать реализацию системы выявления взаимоблокировок. Здесь вам понадобится реализовать некоторое подобие мониторинга потоков, в которой каждая поток сообщает об успешном получении права блокировки и своей попытке получить блокировку. Если потоки и блокировки смоделированы как ориентированный граф, вы можете обнаружить, когда две различных потоки удерживают ресурсы, пытаясь одновременно получить доступ к другим заблокированным ресурсам. Если затем вы сможете заставить блокирующие потоки освободить требуемые ресурсы, то сможете разрешить ситуацию взаимоблокировки автоматически.
Голодание
Планировщик решает, какую из потоков, находящихся в состоянии RUNNABLE, он должен выполнить следующей. Решение основывается на приоритете потоки; поэтому потоки с меньшим приоритетом получают меньше процессорного времени, по сравнению с теми, у которых приоритет выше. То, что выглядит разумным решением, также может стать причиной проблем при злоупотреблении. Если большую часть времени исполняются потоки с высоким приоритетом, то низкоприоритетные потоки как будто начинают "голодать", поскольку не получают достаточно времени для того, чтобы выполнить свою работу должным образом. Поэтому рекомендуется задавать приоритет потоки только тогда, когда для этого есть веские причины.
Неочевидный пример голодания потоки даёт, например, метод finalize(). Он предоставляет в языке Java возможность выполнить код перед тем, как объект будет удалён сборщиком мусора. Но если вы взглянете на приоритет финализирующей потоки, то заметите, что она запускается не с наивысшим приоритетом. Следовательно, возникают предпосылки для нитевого голодания, когда методы finalize() вашего объекта тратят слишком много времени по сравнению с остальным кодом.
Другая проблема со временем исполнения возникает от того, что не определено, в каком порядке потоки проходят блок synchronized. Когда много параллельных потоков проходят некоторый код, который оформлен в блок synchronized, может случиться так, что одним потокам придётся ждать дольше, чем другим, прежде чем войти в блок. В теории они могут никогда туда не попасть.
Решение данной проблемы - так называемая "справедливая" блокировка. Справедливые блокировки учитывают время ожидания потоков, когда определяют, кого пропустить следующим. Пример реализации справедливой блокировке есть в Java SDK: java.util.concurrent.locks.ReentrantLock. Если используется конструктор с логическим флагом, установленным в значение true, то ReentrantLock даёт доступ потоки, которая ждёт дольше остальных. Это гарантирует отсутствие голода но, в тоже время, приводит к проблеме игнорирования приоритетов. Из-за этого процессы с меньшим приоритетом, которые часто ожидают на этом барьере, могут выполняться чаще. Наконец, что немаловажно, класс ReentrantLock может рассматривать только потоки, которые ожидают блокировки, т.е. потоки, которые запускались достаточно часто и достигли барьера. Если приоритет потоки слишком низок, то для неё это не будет происходить часто, и поэтому высокоприоритетные потоки по-прежнему будут проходить блокировку чаще.
Мониторы объектов совместно с wait() и notify()
В многонитевых вычислениях обычной ситуацией является наличие некоторых рабочих потоков, которые ждут, что их производитель создаст для них какую-нибудь работу. Но, как мы узнали, активное ожидание в цикле с проверкой некоторого значения не есть хороший вариант с точки зрения процессорного времени. Использование в этой ситуации метода Thread.sleep() также не особо подходит, если мы хотим начать нашу работу немедленно после поступления.
Для этого язык программирования Java обладает другой структурой, которая может быть использована в данной схеме: wait() и notify(). Метод wait(), наследуемый всеми объектами от класса java.lang.Object, может быть использован для приостановки выполнения текущей потоки и ожидания до тех пор, пока другая поток не разбудит нас, используя метод notify(). Для того, чтобы работать правильно, поток, вызывающая метод wait(), должна удерживать блокировку, которую она предварительно получила, используя ключевое слово synchronized. При вызове wait() блокировка освобождается и поток ожидает, пока другая поток, которая теперь завладела блокировкой, не вызовет notify() для того же экземпляра объекта.
В многонитевом приложении естественно может быть более одной потоки, ожидающей уведомления на каком-то объекте. Поэтому существует два различных метода для побудки потоков: notify() и notifyAll(). В то время как первый метод будит одну из ожидающих потоков, метод notifyAll() пробуждает их все. Но знайте, что, как и в случае ключевого слова synchronized, отсутствует правило, определяющее, какая поток будет разбужена следующей при вызове notify(). В простом примере с производителем и потребителем это не имеет значения, поскольку нам не важно, какая именно поток проснулась.
Следующий код показывает как механизм wait() и notify() может быть использован для организации ожидания потоками-потребителями новой работы, которая добавляется в очередь нитью-производителем:
package a2;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
public class ConsumerProducer {
private static final Queue queue = new ConcurrentLinkedQueue();
private static final long startMillis = System.currentTimeMillis();
public static class Consumer implements Runnable {
public void run() {
while (System.currentTimeMillis() < (startMillis + 10000)) {
synchronized (queue) {
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if (!queue.isEmpty()) {
Integer integer = queue.poll();
System.out.println("[" + Thread.currentThread().getName() + "]: " + integer);
}
}
}
}
public static class Producer implements Runnable {
public void run() {
int i = 0;
while (System.currentTimeMillis() < (startMillis + 10000)) {
queue.add(i++);
synchronized (queue) {
queue.notify();
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
synchronized (queue) {
queue.notifyAll();
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread[] consumerThreads = new Thread[5];
for (int i = 0; i < consumerThreads.length; i++) {
consumerThreads[i] = new Thread(new Consumer(), "consumer-" + i);
consumerThreads[i].start();
}
Thread producerThread = new Thread(new Producer(), "producer");
producerThread.start();
for (int i = 0; i < consumerThreads.length; i++) {
consumerThreads[i].join();
}
producerThread.join();
}
}
Метод main() запускает пять потоков-потребителей и одну поток-производителя, а затем ждёт окончания их работы. После поток-производитель добавляет новое значение в очередь и уведомляет все ожидающие потоки о том, что что-то произошло. Потребители получают блокировку очереди (прим., один произвольный потребитель) и затем засыпают, чтобы быть поднятыми позже, когда очередь снова заполнится. Когда производитель заканчивает свою работу, то он уведомляет всех потребителей, чтобы разбудить. Если бы мы не сделали последний шаг, то потоки-потребители вечно бы ждали следующего уведомления, поскольку мы не задали тайм-аут для ожидания. Вместо этого мы можем использовать метод wait(long timeout), чтобы быть разбуженными, по крайней мере, по прошествии некоторого времени.
Вложенные блоки synchronized совместно с wait() и notify()
Как было сказано в предыдущем разделе, вызов wait() для монитора объекта лишь снимает блокировку по этому монитору. Другие блокировки, которые удерживались той же нитью, не освобождаются. Как это легко понять, в повседневной работе может случиться так, что поток, вызывающая wait() удерживает блокировки дальше. Если другие потоки также ожидают эти блокировки, то может возникнуть ситуация взаимоблокировки. Давайте посмотрим на блокировку в следующем примере:
public class SynchronizedAndWait {
private static final Queue queue = new ConcurrentLinkedQueue();
public synchronized Integer getNextInt() {
Integer retVal = null;
while (retVal == null) {
synchronized (queue) {
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
retVal = queue.poll();
}
}
return retVal;
}
public synchronized void putInt(Integer value) {
synchronized (queue) {
queue.add(value);
queue.notify();
}
}
public static void main(String[] args) throws InterruptedException {
final SynchronizedAndWait queue = new SynchronizedAndWait();
Thread thread1 = new Thread(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
queue.putInt(i);
}
}
});
Thread thread2 = new Thread(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
Integer nextInt = queue.getNextInt();
System.out.println("Next int: " + nextInt);
}
}
});
thread1.start();
thread2.start();
thread1.join();
thread2.join();
}
}
Как мы узнали раньше, добавление synchronized в сигнатуру метода равносильно созданию блока synchronized(this){}. В приведённом выше примере мы случайно добавили ключевое слово synchronized в метод, а после синхронизировали очередь по монитору объекта queue, чтобы отправить данную поток в сон на время ожидания следующего значения из queue. Затем, текущая поток освобождает блокировку по queue, но не блокировку по this. Метод putInt() уведомляет спящую поток, что новое значение было добавлено. Но случайно мы добавили ключевое слово synchronized и в этот метод. Теперь, когда вторая поток заснула, она по прежнему удерживает блокировку. Поэтому первая поток не может войти в метод putInt() пока эта блокировка удерживается второй нитью. В результате имеем ситуацию взаимоблокировки и зависшую программу. Если вы выполните приведённый выше код, это произойдёт сразу после начала работы программы.
В повседневной жизни такая ситуация может не быть столь очевидной. Блокировки, удерживаемые нитью, могут зависеть от параметров и условий, возникающих во время работы, и блок synchronized, вызывающий проблему, может не быть так близок в коде к месту, где мы поместили вызов wait(). Это делает проблематичным поиск таких проблем, особенно если они могут возникать через некоторое время или при высокой нагрузке.
Условия в блоках synchronized
Часто вам требуется проверить выполнение некоторого условия, прежде чем произвести какое-то действие с синхронизированным объектом. Когда у вас есть, например, очередь, вы хотите дождаться её заполнения. Следовательно, вы можете написать метод, проверяющий заполненность очереди. Если она ещё пуста, то вы отправляете текущую поток в сон до тех пор, пока она не будет разбужена:
public Integer getNextInt() {
Integer retVal = null;
synchronized (queue) {
try {
while (queue.isEmpty()) {
queue.wait();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
synchronized (queue) {
retVal = queue.poll();
if (retVal == null) {
System.err.println("retVal is null");
throw new IllegalStateException();
}
}
return retVal;
}
Приведенный выше код синхронизируется по queue прежде чем вызвать wait() и, затем, ожидает в цикле while, пока в queue не появится по крайней мере один элемент. Второй блок synchronized опять использует queue как монитор объекта. Он вызывает метод poll() очереди, чтобы получить значение. В демонстрационных целях выбрасывается исключение IllegalStateException, когда poll возвращает null. Это происходит, когда в queue нет элементов для извлечения.
Когда вы запустите этот пример, то увидите, что IllegalStateException выбрасывается очень часто. Хоть мы и корректно синхронизировались по монитору queue, исключение было выброшено. Причина в том, что у нас есть два различных блока synchronized. Представьте, что у нас есть две потоки, которые прибыли к первому блоку synchronized. Первая поток вошла в блок и провалилась в сон, потому что queue пуста. То же самое истинно и для второй потоки. Теперь, когда обе потоки проснулись (благодаря вызову notifyAll(), вызванному другой нитью для монитора), они обе увидели значение(элемент) в очереди, добавленный производителем. Затем обе прибыли ко второму барьеру. Здесь первая поток вошла и извлекла значение из очереди. Когда вторая поток входит, queue уже пуста. Поэтому, в качестве значения, возвращаемого из queue, она получает null и выбрасывает исключение.
Для предотвращения подобных ситуаций вам необходимо выполнять все операции, зависящие от состояния монитора, в одном и том же блоке synchronized:
Здесь мы выполняем метод poll() в том же самом блоке synchronized, что и метод isEmpty(). Благодаря блоку synchronized мы уверены, что только одна поток выполняет метод для этого монитора в заданный момент времени. Поэтому никакая другая поток не может удалить элементы из queue между вызовами isEmpty() и poll().
Проектирование для многонитевости
Как мы уже увидели в предыдущем разделе, реализация многонитевого приложения часто оказывается чем-то более сложным, чем казалось на первый взгляд. Поэтому важно держать в голове чёткую схему перед началом проекта
Неизменный объект
Одним из правил проектирования, являющихся очень важным в данном контексте, является Неизменяемость. Если вы распределяете экземпляры объектов, например, между различными потоками, то должны уделить внимание тому, чтобы две потоки не изменяли один и тот же объект одновременно. В подобных ситуациях легко управляться с немодифицируемыми объектами, поскольку вы не можете их изменить. Вы всегда должны создавать новый экземпляр, когда хотите изменить данные. Базовый класс java.lang.String - пример неизменяемого класса. Вы получаете новый экземпляр всякий раз, когда хотите изменить строку:
Хоть создание объекта и не проходит бесплатно, цена часто завышена. Вы всегда должны оценивать, что предпочтительнее, простой дизайн с неизменяемыми объектами или отказ от использования неизменяемых объектов с риском получить ошибки параллелизма, которые могут проявиться в проекте позднее.
Далее приведён список правил, которые необходимо применять, чтобы сделать класс неизменяемым:
Все поля должны быть final и private.
Должны отсутствовать сеттеры.
Класс сам по себе должны быть объявлен как final для того, чтобы предотвратить нарушение принципа неизменяемости подклассом.
Если тип поля не является примитивным:
Не должно быть геттера, который передавал бы ссылку в открытом виде напрямую вызывающей стороне.
Не изменяйте ссылочные объекты (или хотя бы делайте эти изменения невидимыми для клиентов объекта).
Экземпляры следующего класса представляют собой сообщения с темой, телом и несколькими парами ключ/значение:
public final class ImmutableMessage {
private final String subject;
private final String message;
private final Map header;
public ImmutableMessage(Map header, String subject, String message) {
this.header = new HashMap(header);
this.subject = subject;
this.message = message;
}
public String getSubject() {
return subject;
}
public String getMessage() {
return message;
}
public String getHeader(String key) {
return this.header.get(key);
}
public Map getHeaders() {
return Collections.unmodifiableMap(this.header);
}
}
Класс - неизменяемый, поскольку все его поля определены как final и private. В классе отсутствуют методы, которые могли бы изменить состояние экземпляра после создания. Возврат ссылок на тему и сообщение безопасен, т.к. класс String сам по себе является неизменяемым. Тот, кто вызвал метод и получил ссылку, например, на сообщение не может изменить его непосредственно. А с полем Map, хранящим заголовки, необходимо быть осторожным. Возврат ссылки на объект Map позволил бы вызывающему изменить его содержимое. Следовательно, нам следует возвращать неизменяемый Map, полученный через вызов метода Collections.unmodifiableMap(). Он вернёт "изображение" Map, позволяющее вызывающим читать значения (которые опять же являются строками), но не допускающее изменений. Исключение UnsupportedOperationException будет выброшено при попытке изменить экземпляр объекта Map.
В данном примере также безопасно возвращать определённые ключи, как это сделано в методе getHeader(String key), поскольку возвращаемый объект типа String вновь неизменяем. Если объект Map будет содержать объекты, которые сами по себе не являются неизменными, данная операция не будет потоке-безопасной.
Разработка API
При разработке публичных методов класса, т.е. его API, вы можете также приспособить его для использования в многонитевой среде. У вас могут быть методы, которые не должны выполняться, когда объект находится в каком-то определённом состоянии. Одно из простых решений - завести приватный флаг, обозначающий, в каком состоянии мы находимся, и выбрасывать, например, исключение IllegalStateException, когда определённый метод не должен быть вызван:
public class Job {
private boolean running = false;
private final String filename;
public Job(String filename) {
this.filename = filename;
}
public synchronized void start() {
if(running) {
throw new IllegalStateException("...");
}
...
}
public synchronized List getResults() {
if(!running) {
throw new IllegalStateException("...");
}
...
}
}
Приведённый шаблон часто называют “balking pattern”, поскольку метод уклоняется от выполнения, когда запущен в неправильном состоянии. Но вы можете реализовать тот же функционал, используя статический фабричный метод (шаблон fabric method) без проверки состояния объекта в каждом методе:
public class Job {
private final String filename;
private Job(String filename) {
this.filename = filename;
}
public static Job createAndStart(String filename) {
Job job = new Job(filename);
job.start();
return job;
}
private void start() {
...
}
public synchronized List getResults() {
...
}
}
Статический фабричный метод создаёт новый экземпляр объекта Job, используя частный конструктор, и сразу вызывает start() для экземпляра. Возвращаемый по ссылке объект уже находится в подходящем для работы состоянии, поэтому метод getResults() требует только синхронизации, но не требует проверки состояния объекта.
Локальное хранилище потоки
До сих пор мы видели, что потоки разделяют одну и ту же память. Это хорошо с позиции производительности. Если бы мы использовали раздельные процессы для того, чтобы выполнять код параллельно, то нам бы потребовались более тяжеловесные методы обмена данными, такие как удалённые вызовы процедур или синхронизация на уровне файловой системы или сети. Но разделение памяти между различными потоками тоже тяжело поддерживать, если синхронизация не реализована должным образом.
Специализированная память, используемая только нашей собственной нитью и никакой другой, обеспечивается в Java посредством класса java.lang.ThreadLocal:
private static final ThreadLocal myThreadLocalInteger = new ThreadLocal();
Тип данных, которые должны быть сохранены в ThreadLocal, задаётся шаблонным параметром T. В примере мы использовали просто Integer, но также могли бы использовать любой другой тип данных. Следующий код демонстрирует использование ThreadLocal:
public class ThreadLocalExample implements Runnable {
private static final ThreadLocal threadLocal = new ThreadLocal();
private final int value;
public ThreadLocalExample(int value) {
this.value = value;
}
@Override
public void run() {
threadLocal.set(value);
Integer integer = threadLocal.get();
System.out.println("[" + Thread.currentThread().getName() + "]: " + integer);
}
public static void main(String[] args) throws InterruptedException {
Thread threads[] = new Thread[5];
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(new ThreadLocalExample(i), "thread-" + i);
threads[i].start();
}
for (int i = 0; i < threads.length; i++) {
threads[i].join();
}
}
}
Вы удивитесь, но каждая поток выводит в точности то значение, которое получила из конструктора, не смотря на то, что переменная threadLocal объявлена как static. Внутренняя реализация класса ThreadLocal гарантирует, что каждый раз при вызове set() данное значение сохраняется в области памяти, доступной только данной потоки. Поэтому при последующем вызове get() вы получаете значение, которое установили ранее, не смотря на то, что другие потоки тем временем могли вызывать set().
Серверы приложений в мире Java EE активно используют возможности ThreadLocal когда у вас есть много параллельных потоков, но каждая поток имеет, например, свой собственный контекст транзакции или безопасности. Если вы не хотите передавать эти объекты при каждом вызове метода, то просто сохраняете их в собственной памяти потоки и позже получаете к ним доступ, когда потребуется.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ