消息隊列處理後台任務帶來的問題
項目中經常會有後台運行任務的需求,比如發送郵件時,因為要連接郵件服務器,往往需要5-10秒甚至更長時間,如果能先給用戶一個成功的提示信息,然後在後台慢慢處理發送郵件的操作,顯然會有更好的用戶體驗。
為了實現類似的需求,Web項目中一般的實現方法是使用消息隊列(Message Queue),比如MemcacheQ,RabbitMQ等等,都是很著名的產品。
消息隊列說白了就是一個最簡單的先進先出隊列,隊列的一個成員就是一段文本。正是因為消息隊列實在太簡單了,當拿著消息隊列時,反而有點無從下手的感覺,因為這僅僅一個發送郵件的任務,就會引申出很多問題:
Resque對後台任務的設計與角色劃分
對以上這些問題,目前為止我能找到的最好答案,並不是來自php,而是來自Ruby的項目Resque,正是由於Resque清晰簡單的解決了後台任務帶來的一系列問題,Resque的設計也被Clone到Python、php、NodeJs等語言:比如Python下的pyres以及PHP下的php-resque等等,這裡有各種語言版本的Resque實現,而在本篇日志裡,我們當然要以PHP版本為例來說明如何用php-resque運行一個後台任務,可能一些細節方面會與Ruby版有出入,但是本文中以php版為准。
Resque是這樣解決這些問題的:
後台任務的角色劃分
其實從上面的問題已經可以看出,只靠一個消息隊列是無法解決所有問題的,需要新的角色介入。在Resque中,一個後台任務被抽象為由三種角色共同完成:
那麼基於這個劃分,一個後台任務在Resque下的基本流程是這樣的:
至此就可以完整的運行完一個後台任務。
在Resque中,還有一個很重要的設計:一個Worker,可以處理一個隊列,也可以處理很多個隊列,並且可以通過增加Worker的進程/線程數來加快隊列的執行速度。
php-resque的安裝
需要提前說明的是,由於涉及到進程的開辟與管理,php-resque使用了php的PCNTL函數,所以只能在Linux下運行,並且需要php編譯PCNTL函數。如果希望用Windows做同樣的工作,那麼可以去找找Resque的其他語言版本,php在Windows下非常不適合做後台任務。
以Ubuntu12.04LTS為例,Ubuntu用apt安裝的php已經默認編譯了PCNTL函數,無需任何配置,以下指令均為root帳號
安裝Redis
apt-get install redis-server
安裝Composer
apt-get install curl cd /usr/local/bin curl -s http://getcomposer.org/installer | php chmod a+x composer.phar alias composer='/usr/local/bin/composer.phar'
使用Composer安裝php-resque
假設web目錄在/opt/htdocs
apt-get install git git-core cd /opt/htdocs git clone git://github.com/chrisboulton/php-resque.git cd php-resque composer install
php-resque的使用
1.編寫一個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則會自動運行這個方法。
2.將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。在終端輸入:
php demo/queue.php PHP_Job
結果可以看到屏幕上輸出:
Queued job b1f01038e5e833d24b46271a0e31f6d6
即Job已經添加成功。注意這裡的Job名稱與我們編寫的Job Class名稱保持一致:PHP_Job
3.查看Job運行情況
php-resque同樣提供了查看Job運行狀態的例子,直接運行:
php demo/check_status.php b1f01038e5e833d24b46271a0e31f6d6
可以看到輸出為:
Tracking status of b1f01038e5e833d24b46271a0e31f6d6. Press [break] to stop. Status of b1f01038e5e833d24b46271a0e31f6d6 is: 1
我們剛才創建的Job狀態為1。在Resque中,一個Job有以下4種狀態:
因為沒有Worker運行,所以剛才創建的Job還是等待狀態。
4.運行Worker
這次我們直接編寫demo/resque.php:
<?php date_default_timezone_set('GMT'); require 'job.php'; require '../bin/resque';
可以看到一個Worker至少需要兩部分:
可以直接包含Job類文件,也可以使用php的自動加載機制,指定好Job Class所在路徑並能實現自動加載
包含Resque的默認Worker: bin/resque
在終端中運行:
QUEUE=default php demo/resque.php
前面的QUEUE部分是設置環境變量,我們指定當前的Worker只負責處理default隊列。也可以使用
QUEUE=* php demo/resque.php
來處理所有隊列。
運行後輸出為
#!/usr/bin/env php *** Starting worker
用ps指令檢查一下:
ps aux | grep resque
可以看到有一個php的守護進程已經在運行了
1000 4607 0.0 0.1 74816 11612 pts/3 S+ 14:52 0:00 php demo/resque.php
再使用之前的檢查Job指令
php demo/check_status.php b1f01038e5e833d24b46271a0e31f6d6
2分鐘後可以看到
Status of b1f01038e5e833d24b46271a0e31f6d6 is: 4
任務已經運行完畢,同時屏幕上應該可以看到輸出的Hello!