Представляем вашему вниманию первую статью из серии о синхронизации потоков в Java, в которой мы рассмотрим основы: состояния гонки, объекты блокировки, объекты условий, а также методы await, signal и signalAll.

[Это первая статья цикла о синхронизации потоков из трех частей на основе книги "Java. Библиотека профессионала. Том 1. Основы", 12-е издание, Кей С. Хорстманн (Cay S. Horstmann). -Ред.]

Практически в каждом реальном многопоточном приложении возможны сценарии, когда двум или более потокам требуется доступ к одним и тем же данным в одно и то же время. Но что произойдет, если два потока одновременно получат доступ к одному и тому же объекту, и каждый из них вызовет метод, изменяющий его состояние? Как вы можете себе представить, в такой ситуации потоки вполне могут наступить друг другу на ноги. В зависимости от того, в каком порядке осуществлялся доступ к данным, их целостность может быть нарушена, а сам объект может быть поврежден. Такая ситуация называется состоянием гонки (race condition).

Пример состояния гонки

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

В качестве примера я выбрал примитивную программу, имитирующую работу банка, основная задача которой заключается в совершении регулярных переводов между счетами. В нашем банке будет открыто 100 счетов, каждый из которых изначально имеет баланс в 1000 долларов. Источник и получатель перевода выбираются случайным образом. В многопоточной системе это может вызвать проблемы, что видно из кода метода transfer класса Bank.

public void transfer(int from, int to, double amount)
   // ВНИМАНИЕ: метод не безопасен для вызова из нескольких потоков
{
   System.out.print(Thread.currentThread());
   accounts[from] -= amount;
   System.out.printf(" %10.2f from %d to %d", amount, from, to);
   accounts[to] += amount;
   System.out.printf(" Total Balance: %10.2f%n", getTotalBalance());
}

А вот код c Runnable инстансами. Метод run занимается переводом денег с заданного банковского счета. На каждой итерации метод run выбирает случайный целевой счет и случайную сумму, вызывает метод transfer объекта Bank, а затем засыпает.

Runnable r = () ->
   {
      try
      {
         while (true)
         {
            int toAccount = (int) (bank.size() * Math.random());
            double amount = MAX_AMOUNT * Math.random();
            bank.transfer(fromAccount, toAccount, amount);
            Thread.sleep((int) (DELAY * Math.random()));
         }
      }
      catch (InterruptedException e) {}
   };

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

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

...
Thread[Thread-11,5,main]
Thread[Thread-12,5,main]
Thread[Thread-14,5,main]
Thread[Thread-13,5,main]
...
Thread[Thread-36,5,main]
Thread[Thread-35,5,main]
Thread[Thread-37,5,main]
Thread[Thread-34,5,main]
Thread[Thread-36,5,main]
...
Thread[Thread-4,5,main]Thread[Thread-33,5,main] 7.31 from 31 to 32 Total Balance: 99979.24
627.50 from 4 to 5 Total Balance: 99979.24
...

Как видите, что-то пошло не так. В течение нескольких операций баланс банка остается на уровне 100 000 долларов, что является правильной суммой для 100 счетов с тысячей долларов на каждом. Но через некоторое время баланс меняется! И эта ошибка может проявиться очень быстро, а может потребоваться очень много времени, чтобы баланс исказился. Тем не менее, такая ситуация не внушает доверия, и вы вряд ли захотите вкладывать свои кровные в такой банк.

Попробуйте обнаружить причину этой проблемы в коде, приведенном в Листинге 1, и классе Bank, приведенном в Листинге 2. Тайна будет раскрыта в следующем разделе.

Листинг 1. unsynch/UnsynchBankTest.java

package unsynch;

/**
 * Эта программа вызывает повреждение данных, когда к структуре данных обращаются сразу несколько потоков.
 * @версия 1.32 2018-04-10
 * @автор Кей Хорстманн
 */

public class UnsynchBankTest
{
   public static final int NACCOUNTS = 100;
   public static final double INITIAL_BALANCE = 1000;
   public static final double MAX_AMOUNT = 1000;
   public static final int DELAY = 10;

   public static void main(String[] args)
   {
      var bank = new Bank(NACCOUNTS, INITIAL_BALANCE);
      for (int i = 0; i < NACCOUNTS; i++)
      {
         int fromAccount = i;
         Runnable r = () ->
            {
               try
               {
                  while (true)
                  {
                     int toAccount = (int) (bank.size() * Math.random());
                     double amount = MAX_AMOUNT * Math.random();
                     bank.transfer(fromAccount, toAccount, amount);
                     Thread.sleep((int) (DELAY * Math.random()));
                  }
               }
               catch {InterruptedException e)
               {
               }
            };
         var t = new
         t.start();
      }
   }
}

Листинг 2. threads/Bank.java

package threads;

import java.util.*;

/**
 * Банк с несколькими банковскими счетами.
 */
public class Bank
{
   private final double[] accounts;

   /**
    * Конструктор.
    * @param n — количество счетов
    * @param initialBalance — изначальный баланс на каждом счете
    */

   public Bank(int n, double initialBalance)
   {
      accounts = new double[n];
      Arrays.fill(accounts, initialBalance);
   }

   /**
    * Осуществляет переводы средств с одного счета на другой.
    * @param from — счет, с которого осуществляется перевод
    * @param to — счет, куда осуществляется перевод
    * @param amount — сумма для перевода
    */

   public void transfer(int from, int to, double amount)
   {
      if (accounts[from] < amount) return;
      System.out.print(Thread.currentThread());
      accounts[from] -= amount;
      System.out.printf(" %10.2f from %d to %d", amount, from, to);
      accounts[to] += amount;
      System.out.printf(" Total Balance: %10.2f%n", getTotalBalance());
   }

   /**
    * Считает сумму остатков на всех счетах.
    * @return общий баланс
    */

   public double getTotalBalance()
   {
      double sum = 0;
      for (double a : accounts)
         sum += a;
      return sum; }
   /**
    * Возвращает количества счетов в банке.
    * @return количество счетов
    */
   public int size()
   {
      return accounts.length;
   }
}

Объяснение состояния гонки

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

accounts[to] += amount;

Проблема заключается в том, что это не атомарные операции. Эта инструкция обрабатывается следующим образом:

• Шаг 1: accounts[to] загружается в регистр.

• Шаг 2: Добавляется amount.

• Шаг 3: Результат перемещается обратно в accounts[to].

Предположим, что первый поток выполняет шаги 1 и 2, а затем выполнение перехватывается. В это время просыпается второй поток и обновляет ту же запись в массиве account. Затем пробуждается первый поток и завершает выполнение шага 3.

Это действие сбрасывает изменения, сделанные вторым потоком. В результате сумма перестает быть корректной (смотрите Рисунок 1). Тестовая программа обнаруживает эту некорректность. (И конечно, существует небольшая вероятность ложной тревоги, если работа потока была прервана во время выполнения тестов!)

Кстати, вы можете самостоятельно посмотреть байткод JVM, выполняющий каждый оператор класса. Просто выполните команду

javap -c -v Bank

для декомпиляции файла Bank.class. Например, строка

accounts[to] += amount;

транслируется в следующий байткод:

aload_0
getfield #2; //Поле accounts:[D
iload_2
dup2
daload
dload_3
dadd
dastore

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

Какова же вероятность возникновения повреждений данных такого рода? На современных процессорах с несколькими ядрами риск возникновения такого повреждения достаточно высок. Я смог увеличить вероятность наблюдения этой проблемы даже на одноядерном процессоре, просто чередуя операторы print с операторами, изменяющими баланс.

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

Если на сильно загруженной машине выполняется много потоков, то даже после удаления операторов print программа все равно будет сбоить. Ошибка может произойти как через несколько минут, так и через несколько часов или даже дней. Но она обязательно произойдет. Честно говоря, в жизни программиста мало что может быть хуже, чем ошибка, которая проявляется нерегулярно.

В общем, проблема заключается в том, что работа метода transfer может быть прервана на середине. Если бы был какой-нибудь способ гарантировать завершение работы метода до того, как поток потеряет управление, то состояние объекта Bank никогда не было бы скомпрометировано таким образом.

Объекты блокировки

В языке Java существует два механизма защиты блока кода от параллельного доступа. Для достижения этой цели предусмотрено ключевое слово synchronized, а в Java 5 появился еще и класс ReentrantLock. Ключевое слово synchronized автоматически предоставляет блокировку и связанное с ней условие, что делает его мощным и удобным решением для большинства случаев, предполагающих явное блокирование.

Тем не менее, я считаю, что для формирования правильного понимания работы ключевого слова synchronized вы сначала должны по отдельности разобраться с блокировками (locks) и условиями (conditions). Пакет java.util.concurrent предоставляет отдельные классы для этих фундаментальных механизмов. О них сейчас и пойдет речь. Как только мы разберемся с этими конструкциями, мы сразу же перейдем к ключевому слову synchronized.

Типовая схема защиты блока кода с помощью ReentrantLock выглядит следующим образом:

