Java. Полное руководство. 8-е издание. С. Н. Тригуб Перевод с английского и редакция
 Скачать 25.04 Mb.
|
|
void/acquire(int число throws InterruptedException Глава 27. Параллельные утилиты 7 Первая форма запрашивает одно разрешение, а вторая — число разрешений. Обычно используется первая форма. Если разрешение не будет предоставлено вовремя вызова метода, то вызывающий поток будет приостановлен до тех пор, пока не будет получено разрешение. Чтобы освободить разрешение, вызовите метод r e l e a s e ( ), который имеет следующие две формы release() void release(int число) Первая форма освобождает одно разрешение, а вторая — то количество разрешений, которое указано в параметре число Чтобы использовать семафор для управления доступом к ресурсу, каждый поток, которому необходимо использовать этот ресурс, должен вначале вызвать метод a c q u i r e ( ), прежде чем обращаться к ресурсу. Когда поток завершает свою работу с ресурсом, он должен вызвать метод r e l e a s e (). Ниже приведен пример использования семафора Пример простого семафора java.u t i l .concurrent.*; class SemDemo { public static void main(String a r g s []) { Semaphore sem = new Semaphored); new IncThread(sem, "A"); new DecThread(sem, "B"); } } // Общий ресурс class Shared { static int count = 0; } // Поток выполнения, увеличивающий значение счетчика на единицу class IncThread implements Runnable { String name; Semaphore sem; IncThread(Semaphore s, String n) { s em = s ; name = n; new Thread(this).st art(); } public void r u n () Запуск " + name); try { // Сначала получаем разрешение + " ожидает разрешения sem.acquire(); System.out.println(name + " получает разрешение Теперь обращаемся к общему ресурсу.ч for(int i=0; i < 5; i++) { Shared.count++; System.out.println(name + ": " + Shared.count); 8 7 Часть II. Библиотека Java 1 1 Если это возможно, разрешаем контекстное переключение Thread.sleep(10); } } catch (InterruptedException ехс) { System.out.println(ехс); } // Освобождаем разрешение + " освобождает разрешение sem.release() ; } } // Поток выполнения, уменьшающий значение счетчика на единицу class DecThread implements Runnable { String name; Semaphore sem; DecThread(Semaphore s, String n) { sem = s; name = n; new Thread(this).start(); } public void run() { System.out.println("Starting " + name); try { // Сначала получаем разрешение + " ожидает разрешения sem.acquire(); System.out.println(name + " получает разрешение Теперь обращаемся к общему ресурсу for(int i=0; i < 5; i++) { Shared.count--; System.out.println(name + " + Shared.count); // Если это возможно, разрешаем контекстное переключение Thread.sleep(10); } } catch (InterruptedException ехс) { System.out.println(ехс); } // Освобождаем разрешение + " освобождает разрешение sem.release() Ниже показаны результаты выполнения этой программы. (Конкретный порядок выполнения потоков может быть иным.) Запуск А А ожидает разрешения. А получает разрешение. А: Запуск В В ожидает разрешения Глава 27. Параллельные утилиты 8 7 А 2 А 3 А 4 А А освобождает разрешение. В получает разрешение. В: 4 В 3 В 2 В 1 ВО В освобождает разрешение. В программе используется семафор для управления доступом к переменной count, которая является статической переменной класса Shared. Значение переменной Shared, count увеличивается на 5 в методе run () класса IncThread и уменьшается на 5 в одноименном методе класса becThread. Для защиты этих двух потоков от одновременного доступа к переменной Shared. count, доступ предоставляется только после того, как будет получено разрешение от управляющего семафора. После того как доступ будет завершен, разрешение освобождается. Таким образом, только один поток водно и тоже время получит доступ к переменной Shared. count, что можно видеть из результатов вывода. Обратите внимание на то, что в методе run () классов IncThread и DecThread вызывается метод sleep () . Он гарантирует, что доступ к переменной Shared count будет синхронизироваться семафором. В методе run () вызов метода sleep () приводит к тому, что вызывающий поток будет приостанавливаться в промежутках между каждым доступом к переменной Shared, count. Как правило, благодаря этому должен выполняться второй поток. Однако поскольку мы работаем с семафором, то второй поток должен ожидать до тех пор, пока первый поток не освободит разрешение, а это происходит только после того, как будут завершены все доступы, производимые первым потоком. Таким образом, значение переменной Shared.count вначале увеличивается на 5 в объекте класса IncThread, а затем уменьшается на 5 в объекте класса DecThread. Увеличения и уменьшения значений происходят строго по порядку. Если бы мы не использовали семафор, то доступы к переменной Shared, count, производимые каждым потоком, осуществлялись бы одновременно, поэтому увеличение и уменьшение значения происходило бы вперемешку. Чтобы убедиться в этом, попробуйте закомментировать вызовы методов ac quire ( ) и release () . Запустив программу, вы увидите, что доступ к переменной Shared, count больше не является синхронизированными каждый поток обращается к переменной Shared, count сразу же, как только им выделяется временной интервал. Несмотря на то что во многих случаях применение семафора не представляет сложности, как это можно было видеть в предыдущей программе, возможны также и более сложные варианты его использования. Ниже показан один из таких примеров. Это переработанная версия программы поставщика и потребителя, представленной в главе 11. Здесь используется два семафора, которые регулируют потоки поставщика и потребителя и гарантируют, что за каждым вызовом метода put () будет следовать соответствующий вызов метода get (). // Реализация поставщика и потребителя, использующая // семафоры для управления синхронизацией import j a v a .ut i1.concurrent.Semaphore; class Q { int n; 8 7 Часть II. Библиотека Java 1 1 Начинаем с недоступного семафора потребителя static Semaphore semCon = new Semaphore(0); static Semaphore semProd = new Semaphored); void g e t () { try { semCon.acquire(); } catch(InterruptedException e) Произошло Получено " + n ) ; semProd.release(); } void p u t (int n) { try { semProd.acquire(); } c a t c h (InterruptedException e) Произошло InterruptedException"); } this.n = Отправлено " + n ) ; semCon.release(); } class Producer implements Runnable { Q q; Producer(Q q) { this.q = q; new Thread(this, "Producer").start(); } public void r u n () { for(int i=0; i < 20; i + +) q.put(i); } } class Consumer implements Runnable { Q q; Consumer(Q q) { this.q = q; new Thread(this, "Consumer").start(); } public void run() { for (int i = 0; i < 20; i + +) q . g e t O ; } } class ProdCon { public static void main(String a r g s []) { Q q = new Q (); new Consumer(q); new Ниже показана часть результатов Глава 27. Параллельные утилиты 8 7 Отправлено О Получено О Отправлено 1 Получено 1 Отправлено 2 Получено 2 Отправлено 3 Получено 3 Отправлено 4 Получено 4 Отправлено 5 Получено Как видите, здесь происходит синхронизация вызовов методов put () и get ( То есть после каждого вызова метода put ( ) следует вызов метода get () , поэтому ни одно значение не может быть пропущено. Если бы семафоры не использовались, вызовы метода put () могли бы происходить без чередования с вызовами метода get () , в результате чего значения были бы пропущены. (Чтобы убедиться в этом, удалите код семафора и посмотрите на полученные результаты.) Последовательность вызовов методов put () и get () обрабатывается двумя семафорами semProd и semCon. Прежде чем метод put () сможет произвести значение, он должен получить разрешение от семафора semProd. После того как значение будет определено, он освобождает семафор semProd. Прежде чем метод get () сможет использовать значение, он должен получить разрешение от семафора. После того как он закончит работу с этим значением, он освобождает семафор semCon. Такой механизм передачи и получения гарантирует, что за каждым вызовом метода put () будет следовать вызов метода get ( Обратите внимание на то, что семафор semCon инициализируется без доступных разрешений. Поэтому метод put () выполняется первым. Возможность задавать исходное состояние синхронизации является одной из наиболее ярких характеристик семафоров. Класс C o u n tD o w n L a tc Иногда необходимо, чтобы поток находился в режиме ожидания до тех пор, пока не произойдет одно или несколько событий. Для этих целей параллельный API предлагает защелку класса CountDownLatch. Этот класс изначально создается с количеством событий, которые должны произойти до того момента, как будет снята защелка. Каждый раз, когда происходит событие, значение счетчика уменьшается. Когда значение счетчика станет равным нулю, защелка будет снята. Класс CountDownLatch имеет следующий конструктор число) Здесь число определяет количество событий, которые должны произойти до того, как будет снята защелка. Для обслуживания защелки поток вызывает метод await () , который имеет следующие формы aw a i t () throws InterruptedException boolean await(long ожидать TimeUnit tu) throws В первом случае ожидание длится до тех пор, пока значение счета, связанного с вызывающим объектом класса CountDownLatch, не станет равно нулю. Во втором случае ожидание длится только в течение определенного периода време 8 7 Часть II. Библиотека ни, определенного параметром ожидать Единицы, представляемые этим параметром, определяются параметром и, который является объектом перечисления Tim eU nit. (Перечисление Tim eU nit рассматривается далее в этой главе) Метод возвращает значение f a l s e , если достигнут сроки значение t r u e — если обратный отсчет достигает нуля. Чтобы известить о событии, нужно вызвать метод countDown (). void При каждом вызове метода countDown () значение счета, связанного с вызывающим объектом, уменьшается на единицу. В следующей программе представлен пример применения класса C ount- DownLatch. В ней создается защелка, снять которую можно будет только по прошествии пяти событий Демонстрация применения CountDownLatch. import java.u t i l .concurrent.CountDownLatch; class CDLDemo { public static void main(String a r g s []) { CountDownLatch cdl = new Запуск MyThread(cdl); try { c d l .a w a i t (); } catch (InterruptedException exc) Завершение MyThread implements Runnable { CountDownLatch latch; MyThread(CountDownLatch c) { latch = c; new T h read(this).start(); } public void r u n () { for(int i = 0; i<5; i++) { System.out.println(i ); latch.countDown(); // обратный отсчет } } } Ниже показаны результаты выполнения этой программы. Запуск 0 1 2 3 4 Завершение Глава 27. Параллельные утилиты 8 7 Внутри метода main () создается защелка класса CountDownLatch по имени cdl, исходное значение которой равно 5. После этого создается экземпляр класса MyThread, который начинает выполнение нового потока. Обратите внимание на то, что защелка cdl передается в качестве параметра конструктору класса MyThread и сохраняется в переменной экземпляра latch. Затем главный поток вызывает метод await () для защелки cdl, в результате чего выполнение главного потока приостанавливается до тех пор, пока счетчик защелки cdl пять раз не уменьшится на единицу. Внутри метода run () конструктора класса MyThread создается цикл, который повторяется пять раз. Вовремя каждого повторения вызывается метод count - Down () для переменной экземпляра latch, который ссылается на защелку cdl в методе main () . После пятого повторения защелка снимается, позволяя возобновить главный поток. Класс CountDownLatch является мощными простым в использовании объектом синхронизации, который будет полезен в тех случаях, когда поток должен находиться в режиме ожидания, пока не произойдет одно или несколько событий. Класс C y c l i c B a r r i e В программировании нередко возникают такие ситуации, когда два или более потоков должны находиться в режиме ожидания в предварительно определенной точке выполнения до тех пор, пока все потоки из этого набора не достигнут этой точки. Для этого параллельные API предлагают класс CyclicBarrier. Он позволяет определить объект синхронизации, который приостанавливается до тех пор, пока определенное количество потоков не достигнет барьерной точки. Класс CyclicBarrier имеет следующие два конструктора y c l i c B a r r i e r (int количПотоков) C y c l i c B a r r i e r (int количПотоков, Runnable действие) Здесь параметр коли ч Потоков определяет количество потоков, которые должны достигнуть барьера до того, как выполнение будет продолжено. Во втором варианте параметр действие определяет поток, который будет выполняться по достижении барьера. Общая процедура использования класса C y c l i c B a r r i e r выглядит следующим образом. В первую очередь необходимо создать объект класса C y cl i с B ar г i e r , указав количество потоков для ожидания. Затем, когда каждый поток достигнет барьера, нужно вызвать метод a w a it () для данного объекта. В результате этого выполнение потока будет приостановлено до тех пор, пока все остальные потоки также не вызовут метод a w a it (). После того как указанное количество потоков достигнет барьера, метод a w a it () вернет результат и выполнение будет возобновлено. Кроме того, если определить какое-нибудь действие, то этот поток будет выполнен. Метод await () имеет следующие две формы a w a i t () throws InterruptedException, B r o k e n B arrierException int await(long ожидать TimeUnit tu) throws InterruptedException, BrokenBarrierException, T i m e В первом случае ожидание длится до тех пор, пока каждый поток не достигнет барьерной точки. Во втором случае ожидание длится в течение определенного периода времени ожидать Единицы времени этого параметра определяются параметром tu . В обоих случаях возвращается значение, показывающее порядок, в соответствии с которым потоки будут достигать барьерной точки. Первый поток возвращает значение, равное количеству ожидаемых потоков минус 1. Последний поток возвращает нуль 8 8 Часть II. Библиотека Ниже показан пример, иллюстрирующий класс C y c l i c B a r r i e r . Он ожидает, пока совокупность трех потоков не достигнет барьерной точки. После того как это произойдет, будет выполнен поток, определяемый при помощи класса a r A c t i o n . // Демонстрация применения CyclicBarrier. import java.util.concurrent.*; class BarDemo { public static void main(String a r g s []) { CyclicBarrier cb = new CyclicBarrier(3, new BarAction() Запуск MyThread(cb, "A"); new MyThread(cb, "B"); new MyThread(c b , "С "); } } // Поток выполнения, использующий CyclicBarrier. class MyThread implements Runnable { CyclicBarrier cbar; String name; MyThread(CyclicBarrier c, String n) { cbar = с ; name = n; new Thread(this).start(); } public void r u n () { System.out.println(name); try { c bar.a w a i t (); } catch (BrokenBarrierException exc) { System.out.println(exc); } catch (InterruptedException exc) { System.out.println(exc); } } } // Объект этого класса вызывается после завершения выполнения CyclicBarrier. class BarAction implements Runnable { public void r u n () Барьер достигнут!"); } } Ниже показаны результаты выполнения. Запуск А В С Глава 27. Параллельные утилиты 8 8 Барьер достигнут! Класс CyclicBarrier можно использовать повторно, так как он освобождает ожидающие потоки каждый раз, когда определенное количество потоков вызывает метод await (). Например, если метод main () в предыдущей программе изменить следующим образом static void main(String a r g s []) { CyclicBarrier cb = new CyclicBarrier(3, new BarAction() Запуск MyThread(cb, "A"); new MyThread(cb, "B"); new MyThread(c b , "C "); new MyThread(c b , " X ") ; new MyThread(cb, "Y"); new MyThread(c b , "Z "то результат выполнения будет таким. Запуск А В С Барьер достигнут Барьер достигнут! Как можно видеть из предыдущего примера, класс CyclicBarrier предлагает элегантное решение ранее сложной задачи. Класс Вероятно, наиболее интересным классом синхронизации является класс Exchanger, предназначенный для упрощения процесса обмена данными между двумя потоками. Работа класса Exchanger довольно проста он просто ожидает, пока два отдельных потока не вызовут его метод exchange () . Как только это произойдет, он произведет обмен данными, имеющимися у обоих потоков. В своем использовании этот механизм является одновременно и элегантными простым. Варианты применения класса Exchanger можно представить очень просто. Например, один поток подготавливает буфер для получения информации через сетевое соединение. Другой поток заполняет этот буфер информацией, получаемой при помощи соединения. Оба потока работают совместно, поэтому каждый раз, когда возникает необходимость в использовании нового буфера, осуществляется обмен данными. Класс Exchanger является обобщенным классом и имеет следующее объявление. Здесь параметр определяет тип данных для обмена. Класс Exchanger определяет единственный метод exchange () , который имеет следующие две формы exchange(V буфер throws InterruptedException |