Фактически класс PriorityBlockingQueue представляет приоритетную очередь с блокирующими операциями выборки. В следующем примере объектами в приоритетной очереди являются задачи, покидающие очередь в порядке приоритетов. Для определения этого порядка в класс PrioritizedTask включается поле priority:
//: concurrency/PriorityBlockingQueueDemo.java
import java.util.concurrent.*;
import java.util.*;
import static net.mindview.util.Print.*;
class PrioritizedTask implements
Runnable, Comparable<PrioritizedTask> {
private Random rand = new Random(47);
private static int counter = 0;
private final int id = counter++;
private final int priority;
protected static List<PrioritizedTask> sequence =
new ArrayList<PrioritizedTask>();
public PrioritizedTask(int priority) {
this.priority = priority;
sequence.add(this);
}
public int compareTo(PrioritizedTask arg) {
return priority < arg.priority ? 1 :
(priority > arg.priority ? -1 : 0);
}
public void run() {
try {
TimeUnit.MILLISECONDS.sleep(rand.nextInt(250));
} catch(InterruptedException e) {
// Приемлемый вариант выхода
}
print(this);
}
public String toString() {
return String.format("[%1$-3d]", priority) +
" Task " + id;
}
public String summary() {
return "(" + id + ":" + priority + ")";
}
public static class EndSentinel extends PrioritizedTask {
private ExecutorService exec;
public EndSentinel(ExecutorService e) {
super(-1); // Минимальный приоритет в этой программе
exec = e;
}
public void run() {
int count = 0;
for(PrioritizedTask pt : sequence) {
printnb(pt.summary());
if(++count % 5 == 0)
print();
}
print();
print(this + " Calling shutdownNow()");
exec.shutdownNow();
}
}
}
class PrioritizedTaskProducer implements Runnable {
private Random rand = new Random(47);
private Queue<Runnable> queue;
private ExecutorService exec;
public PrioritizedTaskProducer(
Queue<Runnable> q, ExecutorService e) {
queue = q;
exec = e; // Используется для EndSentinel
}
public void run() {
// Неограниченная очередь без блокировки.
// Быстрое заполнение случайными приоритетами:
for(int i = 0; i < 20; i++) {
queue.add(new PrioritizedTask(rand.nextInt(10)));
Thread.yield();
}
// Добавление высокоприоритетных задач:
try {
for(int i = 0; i < 10; i++) {
TimeUnit.MILLISECONDS.sleep(250);
queue.add(new PrioritizedTask(10));
}
// Добавление заданий, начиная с наименьших приоритетов:
for(int i = 0; i < 10; i++)
queue.add(new PrioritizedTask(i));
// Предохранитель для остановки всех задач::
queue.add(new PrioritizedTask.EndSentinel(exec));
} catch(InterruptedException e) {
// Приемлемый вариант выхода
}
print("Finished PrioritizedTaskProducer");
}
}
class PrioritizedTaskConsumer implements Runnable {
private PriorityBlockingQueue<Runnable> q;
public PrioritizedTaskConsumer(
PriorityBlockingQueue<Runnable> q) {
this.q = q;
}
public void run() {
try {
while(!Thread.interrupted())
// Использование текущего потока для запуска задачи:
q.take().run();
} catch(InterruptedException e) {
// Приемлемый вариант выхода
}
print("Finished PrioritizedTaskConsumer");
}
}
public class PriorityBlockingQueueDemo {
public static void main(String[] args) throws Exception {
Как и в предыдущем примере, последовательность создания объектов PrioritizedTask сохраняется в контейнере List sequence для сравнения с фактическим порядком выполнения. Метод run() делает небольшую паузу, а затем выводит информацию об объекте, а предохранитель EndSentinel выполняет те же функции, что и прежде.
PrioritizedTaskProducer и PrioritizedTaskConsumer связываются друг с другом через PriorityBlockingQueue. Так как сам блокирующий характер очереди обеспечивает всю необходимую синхронизацию, явная синхронизация не нужна — при чтении вам не нужно думать о том, содержит ли очередь элементы, потому что при отсутствии элементов очередь просто заблокирует читающую сторону.
Управление оранжереей на базе ScheduledExecutor
В главе 10 была представлена система управления гипотетической оранжереей, которая включала (отключала) различные устройства и регулировала их работу. Происходящее можно преобразовать в контекст многозадачности: каждое событие оранжереи представляет собой задачу, запускаемую в заранее заданное время. Класс ScheduledThreadPoolExecutor предоставляет именно тот сервис, который необходим для решения задачи. Используя методы schedule() (однократный запуск задачи) или scheduleAtFixedRate() (повторение задачи с постоянным промежутком), мы создаем объекты Runnable, которые должны запуститься в положенное время. Сравните это решение с тем, что приведено в главе 10, и посмотрите, насколько оно упрощается благодаря готовой функциональностиScheduledThreadPoolExecutor:
//: concurrency/GreenhouseScheduler.java
// Новая реализация іnnerclasses/GreenhouseController.java
// с использованием ScheduledThreadPoolExecutor.
// {Args: 5000}
import java.util.concurrent.*;
import java.util.*;
public class GreenhouseScheduler {
private volatile boolean light = false;
private volatile boolean water = false;
private String thermostat = "Day";
public synchronized String getThermostat() {
return thermostat;
}
public synchronized void setThermostat(String value) {
thermostat = value;
}
ScheduledThreadPoolExecutor scheduler =
new ScheduledThreadPoolExecutor(10);
public void schedule(Runnable event, long delay) {
