Недавно был озадачен проблемой мониторинга нескольких десятков серверов (ну наверно редко кто не сталкивался с такой задачей). Проблему можно описать несколькими правилами:
- Нужно периодически пинговать сервер
- Иногда выполнять какое-либо действие с сервером (например, исполнение команды через ssh), которое засабмитил пользователь
- Действия с серверами могут нескольких типов, у каждого действия свой приоритет
- Таски (из п.1-3) нельзя выполнять одновременно для каждого сервера
- Таски могут завершаться с неудачей, например по причине отсутствия связи с сервером, нужно ждать пока связь восстановится и пытатся выполнить запланированную задачу
Первое решение, которое приходит большинству в голову — запустить для каждого сервера свой поток и там делать свои дела. Это неплохо, но что делать если в процессе мониторинга набор серверов будет меняться? Запускать и завершать потоки в процессе мониторинга как-то неэлегантно. А что делать если серверов тысяча? Иметь тысячу потоков наверно можно, но зачем это делать когда большинство времени поток простаивает и ждет своего времени для очередного пинга?
На данную проблему можно взглянуть с другой стороны и представить ее в виде классической задачи «producer-consumer». У нас есть продюсеры, которые производят таски (пинг, команда ssh) и у нас есть консьюмеры, которые эти таски исполняют. Разумеется продюсеров и консьюмеров у нас не по одну экземпляру. Решить нашу задачу «producer-consumer» в JAVA не просто, а очень просто используя классы PriorityQueue и ExecutorService.
Начнем, как обычно, с юнит-теста:
@Test
public void testOffer() {
PollServerQueue xq = new PollServerQueue();
xq.addTask(new MyTask(1, 11));
xq.addTask(new MyTask(2, 12));
xq.addTask(new MyTask(1, 13));
MyTask t1 = (MyTask)xq.poll();
assertEquals(1, t1.getServerId());
assertEquals(11, t1.getTaskId());
MyTask t2 = (MyTask)xq.poll();
assertEquals(2, t2.getServerId());
assertEquals(12, t2.getTaskId());
MyTask t3 = (MyTask)xq.poll();
assertEquals(null, t3);
xq.FinishTask(1);
MyTask t5 = (MyTask)xq.poll();
assertEquals(1, t5.getServerId());
assertEquals(13, t5.getTaskId());
}
В этом юнит-тесте мы добавили в нашу очередь три задачи типа MyTask (первый аргумент конструктора означает serverId, второй — taskId). Метод poll извлекает задачу из очереди. Если извлечь задачу не удалось (например, задачи кончились или в очереди остались задачи для серверов, для которых уже выполняются задачи) — метод poll возвращает null. Из кода видно, что завершение задачи для serverId=1 ведет к тому, что из очереди можно извлечь следующую задачу для данного сервера.
Ура! Юнит-тест написан, можно писать код. Нам потребуется:
- Структура данных (HashMap) для хранения текущих исполняемых задач для каждого сервера (currentTasks)
- Структура данных (HashMap) для хранения задачи, стоящих в очереди на исполнения. Для каждого сервера — своя очередь (waitingTasks)
- Структура данных (PriorityQueue) для последовательного опроса серверов. Необходимо, чтобы в следующий вызов poll() к нам приходила задача для другого сервера. Короче, структура типа револьвера, только пули после каждого выстрела остаются в барабане (peekOrder)
- Cтруктура (HashSet) для хранения и быстрого поиска идентификаторов серверов в револьвере, чтобы каждый раз не просматривать револьвер с первого до последнего элемента (servers)
- Простой объект для синхронизации (syncObject)
Теперь процедура извлечения таски из очереди будет простой и короткой. И хотя код получился компактным, публиковать его здесь я не вижу смысла, а отошлю вас на https://github.com/get-a-clue/PollServerQueueExample
Disclaimer: код на github'e не является законченным, в частности, в нем отсутствует возможность установки приоритетов для задач внутри очереди для каждого сервера и механизма обработки ошибок и возврата failed таски в очередь. Ну и сам код для пингования. Как говорится, меньше кода — лучше спишь. :)
Автор: getaclue
