GreenhouseScheduler gh = new GreenhouseScheduler();
gh.schedule(gh.new Terminate(), 5000);
// Former "Restart" class not necessary:
gh.repeat(gh.new Bell(), 0, 1000);
gh.repeat(gh.new ThermostatNight(), 0, 2000);
gh.repeat(gh.new LightOn(), 0, 200);
gh.repeat(gh.new LightOff(), 0, 400);
gh.repeat(gh.new WaterOn(), 0, 600);
gh.repeat(gh.new WaterOff(), 0, 800);
gh.repeat(gh.new ThermostatDay(), 0, 1400);
gh.repeat(gh.new CollectData(), 500, 500);
}
} /* (Execute to see output) *///:~
В этой версии, помимо реорганизации кода, добавляется новая возможность: сбор данных о температуре и влажности в оранжерее. Объект DataPoint содержит и выводит одну точку данных, а запланированная задача CollectData генерирует данные имитации и включает их вList<DataPoint> при каждом запуске.
Обратите внимание на ключевые слова volatile и synchronized; благодаря им задачи не мешают работе друг друга. Все методы контейнера List с элементами DataPoint синхронизируются с использованием метода synchronizedList() библиотеки java.util.Соllectiоns при создании List.
Семафоры
При обычной блокировке доступ к ресурсу в любой момент времени разрешается только одной задаче. Семафор со счетчиком позволяет n задачам одновременно обращаться к ресурсу. Можно считать, что семафор «выдает разрешения» на использование ресурса, хотя никаких реальных объектов разрешений в этой схеме нет. В качестве примера рассмотрим концепцию пула объектов: объекты, входящие в пул, «выдаются» для использования, а затем снова возвращаются в пул после того, как пользователь закончит работу с ними. Эта функциональность инкапсулируется в параметризованном классе:
//: concurrency/Pool.java
// Использование Semaphore в Pool ограничивает количество
// задач, которые могут использовать ресурс.
import java.util.concurrent.*;
import java.util.*;
public class Pool<T> {
private int size;
private List<T> items = new ArrayList<T>();
private volatile boolean[] checkedOut;
private Semaphore available;
public Pool(Class<T> classObject, int size) {
this.size = size;
checkedOut = new boolean[size];
available = new Semaphore(size, true);
// Заполнение пула объектами :
for(int i = 0; i < size; ++i)
try {
// Предполагается наличие конструктора по умолчанию:
items.add(classObject.newInstance());
} catch(Exception e) {
throw new RuntimeException(e);
}
}
public T checkOut() throws InterruptedException {
available.acquire();
return getItem();
}
public void checkIn(T x) {
if(releaseItem(x))
available.release();
}
private synchronized T getItem() {
for(int i = 0; i < size; ++i)
if(!checkedOut[i]) {
checkedOut[i] = true;
return items.get(i);
}
return null; // Семафор предотвращает переход в зту точку
if(index == -1) return false; // Отсутствует в списке
if(checkedOut[index]) {
checkedOut[index] = false;
return true;
}
return false; // He был освобожден
}
}
В этой упрощенной форме конструктор использует newInstance() для заполнения пула объектами. Если вам понадобится новый объект, вызовите checkOut(); завершив работу с объектом, передайте его checkIn().
Логический массив checkedOut отслеживает выданные объекты. Для управления его содержимым используются методы getItem() и releaseItem(). В свою очередь, эти методы защищены семафором available, поэтому в checkOut() семафор available блокирует дальнейшее выполнение при отсутствии семафорных разрешений (то есть при отсутствии объектов в пуле). Метод checkIn() проверяет действительность возвращаемого объекта, и, если объект действителен, разрешение возвращается семафору.
Для примера мы воспользуемся классом Fat. Создание объектов этого класса является высокозатратной операцией, а на выполнение конструктора уходит много времени:
//: concurrency/Fat.java
// Объекты, создание которых занимает много времени.
Мы создадим пул объектов Fat, чтобы свести к минимуму затраты на выполнение конструктора. Для тестирования класса Pool будет создана задача, которая забирает объекты Fat для использования, удерживает их в течение некоторого времени, а затем возвращает обратно:
//: concurrency/SemaphoreDemo.java
// Тестирование класса Pool
import java.util.concurrent.*;
import java.util.*;
import static net.mindview.util.Print.*;
// Задача для получения ресурса из пула:
class CheckoutTask<T> implements Runnable {
private static int counter = 0;
private final int id = counter++;
private Pool<T> pool;
public CheckoutTask(Pool<T> pool) {
this.pool = pool;
}
public void run() {
try {
T item = pool.checkOut();
print(this + "checked out " + item);
TimeUnit.SECONDS.sleep(1);
print(this +"checking in " + item);
pool.checkIn(item);
} catch(InterruptedException e) {
// Приемлемый способ завершения
}
}
public String toString() {
return "CheckoutTask " + id + " ";
}
}
public class SemaphoreDemo {
final static int SIZE = 25;
public static void main(String[] args) throws Exception {
blocked.cancel(true); // Выход из заблокированного вызова
print("Checking in objects in " + list);
for(Fat f : list)
pool.checkIn(f);
for(Fat f : list)
pool.checkIn(f); // Второй вызов checkIn игнорируется
exec.shutdown();
}
} /* (Execute to see output) *///:~
В коде main() создается объект Pool для хранения объектов Fat, после чего группа задач CheckoutTask начинает использовать Pool. Далее поток main() начинает выдавать объекты Fat, не возвращая их обратно. После того как все объекты пула будут выданы, семафор запрещает дальнейшие выдачи. Метод run() блокируется, и через две секунды вызывается метод cancel(). Лишние возвраты Pool игнорирует.
Exchanger
Класс Exchanger представляет собой «барьер», который меняет местами объекты двух задач. На подходе к барьеру задачи имеют один объект, а на выходе — объект, ранее удерживавшийся другой задачей. Объекты Exchanger обычно используются в тех ситуациях, когда одна задача создает высокозатратные объекты, а другая задача эти объекты потребляет.
Чтобы опробовать на практике класс Exchanger, мы создадим задачу-поставщика и задачу-потребителя, которые благодаря параметризации и генераторам могут работать с объектами любого типа. Затем эти параметризованные задачи будут применены к классу Fat.ExchangerProducer и ExchangerConsumer меняют местами List<T>; при вызове метода Exchanger.exchange() вызов блокируется до тех пор, пока парная задача не вызовет свой метод exchange(), после чего оба метода exchange() завершаются, а контейнеры List<T>меняются местами:
Exchanger<List<Fat>> xc = new Exchanger<List<Fat>>();
List<Fat>
producerList = new CopyOnWriteArrayList<Fat>(),
consumerList = new CopyOnWriteArrayList<Fat>();
exec.execute(new ExchangerProducer<Fat>(xc,
BasicGenerator.create(Fat.class), producerList));
exec.execute(
new ExchangerConsumer<Fat>(xc,consumerList));
TimeUnit.SECONDS.sleep(delay);
exec.shutdownNow();
}
}
<spoiler text="Output:"> (Sample)
Final value: Fat id: 29999
</spoiler> В методе main() для обеих задач создается один объект Exchanger, а для перестановки создаются два контейнера CopyOnWriteArrayList. Эта разновидность List нормально переносит вызов метода remove() при перемещении по списку, не выдавая исключенияConcurrentModificationException.
ExchangerProducer заполняет список, а затем меняет местами заполненный список с пустым, передаваемым от ExchangerConsumer. Благодаря Exchanger заполнение списка происходит одновременно с использованием уже заполненного списка.
Моделирование
Одна из самых интересных областей применения параллельных вычислений — всевозможные имитации и моделирование. Каждый компонент модели оформляется в виде отдельной задачи, что значительно упрощает его программирование.
Примеры HorseRace.java и GreenhouseScheduler.java, приведенные ранее, тоже можно считать своего рода имитаторами.
