Класс AbstractQueuedSynchronizer

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements Serializable

Предоставляет основу для реализации блокирующих блокировок и связанных синхронизаторов (семафоров, событий и т. д.), использующих очереди ожидания типа «первым пришёл — первым обслужен» (FIFO). Этот класс служит полезной основой для большинства типов синхронизаторов, в которых для представления состояния используется одно атомарное значение int. Подклассы должны определить защищённые методы, изменяющие это состояние и определяющие, что это состояние означает с точки зрения захвата или освобождения данного объекта. На их основе остальные методы этого класса выполняют всю работу с очередями и блокировкой. Подклассы могут поддерживать другие поля состояния, но с точки зрения синхронизации отслеживается только атомарно обновляемое значение int, изменяемое с помощью методов getState(), setState(int) и compareAndSetState(int, int).

Подклассы следует объявлять как непубличные внутренние вспомогательные классы, используемые для реализации свойств синхронизации содержащего их класса. Класс AbstractQueuedSynchronizer не реализует никаких интерфейсов синхронизации. Вместо этого он определяет такие методы, как acquireInterruptibly(int), которые конкретные блокировки и связанные синхронизаторы могут вызывать по мере необходимости для реализации своих публичных методов.

Этот класс поддерживает один или оба режима: эксклюзивный режим по умолчанию и разделяемый режим. При захвате в эксклюзивном режиме попытки захвата другими потоками не могут завершиться успешно. Захват в разделяемом режиме несколькими потоками может (но не обязательно должен) завершиться успешно. Этот класс не «понимает» различий между режимами иначе, чем в механическом смысле: когда захват в разделяемом режиме завершается успешно, следующий ожидающий поток (если он есть) также должен определить, может ли он выполнить захват. Потоки, ожидающие в разных режимах, используют одну и ту же очередь FIFO. Обычно подклассы реализации поддерживают только один из этих режимов, однако оба режима могут использоваться, например, в ReadWriteLock. Подклассам, поддерживающим только эксклюзивный или только разделяемый режим, не нужно определять методы для неиспользуемого режима.

Этот класс определяет вложенный класс AbstractQueuedSynchronizer.ConditionObject, который подклассы, поддерживающие эксклюзивный режим, могут использовать в качестве реализации Condition. Для этого метод isHeldExclusively() должен сообщать, удерживается ли блокировка синхронизации эксклюзивно текущим потоком; вызванный с текущим значением getState() метод release(int) должен полностью освобождать этот объект; а базовая версия метода acquire(int) должна в конечном итоге восстанавливать прежнее состояние захвата объекта, получив сохранённое значение состояния. Другой метод AbstractQueuedSynchronizer не создаёт такое условие, поэтому не используйте его, если это требование не может быть выполнено. Поведение AbstractQueuedSynchronizer.ConditionObject, конечно, зависит от семантики реализации синхронизатора.

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

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

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

Чтобы использовать этот класс как основу синхронизатора, переопределите следующие методы (при необходимости), проверяя и/или изменяя состояние синхронизации с помощью getState(), setState(int) и/или compareAndSetState(int, int):

• tryAcquire(int)
• tryRelease(int)
• tryAcquireShared(int)
• tryReleaseShared(int)
• isHeldExclusively()

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

Вам также могут пригодиться унаследованные методы из AbstractOwnableSynchronizer для отслеживания потока, владеющего эксклюзивным синхронизатором. Рекомендуется использовать их: это позволяет инструментам мониторинга и диагностики помогать пользователям определять, какие потоки удерживают блокировки.

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

Acquire:
    while (!tryAcquire(arg)) {
       enqueue thread if it is not already queued;
       possibly block current thread;
    }

Release:
    if (tryRelease(arg))
       unblock the first queued thread;

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

Поскольку проверки при захвате выполняются до помещения потока в очередь, новый поток, пытающийся выполнить захват, может обойти другие заблокированные потоки, уже находящиеся в очереди. Однако при необходимости можно определить tryAcquire и/или tryAcquireShared, чтобы отключить обход, вызывая внутри один или несколько методов проверки и обеспечивая тем самым справедливый порядок захвата FIFO. В частности, большинство справедливых синхронизаторов могут определить tryAcquire так, чтобы метод возвращал false, если hasQueuedPredecessors() (метод, специально предназначенный для справедливых синхронизаторов) возвращает true. Возможны и другие варианты.

