php-resque的設計和使用_ZenDei技術網路在線

php-resque是一個輕量級的消息隊列，讓我們一起來瞭解一下它的設計和使用 ...

php-resque-1.2-annotated 一個 php-resque 源碼閱讀的項目，歡迎大家star

php-resque的設計

在Resque中，一個後臺任務被抽象為由三種角色共同完成：

Job | 任務：一個Job就是一個需要在後臺完成的任務，比如本文舉例的發送郵件，就可以抽象為一個Job。在Resque中一個Job就是一個Class。
Queue | 隊列：也就是上文的消息隊列，在Resque中，隊列則是由Redis實現的。Resque還提供了一個簡單的隊列管理器，可以實現將Job插入/取出隊列等功能。
Worker | 執行者：負責從隊列中取出Job並執行，可以以守護進程的方式運行在後臺。
那麼基於這個劃分，一個後臺任務在Resque下的基本流程是這樣的：

在Resque中，有一個很重要的設計：一個Worker，可以處理一個隊列，也可以處理很多個隊列，並且可以通過增加Worker的進程/線程數來加快隊列的執行速度。

流程如下：

將一個後臺任務編寫為一個獨立的Class，這個Class就是一個Job。
在需要使用後臺程式的地方，系統將Job Class的名稱以及所需參數放入隊列。
以命令行方式開啟一個Worker，並通過參數指定Worker所需要處理的隊列。
Worker作為守護進程運行，並且定時檢查隊列。
當隊列中有Job時，Worker取出Job並運行，即實例化Job Class並執行Class中的方法。

php-resque的使用

編寫一個Worker
其實php-resque已經給出了簡單的例子， demo/job.php文件就是一個最簡單的Job：

class PHP_Job
{
    public function perform()
    {
        sleep(120);
        fwrite(STDOUT, 'Hello!');
    }
}

這個Job就是在120秒後向STDOUT輸出字元Hello!

在Resque的設計中，一個Job必須存在一個perform方法，Worker則會自動運行這個方法。

將Job插入隊列
php-resque也給出了最簡單的插入隊列實現 demo/queue.php：

if(empty($argv[1])) {
    die('Specify the name of a job to add. e.g, php queue.php PHP_Job');
}

require __DIR__ . '/init.php';
date_default_timezone_set('GMT');
Resque::setBackend('127.0.0.1:6379');

$args = array(
    'time' => time(),
    'array' => array(
        'test' => 'test',
    ),
);

$jobId = Resque::enqueue('default', $argv[1], $args, true);
echo "Queued job ".$jobId."\n\n";

在這個例子中，queue.php需要以cli方式運行，將cli接收到的第一個參數作為Job名稱，插入名為'default'的隊列，同時向屏幕輸出剛纔插入隊列的Job Id。在終端輸入：

cd demo
php queue.php PHP_Job

結果可以看到屏幕上輸出：

Queued job 52f5abf5344094efc417e7ea8f1aa083

即Job已經添加成功。註意這裡的Job名稱與我們編寫的Job Class名稱保持一致：PHP_Job
在這個時候連接redis-cli，可以看到有如下三個key：

1) "resque:job:52f5abf5344094efc417e7ea8f1aa083:status"
2) "resque:queue:default"
3) "resque:queues"

分別用如下命令查看其類型：

type resque:job:52f5abf5344094efc417e7ea8f1aa083:status
type resque:queue:default
type resque:queues

其類型分別是：string/list/set
取出resque:job:52f5abf5344094efc417e7ea8f1aa083:status的內容查看：

get resque:job:52f5abf5344094efc417e7ea8f1aa083:status

其內容如下：

"{\"status\":1,\"updated\":1438095296,\"started\":1438095296}"

其中的status表示Job運行狀態，updated表示更新時間，started表示開始時間。
這裡存放的是job執行狀態的信息。
php-resque同樣也提供了查看Job運行狀態的例子，直接運行：

php check_status.php 52f5abf5344094efc417e7ea8f1aa083

可以看到輸出為：

Tracking status of 52f5abf5344094efc417e7ea8f1aa083. Press [break] to stop. 
Status of 52f5abf5344094efc417e7ea8f1aa083 is: 1

我們剛纔創建的Job狀態為1。在Resque中，一個Job有以下4種狀態：

Resque_Job_Status::STATUS_WAITING = 1; (等待)
Resque_Job_Status::STATUS_RUNNING = 2; (正在執行)
Resque_Job_Status::STATUS_FAILED = 3; (失敗)
Resque_Job_Status::STATUS_COMPLETE = 4; (結束)

取出resque:queue:default的內容查看(key中的default是在之前代碼中定義的queue的名稱)：

lrange resque:queue:default 0 -1

其內容如下：

1) "{\"class\":\"PHP_Job\",\"args\":[{\"time\":1438095296,\"array\":{\"test\":\"test\"}}],\"id\":\"52f5abf5344094efc417e7ea8f1aa083\"}"

其中的class表示Job的類，args表示Job執行時的參數，id表示Job的ID，可以根據這個ID去查詢Job執行狀態的信息。
這裡存放的是每個要執行的Job的相關信息。因為只添加了一個，所以在default的隊列中，只有一個值。

