Статьи

Планирование с Zend Job Queue

Ноябрь 5, 2018

Веб-приложения обычно следуют модели синхронной связи. Однако неинтерактивные и длительные задачи (например, генерация отчетов) лучше подходят для асинхронного выполнения. Один из способов разгрузить задачи для запуска в более позднее время или даже на другом сервере — использовать модуль очереди заданий, доступный как часть Zend Server 5 (хотя и не как часть Community Edition).

Очередь заданий позволяет планировать задания на основе времени, приоритета и даже зависимостей. Задания могут откладываться или выполняться периодически, а главное — работать параллельно! Кроме того, Zend Server сам предоставляет графический интерфейс управления для отслеживания выполнения заданий, включая их состояние, время выполнения и выходные данные.

Основным преимуществом модуля Job Queue является его способность выполнять задачи параллельно. В отличие от заданий cron, Job Queue позволяет:

  • Запуск задач теперь без ожидания их завершения (асинхронное выполнение)
  • Запуск задач один раз, но не сейчас (отложенные задания)
  • Периодическое выполнение задач (повторяющиеся задания, такие как cron, но с полным контролем над ними через PHP API — запуск, остановка, приостановка, возобновление)
  • Возможность запрашивать статус задания, обрабатывать сбои и ставить в очередь через API, а также отслеживать прошлые, текущие и ожидающие задания из графического интерфейса.

Вот некоторые примеры асинхронных задач, в которых может быть полезна очередь заданий:

  • Подготовка данных для следующего запроса (предварительный расчет)
  • Предварительное кэширование данных
  • Генерация периодических отчетов
  • Отправка электронной почты
  • Очистка временных данных или файлов
  • Общение с внешними системами
  • Фоновая синхронизация данных с мобильными устройствами

Использование очереди заданий

API очереди заданий доступен через класс ZendJobQueue . Для выполнения большинства задач вы подключитесь к серверу очереди заданий, ZendJobQueue экземпляр объекта ZendJobQueue и создав работу, используя метод createHttpJob() . 

 <?php $queue = new ZendJobQueue(); $queue->createHttpJob("http://example.com/jobs/somejob.php");

Передача пути к createHttpJob() вместо полного URL-адреса создаст задание со значением $_SERVER["HTTP_HOST"] качестве имени хоста. Следите за случаями, когда $_SERVER["HTTP_HOST"] недоступен, например, когда задание запланировано из сценария cron. 

 <?php // both calls are equivalent $queue->createHttpJob("/jobs/somejob.php"); $queue->createHttpJob("http://" . $_SERVER["HTTP_HOST"] . "/jobs/somejob.php");

Параметры задания могут быть переданы либо как часть строки запроса, либо в виде массива как второй аргумент createHttpJob() . Если в качестве второго аргумента передаются параметры, массив должен быть JSON-совместимым.

Для доступа к параметрам в getCurrentJobParams() можно использовать статический метод getCurrentJobParams() . 

 <?php $params = ZendJobQueue::getCurrentJobParams();

Дополнительные параметры задания доступны через третий аргумент для createHttpJob() . Это ассоциативный массив со следующими ключами:

  • name — необязательное имя работы
  • priority — priority задания, определенный соответствующими константами PRIORITY_LOW , PRIORITY_NORMAL , PRIORITY_HIGH и PRIORITY_URGENT
  • persistent — булево значение, чтобы сохранить работу в истории навсегда
  • predecessor — целочисленный идентификатор задания предшественника.
  • http_headers — дополнительные заголовки HTTP
  • schedule — cron-like команда планирования
  • schedule_time — время, когда задание должно быть выполнено (но оно может выполняться после этого времени в зависимости от загрузки очереди заданий)

Например, создание отложенного или повторяющегося задания будет выглядеть следующим образом: 

 <?php $params = array("p1" => 10, "p2" => "somevalue"); // process in one hour $options = array("schedule_time" => date("Ymd H:i:s", strtotime("+1 hour"))); $queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options); // process every other day at 1:05 am $options = array("schedule" => "5 1 */2 * *"); $queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options);

Неудачи (и успехи) могут быть обработаны следующим образом: 

 <?php try { doSomething(); ZendJobQueue::setCurrentJobStatus(ZendJobQueue::OK); } catch (Exception $e) { ZendJobQueue::setCurrentJobStatus(ZendJobQueue::STATUS_LOGICALLY_FAILED, $e->getMessage()); }

Расширенный пример

Допустим, ваше веб-приложение должно генерировать и отправлять по электронной почте набор отчетов по запросу пользователя.

