Scheduling with Zend Job Queue

Use Zend Job Queue for task scheduling

Core points

• Zend Server's Job Queue module allows for asynchronous execution of non-interactive and long-running tasks, supports parallel operation, delayed or regular execution of tasks, and is managed through the GUI.
• The Job Queue API is accessible through the ZendJobQueue class, which allows creating jobs, passing parameters, and setting other job options such as priority, persistence, and scheduling.
• As shown in the extension example, Job Queue can be used to improve user experience and improve application efficiency, and tasks can be scheduled and executed in parallel, reducing user waiting time.
• While there are other alternatives to handle queues and parallel processing in PHP (e.g. cron, pcntl_fork, Gearman, node.js, and RabbitMQ), Job Queue provides a simple and easy-to-use solution.

Most web applications follow a synchronous communication model. However, non-interactive and long-running tasks (such as report generation) are more suitable for asynchronous execution. One way to offload tasks to a later time or even run on different servers is to use the Job Queue module provided in Zend Server 5 (but not included in the Community Edition). Job Queue allows job scheduling based on time, priority, and even dependencies. Jobs can be delayed or executed regularly and—most importantly—can be run in parallel! Most importantly, Zend Server itself provides a management GUI to track job execution, including its status, execution time, and output. The main advantage of the Job Queue module lies in its ability to execute tasks in parallel. Unlike cron jobs, Job Queue allows:

• Run tasks now without waiting for them to complete (asynchronously)
• Run the task once, but not immediately (delayed job)
• Run tasks regularly (similar to repeated jobs in cron, but they can be fully controlled through the PHP API - start, stop, pause, resume)
• Query job status through the API, process failures and requeuing, and track past, current and pending jobs through the GUI.

Some examples of what Job Queue can be used for asynchronous tasks include:

• Prepare data for the next request (precalculated)
• Precache data
• Generate periodic reports
• Send an email
• Clean temporary data or files
• Communicate with external systems
• Synchronize backend data with mobile devices

How to use Job Queue

Job Queue's API can be used through the ZendJobQueue class. To perform most tasks, you will connect to the Job Queue server by instantiating the ZendJobQueue object and creating a job using the createHttpJob() method.

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

Passing the path to createHttpJob() instead of the full URL will create a job with the value of hostname $_SERVER["HTTP_HOST"]. Note that $_SERVER["HTTP_HOST"] is not available, such as when scheduling a job from a cron script.

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

Job parameters can be passed as part of the query string or as the second parameter of createHttpJob() as an array. If the argument is passed as the second argument, the array must be JSON compatible. To access parameters in the job code, you can use the getCurrentJobParams() static method.

<?php // 这两个调用是等效的
$queue->createHttpJob("/jobs/somejob.php");
$queue->createHttpJob("http://" . $_SERVER["HTTP_HOST"] . "/jobs/somejob.php");

Other job options can be used through the third parameter of createHttpJob(). It is an associative array containing the following keys:

• name – optional job name
• priority – Job priority, defined by the corresponding constants PRIORITY_LOW, PRIORITY_NORMAL, PRIORITY_HIGH and PRIORITY_URGENT
• persistent - Boolean value indicating whether job history is always preserved
• predecessor – integer predecessor job ID
• http_headers – Attached HTTP header
• schedule - cron-style schedule command
• schedule_time – Time the job should be executed (but according to the load of the Job Queue it may actually run after this time)

For example, creating a delayed job or a duplicate job looks like this:

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

Failed (and successful) can be handled as follows:

<?php $params = array("p1" => 10, "p2" => "somevalue");

// 一小时后处理
$options = array("schedule_time" => date("Y-m-d H:i:s", strtotime("+1 hour")));
$queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options);

// 每隔一天凌晨1:05处理
$options = array("schedule" => "5 1 */2 * *");
$queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options);

Extended Example

Suppose your web application must generate and send a set of reports based on the user's request. Typically, since PHP does not support multiprocessing and uses a synchronous communication model, users must wait for all requested reports to be generated one by one and send emails. Using Job Queue in this case not only allows the user to perform other operations of the application (because the work will be done asynchronously), but the application can process multiple reports simultaneously (because the job can be executed in parallel)—so most reports, if not all, will be completed at about the same time.

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

The scheduleReport() function returns a list of job identifiers associated with each scheduled report. In this function, the isJobQueueDaemonRunning() method of the ZendJobQueue class verifies whether the corresponding service is running and whether the job can be scheduled. Depending on the priority of the report, the job can be scheduled to run immediately or after two minutes (in order to reduce the load on the server if many reports are requested at the same time). After the job is scheduled, its ID is saved to the list of all successfully created jobs. Understanding the job ID is very important to be able to monitor jobs or even cancel jobs. Here is what the call to the scheduleReport() function looks like:

<?php function scheduleReport($reportList, $recipient) {
    // 已调度作业列表
    $jobList = array();

    $queue = new ZendJobQueue();

    // 检查Job Queue是否正在运行
    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"]);

            // 除非优先级为紧急，否则在两分钟内执行作业
            if ($report["priority"] != ZendJobQueue::PRIORITY_URGENT) {
                $options["schedule_time"] = date("Y-m-d H:i:s", strtotime("+2 minutes"));
            }

            $jobID = $queue->createHttpJob("http://example.com/jobs/report.php", $params, $options);

            // 将作业ID添加到已成功调度作业的列表中
            if ($jobID !== false) {
                $jobList[] = $jobID;
            }
        }
    }

    return $jobList;
}

As mentioned earlier, scheduled jobs can also be cancelled. However, once the job is in progress, it will be completed. Therefore, if the requested priority is not urgent, the user has two minutes to cancel the delivery of scheduled reports.

The

<?php // 设置每日销售报告和每月财务报告的请求
$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));

// 调度报告
$jobList = scheduleReport($reportList, "user@example.com");

// 验证报告是否已调度
if (empty($jobList)) {
    // 显示错误消息
}

cancelReport() function simply deletes the job from the scheduled report queue that has not started running. Then, the job script looks like this:

The

<?php function cancelReport($jobID) {
    $queue = new ZendJobQueue();
    return $queue->removeJob($jobID);
}

if ($jobID !== false && cancelReport($jobID)) {
    // 作业已成功从队列中删除
}

runReport() function finally prepares and sends reports based on the provided parameters. After completion, the job status is set to Success (logical failure if an error occurs).

Alternatives

Of course, there are alternatives to Job Queue. cron, pcntl_fork and even Java-based solutions via PHP/Java Bridge may be worth looking at, depending on your needs. More interesting tools exist, such as Gearman, node.js, and RabbitMQ.

Summary

While Zend Server's Job Queue isn't the only way to handle queues and parallel processing in PHP, it's an extremely simple solution, supported by "The PHP Company" and is very easy to use. With Zend's PHPCloud becoming more successful, Job Queue adoption should be more extensive. If you want to see the full content of the sample code in this article, you can find it on GitHub. Pictures from Varina and Jay Patel/Shutterstock

FAQs about Zend Queue (FAQ)

What are the main functions of Zend Queue?

Zend Queue is a component of the Zend Framework that provides a simple API for various queueing systems. It allows developers to create, manage, and process data or task queues asynchronously. This means that tasks can be executed in the background, thereby improving the performance and user experience of the web application. It supports multiple backends, such as Array, SQLite, etc.

How does Zend Queue improve the performance of web applications?

Zend Queue improves the performance of web applications by allowing asynchronous processing of tasks. This means that the task can be executed in the background without blocking the main execution thread. This can significantly improve the responsiveness of web applications, as users do not have to wait for the task to complete before continuing to interact with the application.

How to create a new queue in Zend Queue?

To create a new queue in Zend Queue, you can use the createQueue method. This method requires two parameters: the name of the queue and the timeout. The timeout parameter is optional and defaults to null. Here is an example:

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

How to add a message to a queue in Zend Queue?

To add messages to a queue in Zend Queue, you can use the send method. This method requires a parameter: the message to be added to the queue. Here is an example:

<?php // 这两个调用是等效的
$queue->createHttpJob("/jobs/somejob.php");
$queue->createHttpJob("http://" . $_SERVER["HTTP_HOST"] . "/jobs/somejob.php");

How to process messages from queues in Zend Queue?

To process messages from queues in Zend Queue, you can use the receive method. This method retrieves a set of messages from the queue for processing. Here is an example:

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

How to delete a queue in Zend Queue?

To delete a queue in Zend Queue, you can use the deleteQueue method. This method requires a parameter: the name of the queue to be deleted. Here is an example:

<?php $params = array("p1" => 10, "p2" => "somevalue");

// 一小时后处理
$options = array("schedule_time" => date("Y-m-d H:i:s", strtotime("+1 hour")));
$queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options);

// 每隔一天凌晨1:05处理
$options = array("schedule" => "5 1 */2 * *");
$queue->createHttpJob("http://example.com/jobs/somejob.php", $params, $options);

How to check if a queue exists in Zend Queue?

To check if the queue exists in Zend Queue, you can use the isExists method. This method requires one parameter: the name of the queue to be checked. Here is an example:

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

How to calculate the number of messages in a queue in a Zend Queue?

To calculate the number of messages in a queue in a Zend Queue, you can use the count method. This method returns the number of messages in the queue. Here is an example:

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

How to clear all messages in queues in Zend Queue?

To clear all messages in the queue in Zend Queue, you can use the purge method. This method deletes all messages in the queue. Here is an example:

<?php // 这两个调用是等效的
$queue->createHttpJob("/jobs/somejob.php");
$queue->createHttpJob("http://" . $_SERVER["HTTP_HOST"] . "/jobs/somejob.php");

How to set the timeout time of queues in Zend Queue?

To set the timeout time for a queue in Zend Queue, you can use the setTimeout method. This method requires two parameters: the name of the queue and the timeout in seconds. Here is an example:

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

Please note that the above code example is based on Zend_Queue, not the Zend Job Queue mentioned in the article. The API of Zend Job Queue may be slightly different, so you need to refer to the official documentation of Zend Server.

The above is the detailed content of Scheduling with Zend Job Queue. For more information, please follow other related articles on the PHP Chinese website!