php-resque 隊列簡單使用

1、安裝 php-resque

進入項目根目錄，composer 安裝 php-resque

composer require chrisboulton/php-resque

 

2、經常使用方法

一、鏈接 redis

// setBackend($server, $database = 0)
Resque::setBackend('127.0.0.1:6379');

二、向隊列中添加工做

// enqueue($queue, $class, $args = null, $trackStatus = false)
$token = Resque::enqueue('default', 'My_Job', ['name'=>'test'], true);

三、查看工做狀態

$status = (new Resque_Job_Status($token))->get();

四、中止（移除）工做

(new Resque_Job_Status($token))->stop();

 

3、常駐任務處理隊列（示例：worker.php）

// 處理 default 隊列；也能夠填 *，表明全部隊列
$worker = new Resque_Worker('default');
// LOG_NONE 不寫日誌, LOG_NORMAL 普通，LOG_VERBOSE 詳細
$worker->logLevel = Resque_Worker::LOG_VERBOSE;
// 隊列處理時間間隔，單位：秒
$worker->work(5);

注：worker.php 要以命令行的方法執行，並長駐後臺，/usr/local/php/bin/php /xxx/xxx/worker.php

 

4、處理工做的類

class My_Job
{
    /**
     * 前置操做
     * @return void
     */
    public function setUp()
    {
        // ... Set up environment for this job
    }

    /**
     * 消費隊列
     * @return void
     */
    public function perform()
    {
        // execute a job
    }

    /**
     * 後置操做
     * @return void
     */
    public function tearDown()
    {
        // ... Remove environment for this job
    }
}

 

5、完整例子

test.php

<?php
require 'vendor/autoload.php';

/**
 * php-resque 隊列服務
 */
class Queue
{
    private $jobStatusObject = [];

    function __construct($server, $database = 0)
    {
        // 鏈接 redis
        Resque::setBackend($server);
    }

    private function getJobStatusObject($token)
    {
        if (!isset($this->jobStatusObject[$token])) {
            $this->jobStatusObject[$token] = new Resque_Job_Status($token);
        }
        
        return $this->jobStatusObject[$token];
    }

    /**
     * 入隊並返回 token
     * @param  array  $args 參數
     * @return string
     */
    public function enqueue($args = [])
    {
        // 隊列名稱、指定消費類、參數
        return Resque::enqueue('default', 'My_Job', $args, true);
    }

    /**
     * 查詢 job 的狀態
     * @param  string $token 任務token
     * @return array         狀態信息
     */
    public function getJobStatus($token)
    {
        $status = $this->getJobStatusObject($token)->get();
        $statusOptions = [
            1 => 'STATUS_WAITING',
            2 => 'STATUS_RUNNING',
            3 => 'STATUS_FAILED',
            4 => 'STATUS_COMPLETE'
        ];
        return isset($statusOptions[$status])? $statusOptions[$status]:$status;
    }

    /**
     * 清除 job
     * @param  string $token 任務token
     * @return boolean         狀態信息
     */
    public function clearJob($token)
    {
        if ($this->getJobStatusObject($token) == 'STATUS_COMPLETE')
        {
            return $this->getJobStatusObject($token)->stop();
        }
        else
        {
            error_log("非已完成任務，不可清除。token: {$token}\n", 3, 'logs.txt');
            return false;
        }
    }
}

empty($_GET) && die('url參數或數據不能爲空！！');
$token = isset($_GET['token'])? trim($_GET['token']):'';
$act   = isset($_GET['act'])? trim($_GET['act']):'';

$queue = new Queue('127.0.0.1:6379');
if (empty($act))
{
    // 任務入隊
    $token = $queue->enqueue($_GET);
    error_log("token: {$token}\n", 3, 'logs.txt');
    echo 'token: ', $token, '<hr>';
    exit;
}
elseif ($act == 'status' && !empty($token))
{
    // 查看任務狀態
    var_dump($queue->getJobStatus($token));
    exit;
}
elseif ($act == 'del' && !empty($token))
{
    // 移除任務
    var_dump($queue->clearJob($token));
    exit;
}
?>

測試url：

http://localhost/redis/queue/test.php?name=test
// token: 37efa5889472ef04108a55ba8cc448e4

http://localhost/redis/queue/test.php?act=status&token=37efa5889472ef04108a55ba8cc448e4
// string(14) "STATUS_WAITING"

http://localhost/redis/queue/test.php?act=del&token=37efa5889472ef04108a55ba8cc448e4
// bool(false)

 

worker.php

<?php
// 消費（處理）隊列

require 'vendor/autoload.php';

/**
 * 任務
 */
class My_Job
{
    /**
     * 前置操做
     * @return void
     */
    public function setUp()
    {
        // ... Set up environment for this job
    }

    /**
     * 消費隊列
     * @return void
     */
    public function perform()
    {
        // 按本身的業務處理 job
        error_log(var_export($this->args, true)."\n\n", 3, 'logs.txt');
        
        // 拋出錯誤則表示工做處理失敗，不然工做狀態將更新爲完成（STATUS_COMPLETE）
        // 注：return false 無論用，必定要 throw Exception 才行
        // throw new Exception('Unable to run this job!');
    }

    /**
     * 後置操做
     * @return void
     */
    public function tearDown()
    {
        // ... Remove environment for this job
    }
}

// 處理 default 隊列；也能夠填 *，表明全部隊列
$worker = new Resque_Worker('default');
// LOG_NONE 不寫日誌, LOG_NORMAL 普通，LOG_VERBOSE 詳細
$worker->logLevel = Resque_Worker::LOG_VERBOSE;
// 隊列處理時間間隔，單位：秒
$worker->work(5);
?>