消息隊列處理后臺任務(wù)帶來的問題
項目中經(jīng)常會有后臺運行任務(wù)的需求，比如發(fā)送郵件時，因為要連接郵件服務(wù)器，往往需要5-10秒甚至更長時間，如果能先給用戶一個成功的提示信息，然后在后臺慢慢處理發(fā)送郵件的操作，顯然會有更好的用戶體驗。

為了實現(xiàn)類似的需求，Web項目中一般的實現(xiàn)方法是使用消息隊列(Message Queue)，比如MemcacheQ，RabbitMQ等等，都是很著名的產(chǎn)品。

消息隊列說白了就是一個最簡單的先進(jìn)先出隊列，隊列的一個成員就是一段文本。正是因為消息隊列實在太簡單了，當(dāng)拿著消息隊列時，反而有點無從下手的感覺，因為這僅僅一個發(fā)送郵件的任務(wù)，就會引申出很多問題：

• 消息隊列只能存儲字符串類型的數(shù)據(jù)，如何將一個發(fā)送郵件這樣的“任務(wù)”，轉(zhuǎn)換為消息隊列中的一個“消息”?
• 消息隊列只負(fù)責(zé)數(shù)據(jù)的存放與進(jìn)出，本身不能執(zhí)行任何程序，那么我們要如何從消息隊列中一個一個取出數(shù)據(jù)，再將這些數(shù)據(jù)轉(zhuǎn)化回任務(wù)并執(zhí)行。
• 我們無法預(yù)知消息隊列何時會有數(shù)據(jù)產(chǎn)生，所以我們的任務(wù)執(zhí)行程序還需要具備監(jiān)控消息隊列的能力，也就是一個常駐后臺的守護(hù)進(jìn)程。
• 一般的Web應(yīng)用PHP都以cgi方式運行，無法常駐內(nèi)存。我們知道php還有cli模式，那么守護(hù)進(jìn)程是否能以php cli來實現(xiàn)，效率如何？
• 當(dāng)守護(hù)進(jìn)程運行時，Web應(yīng)用能否與后臺守護(hù)進(jìn)程交互，實現(xiàn)開啟/殺死進(jìn)程的功能以及獲得進(jìn)程的運行狀態(tài)？

Resque對后臺任務(wù)的設(shè)計與角色劃分

對以上這些問題，目前為止我能找到的最好答案，并不是來自php，而是來自Ruby的項目Resque，正是由于Resque清晰簡單的解決了后臺任務(wù)帶來的一系列問題，Resque的設(shè)計也被Clone到Python、php、NodeJs等語言：比如Python下的pyres以及PHP下的php-resque等等，這里有各種語言版本的Resque實現(xiàn)，而在本篇日志里，我們當(dāng)然要以PHP版本為例來說明如何用php-resque運行一個后臺任務(wù)，可能一些細(xì)節(jié)方面會與Ruby版有出入，但是本文中以php版為準(zhǔn)。

Resque是這樣解決這些問題的：

后臺任務(wù)的角色劃分
其實從上面的問題已經(jīng)可以看出，只靠一個消息隊列是無法解決所有問題的，需要新的角色介入。在Resque中，一個后臺任務(wù)被抽象為由三種角色共同完成：

• Job | 任務(wù) ： 一個Job就是一個需要在后臺完成的任務(wù)，比如本文舉例的發(fā)送郵件，就可以抽象為一個Job。在Resque中一個Job就是一個Class。
• Queue | 隊列 ： 也就是上文的消息隊列，在Resque中，隊列則是由Redis實現(xiàn)的。Resque還提供了一個簡單的隊列管理器，可以實現(xiàn)將Job插入/取出隊列等功能。
• Worker | 執(zhí)行者 ： 負(fù)責(zé)從隊列中取出Job并執(zhí)行，可以以守護(hù)進(jìn)程的方式運行在后臺。

那么基于這個劃分，一個后臺任務(wù)在Resque下的基本流程是這樣的：

• 將一個后臺任務(wù)編寫為一個獨立的Class，這個Class就是一個Job。
• 在需要使用后臺程序的地方，系統(tǒng)將Job Class的名稱以及所需參數(shù)放入隊列。
• 以命令行方式開啟一個Worker，并通過參數(shù)指定Worker所需要處理的隊列。
• Worker作為守護(hù)進(jìn)程運行，并且定時檢查隊列。
• 當(dāng)隊列中有Job時，Worker取出Job并運行，即實例化Job Class并執(zhí)行Class中的方法。

至此就可以完整的運行完一個后臺任務(wù)。

在Resque中，還有一個很重要的設(shè)計：一個Worker，可以處理一個隊列，也可以處理很多個隊列，并且可以通過增加Worker的進(jìn)程/線程數(shù)來加快隊列的執(zhí)行速度。

php-resque的安裝
需要提前說明的是，由于涉及到進(jìn)程的開辟與管理，php-resque使用了php的PCNTL函數(shù)，所以只能在Linux下運行，并且需要php編譯PCNTL函數(shù)。如果希望用Windows做同樣的工作，那么可以去找找Resque的其他語言版本，php在Windows下非常不適合做后臺任務(wù)。

以Ubuntu12.04LTS為例，Ubuntu用apt安裝的php已經(jīng)默認(rèn)編譯了PCNTL函數(shù)，無需任何配置，以下指令均為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
假設(shè)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已經(jīng)給出了簡單的例子， demo/job.php文件就是一個最簡單的Job：

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

這個Job就是在120秒后向STDOUT輸出字符Hello!

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

2.將Job插入隊列
php-resque也給出了最簡單的插入隊列實現(xiàn) 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接收到的第一個參數(shù)作為Job名稱，插入名為'default'的隊列，同時向屏幕輸出剛才插入隊列的Job Id。在終端輸入：

php demo/queue.php PHP_Job

結(jié)果可以看到屏幕上輸出：

Queued job b1f01038e5e833d24b46271a0e31f6d6

即Job已經(jīng)添加成功。注意這里的Job名稱與我們編寫的Job Class名稱保持一致：PHP_Job

3.查看Job運行情況
php-resque同樣提供了查看Job運行狀態(tài)的例子，直接運行：

php demo/check_status.php b1f01038e5e833d24b46271a0e31f6d6

可以看到輸出為：

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

我們剛才創(chuàng)建的Job狀態(tài)為1。在Resque中，一個Job有以下4種狀態(tài)：

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

因為沒有Worker運行，所以剛才創(chuàng)建的Job還是等待狀態(tài)。

4.運行Worker
這次我們直接編寫demo/resque.php：

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

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

可以直接包含Job類文件，也可以使用php的自動加載機制，指定好Job Class所在路徑并能實現(xiàn)自動加載
包含Resque的默認(rèn)Worker： bin/resque
在終端中運行：

QUEUE=default php demo/resque.php

前面的QUEUE部分是設(shè)置環(huán)境變量，我們指定當(dāng)前的Worker只負(fù)責(zé)處理default隊列。也可以使用

QUEUE=* php demo/resque.php

來處理所有隊列。

運行后輸出為

#!/usr/bin/env php
*** Starting worker

用ps指令檢查一下：

ps aux | grep resque

可以看到有一個php的守護(hù)進(jìn)程已經(jīng)在運行了

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

任務(wù)已經(jīng)運行完畢，同時屏幕上應(yīng)該可以看到輸出的Hello!

最新評論