При участии Тима Перлса, Джошуа Блоха, Джозева Боубира, Дэвида Холмса и Дага Ли Параллельное программирование в java на
 Скачать 4.97 Mb.
|
|
Листинг 14.9 Повторно закрываемый затвор, использующий методы wait and notifyAll Предикат условия, используемый методом await , сложнее простой проверки метода isOpen . Это необходимо, потому что если N потоков ожидают у затвора в то время, когда он открыт, им всем должно быть позволено продолжить выполнение. Но, если бы затвор открывался и закрывался в быстрой последовательности, не все потоки могли бы быть освобождены, если бы метод await проверял только метод isOpen : к тому времени, когда все потоки получают уведомление, повторно захватывают блокировку и выходят из метода wait , затвор, возможно, уже вновь закрылся. Поэтому класс ThreadGate использует несколько более сложный предикат условия: каждый раз, когда затвор открывается, счетчик “генерации” увеличивается, и поток может пройти метод await , если затвор открыт сейчас, или если затвор открылся с тех пор, как этот поток прибыл к нему. Так как класс ThreadGate поддерживает только ожидание открытия затвора, он выполняет уведомление только в методе open ; для поддержки обеих операций “ожидать открытия” и “ожидать закрытия”, ему придётся выполнять уведомления и в методе open , и в методе close . Это является иллюстрацией того, почему зависящие от состояния классы могут быть хрупкими в поддержке - добавление новой, зависящей от состояния, операции может потребовать изменения множества веток выполнения кода, которые изменяют состояние объекта, так чтобы соответствующие уведомления могли быть выполнены. 14.2.6 Вопросы безопасности подклассов Использование одиночного уведомления или уведомления по условию вводит ограничения, которые могут усложнить реализацию подклассов [CPJ 3.3.3.3]. Если вы в целом хотите поддерживать подклассы, вам необходимо структурировать свой класс так, чтобы подклассы могли добавлять соответствующее уведомление от имени базового класса, если подкласс выполняет это таким образом, который нарушает одно из требований одиночного уведомления или уведомления по условию. Класс, зависящий от состояния, должен либо полностью раскрывать (и документировать) свои протоколы ожидания и уведомления для подклассов, либо вообще не включать в них подклассы. (Это расширение принципа “дизайн и документирование для наследования, или же его запрет” [EJ пункт 15].) По крайней мере, при разработке класса, зависящего от состояния, в целях наследования требуется предоставить доступ к очередям условий и блокировкам, а также документировать предикаты условий и политику синхронизации; также может потребоваться предоставить переменные состояния, лежащие в основе. (Самое худшее, что может сделать зависящий от состояния класс, - это предоставить своё состояние подклассам, но не документировать свои протоколы ожидания и уведомления; это похоже на класс, предоставляющий свои переменные состояния, но не документирующий свои инварианты.) Одним из вариантов сделать это, является запрет на создание подклассов, либо объявив класс как final , либо скрывая от подклассов очереди условий, блокировки и переменные состояния. В противном случае, если подкласс делает что-то для дискредитации способа, с помощью которого базовый класс использует метод notify , он должен быть в состоянии возместить ущерб. Рассмотрим неограниченный блокирующий стек, в котором операция изъятия (pop) блокируется, если стек пуст, но операция вложить (push) всегда может быть выполнена. Это удовлетворяет требованию для одиночного уведомления. Если этот класс использует одиночное уведомление, а подкласс добавляет блокирующий метод “последовательно изъять два элемента”, то теперь существует два класса ожиданий: ожидающие появления одного элемента и ожидающие появления двух. Но если базовый класс предоставляет очередь условий и документирует свои протоколы по её использования, подкласс может переопределить метод вложения, для выполнения метода notifyAll , таким образом, восстанавливая безопасность. 14.2.7 Инкапсулирование очередей условий Обычно лучше инкапсулировать очередь условий, чтобы она была недоступна вне иерархии классов, в которой она используется. В противном случае у вызывающих объектов может возникнуть соблазн подумать, что они понимают ваши протоколы ожидания и уведомления и могут использовать их в не соответствующей вашему дизайну манере. (Невозможно принудительно применить требование о единообразном ожидании для одиночного уведомления, если объект очереди условий недоступен для кода, который вы не контролируете; если чужой код ошибочно ожидает на очереди условий, это может привести к нарушению протокола уведомлений и вызвать захват сигнала.) К сожалению, этот совет - инкапсулировать объекты, используемые в качестве очередей условий - не согласуется с наиболее распространенным шаблоном проектирования для потокобезопасных классов, в котором внутренняя блокировка объекта используется для защиты его состояния. Класс BoundedBuffer иллюстрирует эту распространенную идиому, в нём объект буфера сам по себе является и блокировкой и очередью условий. Однако класс BoundedBuffer может быть легко реструктурирован для использования приватного объекта блокировки и очереди условий; единственное различие будет заключаться в том, что он больше не будет поддерживать любую форму блокировки на стороне клиента. 14.2.8 Протоколы входа и выхода Веллингс (Wellings, 2004) характеризует надлежащее использование методов wait и notify с точки зрения протоколов входа (entry) и выхода (exit). Для каждой операции, зависящей от состояния, и для каждой операции, изменяющей состояние, от которого зависит другая операция, необходимо определить и задокументировать протоколы входа и выхода. Протокол входа - это предикат условия операции; протокол выхода включает в себя проверку всех переменных состояния, которые были изменены операцией, чтобы узнать, не вызвали ли они выполнение какого-либо другого предиката условия, и если это так, то отправить уведомление связанной очереди условий. Класс AbstractQueuedSynchronizer , на основе которого построено большинство зависимых от состояния классов из пакета java.util.concurrent (см. раздел 14.4 ), использует концепцию протокола выхода. Вместо того чтобы позволить классам синхронизатора выполнять своё собственное уведомление, он вместо этого требует, чтобы методы синхронизатора возвращали значение, указывающее, могут ли его действия разблокировать один или несколько ожидающих потоков. Это явное требование API затрудняет возникновение ситуации, при которой можно “забыть” выполнить уведомление о некоторых переходах состояний. 14.3 Явные объекты условия Как мы видели в главе 13 , явные блокировки могут быть полезны в тех случаях, когда встроенные блокировки не обладают достаточной гибкостью. Подобно тому, как интерфейс Lock является обобщением внутренних блокировок, интерфейс Condition (см. листинг 14.10) является обобщением внутренних очередей условий. public interface Condition { void await() throws InterruptedException; boolean await(long time, TimeUnit unit) throws InterruptedException; long awaitNanos(long nanosTimeout) throws InterruptedException; void awaitUninterruptibly(); boolean awaitUntil(Date deadline) throws InterruptedException; void signal(); void signalAll(); } Листинг 14.10 Интерфейс Condition Внутренние очереди условий имеют несколько недостатков. Каждая внутренняя блокировка может быть связана только с одной очередью условий, что означает, что в классах подобных классу BoundedBuffer , множество потоков может ожидать на одной очереди условий различных предикатов условий, а наиболее распространенный шаблон блокировки включает предоставление объекта очереди условий. Оба этих фактора делают невозможным применение единообразного требования к ожидающим, в отношении использования метода notify . Если вы хотите написать параллельный объект с несколькими предикатами условий или хотите осуществлять больший контроль над видимостью очереди условий, явные классы Lock и Condition предлагают более гибкую альтернативу внутренним блокировкам и очередям условий. Экземпляр Condition связан с одиночным экземпляром Lock , так же как очередь условий связана с одиночной внутренней блокировкой; для создания экземпляра Condition , вызовите метод Lock.newCondition у связанной блокировки. Подобно тому, как интерфейс Lock предлагает более богатый набор функций, чем внутренняя блокировка, интерфейс Condition предлагает более богатый набор функций, чем внутренние очереди условий: несколько наборов ожиданий (wait set) для каждой блокировки, прерываемые и непрерываемые ожидания по условию, ожидания, ограниченные по сроку, и выбор из справедливого и несправедливого помещения в очередь. В отличие от встроенных очередей условий, можно иметь столько объектов Condition для каждого объекта Lock , сколько вам необходимо. Объекты Condition наследуют настройки справедливости от связанного с ними экземпляра Lock ; в случае использования справедливых блокировок, потоки освобождаются из метода Condition.await в порядке FIFO. Предупреждение об опасности: эквивалентом методов wait , notify и notifyAll для объектов Condition являются методы await , signal и signalAll . Однако класс Condition 156 расширяет класс Object , что означает, что он также имеет методы wait и notify . Обязательно используйте правильные версии методов - await и signal - вместо наследуемых! В листинге 14.11 показана еще одна реализация ограниченного буфера, на этот раз использующая два экземпляра Condition , - notFull и notEmpty - для явного представления предикатов условий “не полон” и “не пуст”. Когда метод take блокируется из-за того, что буфер пуст, он ожидает на условии notEmpty , и метод put разблокирует любые потоки, заблокированные в методе take , отправляя сигнал условию notEmpty @ThreadSafe public class ConditionBoundedBuffer // CONDITION PREDICATE: notFull (count < items.length) private final Condition notFull = lock.newCondition(); // CONDITION PREDICATE: notEmpty (count > 0) private final Condition notEmpty = lock.newCondition(); @GuardedBy("lock") private final T[] items = (T[]) new Object[BUFFER_SIZE]; @GuardedBy("lock") private int tail, head, count; // BLOCKS-UNTIL: notFull 156 Подразумевается конкретная реализация интерфейса Condition public void put(T x) throws InterruptedException { lock.lock(); try { while (count == items.length) notFull.await(); items[tail] = x; if (++tail == items.length) tail = 0; ++count; notEmpty.signal(); } finally { lock.unlock(); } } // BLOCKS-UNTIL: notEmpty public T take() throws InterruptedException { lock.lock(); try { while (count == 0) notEmpty.await(); T x = items[head]; items[head] = null; if (++head == items.length) head = 0; --count; notFull.signal(); return x; } finally { lock.unlock(); } } } Листинг 14.11 Ограниченный буфер, использующий явные условные переменные Поведение класса ConditionBoundedBuffer аналогично поведению класса BoundedBuffer , но использование им очередей условий удобнее для чтения - легче анализировать класс, использующий множество экземпляров Condition , чем тот, что использует единственную внутреннюю очередь условий с несколькими предикатами условий. Разделив два предиката условия на отдельные наборы ожидания, интерфейс Condition упрощает выполнение требований для одиночного уведомления. Использование более эффективного метода signal , вместо метода signalAll , уменьшает количество переключений контекста и количество захватов блокировок, вызванных каждой операцией буфера. Как и во внутренних блокировках и в очередях условий, трехсторонняя связь между блокировкой, предикатом условия и условной переменной также должна выполняться при использовании явных реализаций Lock и Condition Переменные, участвующие в предикате условия, должны быть защищены экземпляром блокировки Lock , и блокировка Lock должна удерживаться при тестировании предиката условия и при вызове методов await и signal 157 Выбирайте между использованием явных экземпляров Condition и внутренних очередей условий так же, как если бы вы выбирали между классом ReentrantLock и блоком synchronized : используйте экземпляр Condition , если вам нужны его расширенные функции, такие как справедливая очередь или несколько наборов ожидания на каждую блокировку, в противном случае, отдайте предпочтение использованию внутренних очередей условий. (Если вы уже используете класс ReentrantLock , потому что вам нужны его расширенные функции, выбор уже сделан.) 14.4 Анатомия синхронизатора Интерфейсы классов ReentrantLock и Semaphore имеют много общего. Оба класса выступают в качестве “затворов”, позволяя одновременно только ограниченному числу потоков продолжать выполнение; потоки прибывают к затвору и допускаются через него (методы lock или acquire успешно возвращают управление), ожидают (методы lock или acquire блокируются), или возвращаются назад (методы tryLock или tryAcquire возвращают false , указывая, что блокировка или разрешение не были получены за отведённое на ожидание время). Далее, оба класса позволяют выполнять прерываемые, непрерываемые и ограниченные по времени попытки захвата, и оба позволяют выбрать справедливое или несправедливое помещение в очередь ожидания потоков. Учитывая эту общность, вы можете подумать, что класс Semaphore был реализован поверх класса ReentrantLock , или, возможно, класс ReentrantLock был реализован как класс Semaphore с одним разрешением. Это было бы вполне практично; это распространённое упражнение, заключающееся в том, чтобы доказать, что подсчитывающий семафор может быть реализован с помощью блокировки (как в случае класса SemaphoreOnLock , приведённого в листинге 14.12) и что блокировка может быть реализована с помощью подсчитывающего семафора. // Not really how java.util.concurrent.Semaphore is implemented @ThreadSafe public class SemaphoreOnLock { private final Lock lock = new ReentrantLock(); // CONDITION PREDICATE: permitsAvailable (permits > 0) private final Condition permitsAvailable = lock.newCondition(); @GuardedBy("lock") private int permits; SemaphoreOnLock(int initialPermits) { lock.lock(); try { permits = initialPermits; } finally { lock.unlock(); } } 157 Класс ReentrantLock требует, чтобы при вызове методов signal или signalAll удерживалась блокировка, но реализациям Lock разрешается создавать экземпляры Condition , которые не имеют этого требования. // BLOCKS-UNTIL: permitsAvailable public void acquire() throws InterruptedException { lock.lock(); try { while (permits <= 0) permitsAvailable.await(); --permits; } finally { lock.unlock(); } } public void release() { lock.lock(); try { ++permits; permitsAvailable.signal(); } finally { lock.unlock(); } } } Листинг 14.12 Подсчитывающий семафор, реализованный с помощью интерфейса Lock На самом деле, они оба реализованы с использованием общего базового класса AbstractQueuedSynchronizer (AQS) - как, впрочем, и многие другие синхронизаторы. Класс AQS представляет собой фреймворк для построения блокировок и синхронизаторов, и с его использованием может быть легко и эффективно построен на удивление обширный ряд синхронизаторов. Не только классы ReentrantLock и Semaphore построены с использованием класса AQS, но также и CountDownLatch , ReentrantReadWriteLock , SynchronousQueue 158 , и FutureTask Класс AQS обрабатывает множество деталей реализации синхронизатора, например, таких как очередь FIFO ожидающих потоков. Конкретные синхронизаторы могут определять гибкие критерии того, должно ли потоку быть позволено пройти или от него требуется ожидать. Использование класса AQS для создания синхронизаторов имеет ряд преимуществ. Это не только приводит к значительному сокращению усилий, необходимых для реализации, но и не приводит к необходимости расплачиваться за возникновение нескольких точек конкуренции, в отличие от случая построения одного синхронизатора поверх другого. В классе SemaphoreOnLock получение разрешения может быть заблокировано в двух местах - один раз на блокировке, защищающей состояние семафора, а затем вновь, в случае если разрешение недоступно. Синхронизаторы, построенные с помощью класса AQS, имеют только одну точку, в которой они могут быть заблокированы, приводя, таким образом, к снижению издержек на переключение контекста и улучшению пропускной способности. Класс AQS был разработан для обеспечения масштабируемости, и 158 В Java 6 класс SynchronousQueue на основе AQS был заменён на (более масштабируемую) неблокирующую версию. все синхронизаторы в пакете java.util.concurrent , что были построены с его использованием, получают преимущества от этого. 14.5 Класс AbstractQueuedSynchronizer Большинство разработчиков, вероятно, никогда не будут использовать класс AQS напрямую; стандартный набор синхронизаторов охватывает довольно широкий диапазон ситуаций. Но, видение того, как реализованы стандартные синхронизаторы, может помочь в прояснении того, как они работают. Основные операции, которые выполняет синхронизатор, основанный на классе AQS, это несколько вариантов захвата и освобождения. Операция захвата зависит от состояния и всегда может быть заблокирована. В случае блокировки или семафора смысл операции захвата прост - захват блокировки или разрешение - и вызывающему объекту, возможно, придется ожидать, пока синхронизатор находится в том состоянии, в котором это может произойти. С классом CountDownLatch , захват означает “ожидать, пока защелка достигнет своего терминального состояния”, а в случае класса FutureTask , это означает “ожидать, пока задача не будет завершена”. Освобождение является не блокирующей операцией; освобождение может позволить потокам, заблокированным в процессе захвата, продолжить своё выполнение. Чтобы класс зависел от состояния, он должен иметь некоторое состояние. Класс AQS берет на себя задачу управления некоторыми состояниями для класса синхронизатора: он управляет информацией о состоянии как единым целым, которым можно управлять с помощью защищенных методов getState , setState и compareAndSetState . Этот механизм можно использовать для представления произвольного состояния; например, класс ReentrantLock использует его для отображения того, сколько раз поток-владелец захватывал блокировку, класс Semaphore использует его для представления количества оставшихся разрешений, а класс FutureTask использует его для отображения состояния задачи (не начата, выполняется, завершена, отменена). Синхронизаторы также могут сами управлять дополнительными переменными состояния; например, класс ReentrantLock отслеживает текущего владельца блокировки, чтобы различать реентерабельные и конкурирующие запросы на захват блокировки. Захват и освобождение блокировки принимает в классе AQS форму, приведённую в листинге 14.13. В зависимости от синхронизатора, захват может быть исключительным (exclusive), как в случае класса ReentrantLock , или неисключительным (nonexclusive), как в случае классов Semaphore и CountDownLatch . Операция захвата состоит из двух частей. Во-первых, синхронизатор решает, позволяет ли текущее состояние выполнить захват; если это так, потоку позволяется продолжить выполнение, а если нет, операция захвата блокируется или происходит сбой. Это решение определяется семантикой синхронизатора; например, захват блокировки может быть успешен, если блокировка не удерживается, и захват защелки может быть успешен, если защелка находится в терминальном состоянии. boolean acquire() throws InterruptedException { while (state does not permit acquire) { if (blocking acquisition requested) { enqueue current thread if not already queued block current thread } else return failure } possibly update synchronization state dequeue thread if it was queued return success } void release() { update synchronization state if (new state may permit a blocked thread to acquire) unblock one or more queued threads } |