myLock.lock(); // объект ReentrantLock 
try
{
   <em>critical section</em>
}
finally
{
   myLock.unlock(); // так мы гарантируем, что блокировка будет снята, даже если было выброшено исключение
}

Эта конструкция гарантирует, что в критическую секцию может войти только один поток за раз. Как только один поток вызовет метод lock объекта блокировки (lock object), ни один другой поток не сможет пройти дальше этого lock-оператора. Когда другие потоки вызывают lock, они деактивируются до тех пор, пока первый поток не разблокирует объект блокировки.

Внимание: Очень важно, чтобы операция unlock была заключена в блок finally, поскольку если код в критической секции выбросит исключение, то блокировка должна быть снята. В противном случае другие потоки будут заблокированы навсегда.

Обратите внимание, что при использовании блокировок нельзя применять оператор try-with-resources. В первую очередь потому, что метод unlock не называется close. Но даже если бы он был переименован, оператор try-with-resources не сработал бы, поскольку его заголовок предполагает объявления новой переменной. Однако, когда вы используете объект блокировки вы хотите продолжать использовать ту же самую переменную, которая должна быть общей для всех потоков. Таким образом, это не сработает.

Вы можете использовать блокировку для защиты метода transfer класса Bank, как показано ниже:

public class Bank
{
   private Lock bankLock = new ReentrantLock();
   ...
   public void transfer(int from, int to, int amount)
   {
      bankLock.lock();
      try
      {
         System.out.print(Thread.currentThread());
         accounts[from] -= amount;
         System.out.printf(" %10.2f from %d to %d", amount, from, to);
         accounts[to] += amount;
         System.out.printf(" Total Balance: %10.2f%n", getTotalBalance());
      }
      finally
      {
         bankLock.unlock();
      }
   }
}

Предположим, что один поток вызывает transfer и блокирует его. Предположим, что второй поток также вызывает transfer. Второй поток не может получить блокировку и сам блокируется при вызове метода lock. Он деактивируется и должен ждать, пока первый поток закончит выполнение метода transfer. Второй поток сможет продолжить работу, только когда первый поток снимет блокировку (смотрите Рисунок 2).

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

Обратите внимание, что каждый объект Bank имеет свой собственный объект ReentrantLock. Если два потока пытаются получить доступ к одному и тому же объекту Bank, то блокировка будет служить сериализатором доступа. Однако, если же два потока работают с разными объектам Bank, то каждый из них заполучает свою собственную блокировку, и ни один из потоков не блокируется. Это вполне логично, ведь потоки не могут помешать друг другу при работе с разными экземплярами Bank.

Блокировка называется реентерабельной (reentrant), поскольку поток может неоднократно получать блокировку, которой он уже владеет. У блокировки есть счетчик удержаний (hold count), который отслеживает вложенные вызовы метода lock. Для снятия блокировки поток должен вызывать unlock на каждый вызов метода lock. Благодаря этой особенности код, защищенный блокировкой, может вызывать другой метод, использующий ту же самую блокировку.

Например, метод transfer вызывает метод getTotalBalance, который также блокирует объект bankLock, счетчик удержания которого теперь равен 2. После выхода из метода getTotalBalance счетчик удержания снова становится равным 1. После выхода из метода transfer счетчик удержания становится равным 0, и поток снимает блокировку.

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

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

Объекты условий

Иногда поток входит в критическую секцию только для того, чтобы обнаружить, что он не может продолжать работу до тех пор, пока не будет выполнено какое-либо условие. Для управления потоками, которые получили блокировку, но не могут выполнять работу, можно использовать объект условия – condition object. (Так исторически сложилось, что объекты условий зачастую называют условными переменными – condition variables).

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

if (bank.getBalance(from) >= amount) bank.transfer(from, to, amount);

Может случиться так, что между успешным результатом теста и вызовом transfer текущий поток будет деактивирован, как показано ниже:

if (bank.getBalance(from) >= amount)
// поток может быть деактивирован в этот момент
bank.transfer(from, to, amount);

К тому моменту, когда поток снова будет запущен, баланс счета может стать меньше суммы перевода. Необходимо убедиться в том, что никакой другой поток не сможет изменить баланс между тестом и событием перевода. Для этого необходимо защитить тест и действие перевода с помощью блокировки следующим образом:

public void transfer(int from, int to, int amount)
{
   bankLock.lock();
   try
   {
      while (accounts[from] < amount)
      {
         // ожидание
         ...
      }
      // перевод средств
      ...
   }
   finally
   {
      bankLock.unlock();
   }
}

Что делать, если на счету недостаточно средств? Можно подождать, пока какой-нибудь другой поток пополнит счет. Но этот поток только что получил эксклюзивный доступ к bankLock, поэтому ни один другой поток не сможет пополнить счет. Вот здесь-то и приходят на помощь объекты условий.

Объект блокировки может иметь один или несколько связанных с ним объектов Condition. Получить объект условия можно с помощью метода newCondition. Принято давать каждому такому объекту имя, вызывающее то условие, которое он представляет. Например, ниже задается объект для представления условия "достаточное количество средств" (“sufficient funds”):

class Bank
{
   private Condition sufficientFunds;
   ...
   public Bank()
   {
      ...
      sufficientFunds = bankLock.newCondition();
   }
}

Если метод transfer обнаруживает, что средств не достаточно, он вызывает

sufficientFunds.await();

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

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

Когда другой поток переведет средства, он должен вызвать метод

sufficientFunds.signalAll();

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

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

Обычно вызов await находится внутри цикла следующего вида:

while (!(OK to proceed))
   condition.await();

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

Практическое правило заключается в том, чтобы вызывать signalAll всякий раз, когда состояние объекта изменяется таким образом, что это может быть нужно ожидающим потокам. Например, при изменении баланса счета ожидающим потокам следует дать еще один шанс проверить баланс. Например, в следующем коде вызов signalAll производится после завершения перевода средств:

public void transfer(int from, int to, int amount)
{
   bankLock.lock(); try
   {
      While (accounts[from] < amount)
         sufficientFunds.await();
      // перевод средств
      ...
      sufficientFunds.signalAll();
   }
   finally
   {
      bankLock.unlock();
   }
}

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

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

Стоит отметить, что поток может вызывать await, signalAll, или signal по условию только в том случае, если он владеет блокировкой этого условия.

Если вы запустите пример программы из Листинга 3, вы больше не заметите никаких сбоев: общий баланс навсегда останется на уровне 100 000 долларов. Ни один счет не имеет отрицательного баланса. (Для завершения работы программы нажмите Ctrl+C.) Можно также заметить, что программа работает немного медленнее - такова цена, которую приходится платить за дополнительный бухгалтерский учет, связанный с механизмом синхронизации.

Листинг 3. synch/Bank.java

package synch;

import java.util.*;
import java.util.concurrent.locks.*;

/**
 * Банк с несколькими банковскими счетами, который использует блокировки для сериализации доступа.
 */

public class Bank
{
   private final double[] accounts;
   private Lock bankLock;
   private Condition sufficientFunds;

   /**
    * Конструктор.
    * @param n — количество счетов
    * @param initialBalance — изначальный баланс на каждом счете
    */

   public Bank(int n, double initialBalance)
   {
      accounts = new double[n];
      Arrays.fill(accounts, initialBalance);
      bankLock = new ReentrantLock();
      sufficientFunds = bankLock.newCondition();
   }

   /**
    * Осуществляет переводы средств с одного счета на другой.
    * @param from — счет, с которого осуществляется перевод
    * @param to — счет, куда осуществляется перевод
    * @param amount — сумма для перевода
    */

   public void transfer(int from, int to, double amount) throws InterruptedException
   {
      bankLock.lock();
      try
      {
         while (accounts[from] < amount)
            sufficientFunds.await();
         System.out.print(Thread.currentThread());
         accounts[from] -= amount;
         System.out.printf(" %10.2f from %d to %d", amount, from, to);
         accounts[to] += amount;
         System.out.printf(" Total Balance: %10.2f%n", getTotalBalance());
         sufficientFunds.signalAll();
      }
      finally
      {
         bankLock.unlock();
      }
   }

   /**
    * Считает сумму остатков на всех счетах.
    * @return общий баланс
    */


   public double getTotalBalance()
   {
      bankLock.lock();
      try
      {
         double sum = 0;
         for (double a : accounts)
            sum += a;
            return sum;
      }
      finally
      {
         bankLock.unlock();
      }
   }

   /**
    * Возвращает количества счетов в банке.
    * @return количество счетов
    */

   public int size()
   {
      return accounts.length;
   }
}

Заключение

В этой первой статье из цикла трех частей, посвященного синхронизации потоков, были рассмотрены основы состояний гонки, объектов блокировок, объектов условий, методов await, signal и signalAll.

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

В заключении серии будут описаны волатильные поля, окончательные переменные, атомарные операции, взаимоблокировки, устаревшие методы stop и suspend  и инициализация по требованию.

Дополнительные материалы

• Java Language Specification: Threads and locks

• The Java Tutorials: The try-with-resources statement

• Interface Condition

Материал подготовлен в преддверии старта онлайн-курса "Java Developer. Professional".