取出resque:queues的內容查看：

smembers resque:queues

其內容如下：

1) "default"

這裡存放的是所有隊列的名稱。因為只有一個，所以在queues的集合中，只有一個值。

因為沒有Worker運行，所以剛纔創建的Job還是等待狀態。
運行Worker
這次我們直接編寫demo/resque.php：

date_default_timezone_set('GMT');
require 'job.php';
require '../bin/resque';

可以看到一個Worker至少需要兩部分：

可以直接包含Job類文件，也可以使用php的自動載入機制，指定好Job Class所在路徑並能實現自動載入
包含Resque的預設Worker： bin/resque
在終端中運行：

QUEUE=default php resque.php

前面的QUEUE部分是設置環境變數，我們指定當前的Worker只負責處理default隊列。也可以使用

QUEUE=* php resque.php

來處理所有隊列。

運行後輸出為

#!/usr/bin/env php
*** Starting worker jun-Ubuntu:23437:*

用ps指令檢查一下：

ps aux | grep resque

可以看到有一個php的守護進程已經在運行了

jun      23437  1.0  0.3 314148 14884 pts/16   S+   23:23   0:00 php resque.php

在這個時候再連接到redis-cli，查看key，可以看到如下key：

1) "resque:job:52f5abf5344094efc417e7ea8f1aa083:status"
2) "resque:workers"
3) "resque:queues"
4) "resque:worker:jun-Ubuntu:25122:*:started"
5) "resque:worker:jun-Ubuntu:25122:*"

分別查看新增的key是什麼類型：

type resque:workers
type resque:worker:jun-Ubuntu:25122:*:started
type resque:worker:jun-Ubuntu:25122:*

其類型分別是set/string/string
分別取出其內容，命令就不再寫了，請參考之前的內容
resque:workers中的內容如下：

1) "jun-Ubuntu:25122:*"

這裡存放的是所有worker的進程ID。因為只有一個，所以在workers的集合中，只有一個值。
resque:worker:jun-Ubuntu:25122::started中的內容如下(key中的jun-Ubuntu:25122:是worker的host+進程ID+queue的名稱)：

"Tue Jul 28 15:29:37 GMT 2015"

這裡存放的是Job啟動的時間。
resque:worker:jun-Ubuntu:25122:中的內容如下(key中的jun-Ubuntu:25122:是worker的host+進程ID+queue的名稱)：

"{\"queue\":\"default\",\"run_at\":\"Tue Jul 28 15:29:37 GMT 2015\",\"payload\":{\"class\":\"PHP_Job\",\"args\":[{\"time\":1438097296,\"array\":{\"test\":\"test\"}}],\"id\":\"52f5abf5344094efc417e7ea8f1aa083\"}}"

這裡存放的是這個worker當前執行的Job的所有信息。
於此同時，resque:job:52f5abf5344094efc417e7ea8f1aa083:status中的內容變為如下內容：

"{\"status\":2,\"updated\":1438097377}"

狀態變為2(正在執行)了。
也可以使用之前的檢查Job指令

php check_status.php 52f5abf5344094efc417e7ea8f1aa083

2分鐘後再連接到redis-cli上去查看key，可以看到如下key：

1) "resque:job:52f5abf5344094efc417e7ea8f1aa083:status"
2) "resque:workers"
3) "resque:stat:processed"
4) "resque:stat:processed:jun-Ubuntu:25122:*"
5) "resque:queues"
6) "resque:worker:jun-Ubuntu:25122:*:started"

其中的resque:stat:processed和resque:stat:processed:jun-Ubuntu:25122:都是string類型，分別表示所有worker執行job成功的個數和worker為jun-Ubuntu:25122:的執行job成功的個數。
這個時候再去查看以下resque:job:52f5abf5344094efc417e7ea8f1aa083:status的內容，發現狀態已經變為4(結束)了。
也可以使用之前的檢查Job指令查看，其結果如下：

Status of 52f5abf5344094efc417e7ea8f1aa083 is: 4

這表示任務已經運行完畢，同時屏幕上應該可以看到輸出的Hello!

至此我們已經成功的完成了一個最簡單的Resque實例的全部演示，更複雜的情況以及遺留的問題會在下一次的日誌中說明。

總結一下Redis中的key對應的內容及其含義如下：

resque:workers (set) - 存放所有的worker，每一個值都是{worker host}:{進程ID}:{queue的名稱}
resque:queues (set) - 存放所有queue的名稱
resque:queue:default (list) - 保存這個隊列中等待執行的Job
resque:job:52f5abf5344094efc417e7ea8f1aa083:status (string) - 存放job的狀態信息
resque:stat:processed (string) - 保存所有worker執行job成功的個數
resque:stat:processed:jun-Ubuntu:25122:* (string) - 保存一個worker執行job成功的個數
resque:worker:jun-Ubuntu:25122:*:started (string) - 保存一個worker的啟動時間
resque:worker:jun-Ubuntu:25122:* (string) - 保存一個worker當前執行的Job的所有信息

參考摘錄：
PHP的輕量消息隊列php-resque使用說明