Как правило, учитывая тот факт, что PHP не поддерживает многопроцессорную обработку и используется модель синхронного взаимодействия, пользователю придется ждать, пока все запрошенные отчеты не будут сгенерированы по одному и отправлены по электронной почте.

Использование очереди заданий в этом случае позволит пользователю не только выполнять другие действия с приложением (поскольку работа будет выполняться асинхронно), но и приложение может обрабатывать несколько отчетов одновременно (поскольку задания могут выполняться параллельно) — поэтому большинство отчетов (если не все) будут закончены примерно в одно и то же время. 

 <?php function scheduleReport($reportList, $recipient) { // list of scheduled jobs $jobList = array(); $queue = new ZendJobQueue(); // check that Job Queue is running if ($queue->isJobQueueDaemonRunning() && count($reportList) > 0) { foreach ($reportList as $report) { $params = array("type" => $report["type"], "start" => $report["start"], "length" => $report["length"], "recipient" => $recipient); $options = array("priority" => $report["priority"]); // execute the job in two minutes unless the priority is urgent if ($report["priority"] != ZendJobQueue::PRIORITY_URGENT) { $options["schedule_time"] = date("Ymd H:i:s", strtotime("+2 minutes")); } $jobID = $queue->createHttpJob("http://example.com/jobs/report.php", $params, $options); // add job ID to the list of successfully scheduled jobs if ($jobID !== false) { $jobList[] = $jobID; } } return $jobList; }

Функция scheduleReport() возвращает список идентификаторов заданий, связанных с каждым запланированным отчетом. В этой функции метод isJobQueueDaemonRunning() класса ZendJobQueue проверяет, работает ли соответствующая служба и можно ли планировать задания.

В зависимости от приоритета отчета задание может быть запланировано на немедленное выполнение или на две минуты позже (чтобы снизить нагрузку на сервер, если запрашивается много отчетов одновременно). Как только задание запланировано, его идентификатор сохраняется в списке всех успешно созданных заданий. Важно знать идентификатор работы, чтобы иметь возможность отслеживать работу или даже отменить ее.

Вот как выглядит вызов функции scheduleReport() : 

 <?php // setup request for a daily sales report and monthly financial report $reportList = array( array("type" => "sales", "start" => "2011-12-09 00:00:00", "length" => 1, "priority" => ZendJobQueue::PRIORITY_URGENT), array("type" => "finance", "start" => "2011-11-01 00:00:00", "length" => 30, "priority" => ZendJobQueue::PRIORITY_NORMAL)); // schedule reports $jobList = scheduleReport($reportList, "user@example.com"); // verify that reports were scheduled if (empty($jobList)) { // show error message }

Как упоминалось ранее, также возможно отменить запланированное задание. Как только работа выполняется, она будет закончена. Таким образом, если приоритет запроса не является срочным, у пользователя есть две минуты, чтобы отменить доставку запланированного отчета. 

 <?php function cancelReport($jobID) { $queue = new ZendJobQueue(); return $queue->removeJob($jobID); } if ($jobID !== false && cancelReport($jobID)) { // the job was successfully removed from the queue }

Функция cancelReport() просто удаляет задание из очереди запланированных отчетов, которые еще не начали выполняться.

Тогда скрипт задания выглядит так: 

 <?php function runReport() { $params = ZendJobQueue::GetParamList(); try { $report = prepareReport($params["type"], $params["start"], $params["length"]); sendReport($params["recipient"], $report); ZendJobQueue::setCurrentJobStatus(ZendJobQueue::OK); } catch (Exception $e) { ZendJobQueue::setCurrentJobStatus(ZendJobQueue::STATUS_LOGICALLY_FAILED, $e->getMessage()); } }

runReport() функция runReport() готовит и отправляет отчет на основе предоставленных параметров. После завершения состояние задания устанавливается как успешное (или логически не выполненное, если произошла ошибка).

альтернативы

Конечно, есть альтернативы Job Queue. Такие решения, как cron, pcntl_fork или даже что-то на основе Java, основанное на PHP / Java Bridge, могут или не могут быть полезны в зависимости от ваших потребностей. Также существуют более интересные инструменты, такие как Gearman , node.js и RabbitMQ .

Резюме

Хотя очередь заданий Zend Server не является единственным способом обработки очередей и параллельной обработки в PHP, это чрезвычайно простое решение, поддерживаемое «Компанией PHP», и его очень легко начать использовать. И с ростом успеха Zend PHPCloud принятие Job Queue должно стать еще более распространенным.

Если вы хотите просмотреть пример кода из этой статьи в полном объеме, вы можете найти его на GitHub.

Изображение через Варину и Джея Пателя / Shutterstock

33
Share: