與Zend作業隊列安排

利用Zend Job Queue進行任務調度

核心要點

• Zend Server的Job Queue模塊允許異步執行非交互式和長時間運行的任務，支持並行運行、延遲或定期執行任務，並通過GUI進行管理。
• Job Queue API可通過ZendJobQueue類訪問，該類允許創建作業、傳遞參數以及設置其他作業選項，例如優先級、持久性和調度。
• 如擴展示例中生成和發送電子郵件報告所示，Job Queue可用於改善用戶體驗和提高應用程序效率，任務可以被調度和並行執行，從而減少用戶的等待時間。
• 雖然存在其他替代方案來處理PHP中的隊列和並行處理（例如cron、pcntl_fork、Gearman、node.js和RabbitMQ），但Job Queue提供了一個簡單易用的解決方案。

大多數Web應用程序都遵循同步通信模型。但是，非交互式和長時間運行的任務（例如報告生成）更適合異步執行。一種將任務卸載到稍後時間甚至在不同服務器上運行的方法是使用Zend Server 5中提供的Job Queue模塊（但社區版中不包含）。 Job Queue允許基於時間、優先級甚至依賴關係進行作業調度。作業可以被延遲或定期執行，並且——最重要的是——可以並行運行！最重要的是，Zend Server本身提供了一個管理GUI來跟踪作業的執行，包括其狀態、執行時間和輸出。 Job Queue模塊的主要優勢在於其並行執行任務的能力。與cron作業不同，Job Queue允許：

• 立即運行任務，無需等待它們完成（異步執行）
• 運行一次任務，但不是立即運行（延遲作業）
• 定期運行任務（類似於cron的重複作業，但可以通過PHP API完全控制它們——啟動、停止、暫停、恢復）
• 通過API查詢作業狀態、處理失敗和重新排隊，以及通過GUI跟踪過去、當前和待處理的作業。

Job Queue可用於異步任務的一些示例包括：

• 為下一個請求準備數據（預計算）
• 預緩存數據
• 生成定期報告
• 發送電子郵件
• 清理臨時數據或文件
• 與外部系統通信
• 與移動設備進行後台數據同步

Job Queue 使用方法

Job Queue的API可通過ZendJobQueue類使用。要執行大多數任務，您將通過實例化ZendJobQueue對象並使用createHttpJob()方法創建作業來連接到Job Queue服務器。

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

傳遞路徑到createHttpJob()而不是完整的URL將創建一個作業，其主機名為$_SERVER["HTTP_HOST"]的值。注意$_SERVER["HTTP_HOST"]不可用的情況，例如從cron腳本調度作業時。

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

作業參數可以作為查詢字符串的一部分傳遞，也可以作為createHttpJob()的第二個參數以數組形式傳遞。如果參數作為第二個參數傳遞，則該數組必須與JSON兼容。要在作業代碼中訪問參數，可以使用getCurrentJobParams()靜態方法。

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

其他作業選項可通過createHttpJob()的第三個參數使用。它是一個關聯數組，包含以下鍵：

• name – 可選的作業名稱
• priority – 作業優先級，由相應的常量PRIORITY_LOW、PRIORITY_NORMAL、PRIORITY_HIGH和PRIORITY_URGENT定義
• persistent – 布爾值，指示是否永遠保留作業歷史記錄
• predecessor – 整數前驅作業ID
• http_headers – 附加的HTTP標頭
• schedule – cron樣式的調度命令
• schedule_time – 作業應執行的時間（但根據Job Queue的負載，它實際上可能在此時間之後運行）

例如，創建延遲作業或重複作業如下所示：

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

失敗（和成功）可以如下方式處理：

<?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);

擴展示例

假設您的Web應用程序必鬚根據用戶的請求生成和發送一組報告。通常，由於PHP不支持多處理並且使用同步通信模型，用戶必須等待所有請求的報告逐個生成並發送電子郵件。在這種情況下使用Job Queue不僅允許用戶執行應用程序的其他操作（因為工作將異步完成），而且應用程序可以同時處理多個報告（因為作業可以並行執行）——因此大多數報告（如果不是全部）將在大約相同的時間完成。

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

scheduleReport()函數返回與每個已調度報告關聯的作業標識符列表。在此函數中，ZendJobQueue類的isJobQueueDaemonRunning()方法驗證相應的服務是否正在運行以及是否可以調度作業。根據報告的優先級，作業可以立即調度運行或兩分鐘後運行（如果同時請求許多報告，則為了減少服務器的負載）。調度作業後，其ID將保存到所有成功創建的作業列表中。了解作業ID對於能夠監控作業甚至取消作業非常重要。以下是對scheduleReport()函數的調用的樣子：

<?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;
}

如前所述，也可以取消已調度的作業。但是，一旦作業正在進行中，它將完成。因此，如果請求的優先級不是緊急的，則用戶有兩分鐘的時間來取消已調度報告的交付。

<?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()函數只是從尚未開始運行的已調度報告隊列中刪除作業。然後，作業腳本如下所示：

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

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

runReport()函數最終根據提供的參數準備和發送報告。完成後，作業狀態將設置為成功（如果發生錯誤，則為邏輯失敗）。

替代方案

當然，Job Queue也有替代方案。 cron、pcntl_fork甚至通過PHP/Java Bridge的基於Java的解決方案可能值得研究，這取決於您的需求。還存在更有趣的工具，例如Gearman、node.js和RabbitMQ。

總結

雖然Zend Server的Job Queue並不是處理PHP中隊列和並行處理的唯一方法，但它是一個極其簡單的解決方案，由“The PHP Company”支持，並且非常易於使用。隨著Zend的PHPCloud越來越成功，Job Queue的採用應該會更加廣泛。如果您想查看本文中示例代碼的完整內容，您可以在GitHub上找到它。 圖片來自Varina和Jay Patel/Shutterstock

關於Zend Queue的常見問題解答（FAQ）

Zend Queue的主要功能是什麼？

Zend Queue是Zend Framework的一個組件，它為各種排隊系統提供了一個簡單的API。它允許開發人員異步創建、管理和處理數據或任務隊列。這意味著任務可以在後台執行，從而提高Web應用程序的性能和用戶體驗。它支持多個後端，例如Array、SQLite等。

Zend Queue如何提高Web應用程序的性能？

Zend Queue通過允許異步處理任務來提高Web應用程序的性能。這意味著任務可以在後台執行，而不會阻塞主執行線程。這可以顯著提高Web應用程序的響應速度，因為用戶不必等待任務完成才能繼續與應用程序交互。

如何在Zend Queue中創建一個新隊列？

要在Zend Queue中創建一個新隊列，您可以使用createQueue方法。此方法需要兩個參數：隊列的名稱和超時時間。超時參數是可選的，默認為null。這是一個示例：

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

如何在Zend Queue中向隊列添加消息？

要在Zend Queue中向隊列添加消息，您可以使用send方法。此方法需要一個參數：要添加到隊列的消息。這是一個示例：

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

如何在Zend Queue中處理來自隊列的消息？

要在Zend Queue中處理來自隊列的消息，您可以使用receive方法。此方法從隊列中檢索一組消息進行處理。這是一個示例：

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

如何在Zend Queue中刪除隊列？

要在Zend Queue中刪除隊列，您可以使用deleteQueue方法。此方法需要一個參數：要刪除的隊列的名稱。這是一個示例：

<?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);

如何檢查Zend Queue中是否存在隊列？

要在Zend Queue中檢查隊列是否存在，您可以使用isExists方法。此方法需要一個參數：要檢查的隊列的名稱。這是一個示例：

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

如何計算Zend Queue中隊列中的消息數量？

要在Zend Queue中計算隊列中的消息數量，您可以使用count方法。此方法返回隊列中的消息數量。這是一個示例：

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

如何清除Zend Queue中隊列中的所有消息？

要在Zend Queue中清除隊列中的所有消息，您可以使用purge方法。此方法將刪除隊列中的所有消息。這是一個示例：

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

如何設置Zend Queue中隊列的超時時間？

要在Zend Queue中設置隊列的超時時間，您可以使用setTimeout方法。此方法需要兩個參數：隊列的名稱和超時時間（以秒為單位）。這是一個示例：

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

請注意，以上代碼示例基於Zend_Queue，而不是文中提到的Zend Job Queue。 Zend Job Queue的API可能略有不同，需要參考Zend Server的官方文檔。

以上是與Zend作業隊列安排的詳細內容。更多資訊請關注PHP中文網其他相關文章！