Пропускная способность и масштабируемость обычно максимальны при использовании стратегии обхода по умолчанию (также известной как жадная, отказ от очередности и предотвращение конвоев). Хотя она не гарантирует ни справедливости, ни отсутствия голодания, более ранние потоки в очереди могут повторно пытаться выполнить захват раньше более поздних, и каждая такая повторная попытка имеет равные шансы на успех в сравнении с новыми потоками. Кроме того, хотя при захвате обычно не используется «активное ожидание», перед блокировкой могут выполняться многократные вызовы tryAcquire, чередующиеся с другими вычислениями. Это даёт большинство преимуществ активного ожидания, когда эксклюзивная блокировка удерживается недолго, без большинства его недостатков в противном случае. При желании можно дополнить этот подход проверками «быстрого пути» перед вызовами методов захвата, предварительно проверяя hasContended() и/или hasQueuedThreads(), чтобы выполнять такие проверки только тогда, когда вероятность конкуренции за синхронизатор невелика.

Этот класс обеспечивает эффективную и масштабируемую основу для синхронизации, в частности, ограничивая область применения синхронизаторами, которые могут использовать int состояние, параметры захвата и освобождения, а также внутреннюю очередь ожидания FIFO. Если этого недостаточно, синхронизаторы можно создавать на более низком уровне с помощью классов atomic, собственных классов Queue и средств блокировки LockSupport.

Примеры использования

Ниже приведён класс нерекурсивной блокировки взаимного исключения, в котором нулевое значение обозначает разблокированное состояние, а единица — заблокированное. Хотя нерекурсивной блокировке не обязательно записывать поток-владелец, этот класс всё же делает это, чтобы упростить мониторинг использования. Он также поддерживает условия и предоставляет несколько методов инструментирования:

class Mutex implements Lock, java.io.Serializable {

  // Our internal helper class
  private static class Sync extends AbstractQueuedSynchronizer {
    // Acquires the lock if state is zero
    public boolean tryAcquire(int acquires) {
      assert acquires == 1; // Otherwise unused
      if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
      }
      return false;
    }

    // Releases the lock by setting state to zero
    protected boolean tryRelease(int releases) {
      assert releases == 1; // Otherwise unused
      if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
      setExclusiveOwnerThread(null);
      setState(0);
      return true;
    }

    // Reports whether in locked state
    public boolean isLocked() {
      return getState() != 0;
    }

    public boolean isHeldExclusively() {
      // a data race, but safe due to out-of-thin-air guarantees
      return getExclusiveOwnerThread() == Thread.currentThread();
    }

    // Provides a Condition
    public Condition newCondition() {
      return new ConditionObject();
    }

    // Deserializes properly
    private void readObject(ObjectInputStream s)
        throws IOException, ClassNotFoundException {
      s.defaultReadObject();
      setState(0); // reset to unlocked state
    }
  }

  // The sync object does all the hard work. We just forward to it.
  private final Sync sync = new Sync();

  public void lock()              { sync.acquire(1); }
  public boolean tryLock()        { return sync.tryAcquire(1); }
  public void unlock()            { sync.release(1); }
  public Condition newCondition() { return sync.newCondition(); }
  public boolean isLocked()       { return sync.isLocked(); }
  public boolean isHeldByCurrentThread() {
    return sync.isHeldExclusively();
  }
  public boolean hasQueuedThreads() {
    return sync.hasQueuedThreads();
  }
  public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
  }
  public boolean tryLock(long timeout, TimeUnit unit)
      throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
  }
}

Ниже приведён класс защёлки, похожий на CountDownLatch, но для его срабатывания требуется только один signal. Поскольку защёлка не является эксклюзивной, в ней используются методы захвата и освобождения shared.

class BooleanLatch {

  private static class Sync extends AbstractQueuedSynchronizer {
    boolean isSignalled() { return getState() != 0; }

    protected int tryAcquireShared(int ignore) {
      return isSignalled() ? 1 : -1;
    }

    protected boolean tryReleaseShared(int ignore) {
      setState(1);
      return true;
    }
  }

  private final Sync sync = new Sync();
  public boolean isSignalled() { return sync.isSignalled(); }
  public void signal()         { sync.releaseShared(1); }
  public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
  }
}