php 使用beanstalk 消息隊列

Beanstalkd 消息隊列

一.基本信息
Beanstalkd，一個高性能、輕量級的分佈式內存隊列系統，最初設計的目的是想經過後臺異步執行耗時的任務來下降高容量Web應用系統的頁面訪問延遲，支持過有9.5 million用戶的Facebook Causes應用。後來開源，如今有PostRank大規模部署和使用，天天處理百萬級任務。Beanstalkd是典型的類Memcached設計，協議和使用方式都是一樣的風格，因此使用過memcached的用戶會以爲Beanstalkd似曾相識。

 

二 特性

Beanstalkd 是一個輕量級消息中間件，它最大特色是將本身定位爲基於管道  (tube) 和任務 (job) 的工做隊列 (work-queue)：

Beanstalkd 支持任務優先級 (priority), 延時 (delay), 超時重發 (time-to-run) 和預留 (buried), 可以很好的支持分佈式的後臺任務和定時任務處理。

它的內部實現採用 libevent, 服務器-客戶端之間用相似 memcached 的輕量級通信協議，具備有很高的性能。

儘管是內存隊列, beanstalkd 提供了 binlog 機制, 當重啓 beanstalkd 時，當前任務狀態可以從紀錄的本地 binlog 中恢復。

1）管道（tube）

管道相似於消息主題 (topic), 在一個 Beanstalkd 中能夠支持多個管道, 每一個管道都有本身的發佈者 (producer) 和消費者 (consumer). 管道之間互相不影響。

2）任務（job）

Beanstalkd 用任務 (job) 代替消息 (message) 的概念。與消息不一樣，任務有一系列狀態：

READY- 須要當即處理的任務，當延時 (DELAYED) 任務到期後會自動成爲當前任務；
DELAYED- 延遲執行的任務, 當消費者處理任務後, 能夠用將消息再次放回 DELAYED 隊列延遲執行；
RESERVED- 已經被消費者獲取, 正在執行的任務。Beanstalkd 負責檢查任務是否在 TTR(time-to-run) 內完成；
BURIED- 保留的任務: 任務不會被執行，也不會消失，除非有人把它 "踢" 回隊列；
DELETED- 消息被完全刪除。Beanstalkd 再也不維持這些消息。

3）任務優先級 (priority)

任務 (job) 能夠有 0~2^32 個優先級, 0 表明最高優先級。 beanstalkd 採用最大最小堆 (Min-max heap) 處理任務優先級排序, 任什麼時候刻調用 reserve 命令的消費者老是能拿到當前優先級最高的任務, 時間複雜度爲 O(logn).

4）延時任務 (delay)

有兩種方式能夠延時執行任務 (job): 生產者發佈任務時指定延時；或者當任務處理完畢後, 消費者再次將任務放入隊列延時執行 (RELEASE with <delay>)。

5）任務超時重發 (time-to-run)

Beanstalkd 把任務返回給消費者之後：消費者必須在預設的 TTR (time-to-run) 時間內發送 delete / release/ bury 改變任務狀態；不然 Beanstalkd 會認爲消息處理失敗，而後把任務交給另外的消費者節點執行。若是消費者預計在 TTR (time-to-run) 時間內沒法完成任務, 也能夠發送 touch 命令, 它的做用是讓 Beanstalkd 從系統時間從新計算 TTR (time-to-run).

6）任務預留 (buried)

若是任務由於某些緣由沒法執行, 消費者能夠把任務置爲 buried 狀態讓 Beanstalkd 保留這些任務。管理員能夠經過 peek buried 命令查詢被保留的任務，而且進行人工干預。簡單的, kick <n> 可以一次性把 n 條被保留的任務踢回隊列。

 

二 安裝

請看安裝教程：http://www.cnblogs.com/sien6/p/8177023.html

 

三 Beanstalkd 協議

Beanstalkd 採用類 memcached 協議, 客戶端經過文本命令與服務器交互。

這些命令能夠簡單的分紅三組:

1）生產類 - use <tube> / put <priority> <delay> <ttr> [bytes]:  

生產者用 use 選擇一個管道 (tube), 而後用 put 命令向管道發佈任務 (job).    

2）消費類  - watch <tubes> / reserve / delete <id> / release <id> <priority> <delay> / bury <id> / touch <id>

消費者用 watch 選擇多個管道 (tube), 而後用 reserve 命令獲取待執行的任務，這個命令是阻塞的。客戶端直到有任務可執行才返回。當任務處理完畢後, 消費者能夠完全刪除任務 (DELETE), 釋聽任務讓別人處理 (RELEASE), 或者保留 (BURY) 任務。

3） 維護類 - peek job / peek delayed / peek ready / peek buried / kick <n>

用於維護管道內的任務狀態, 在不改變任務狀態的條件下獲取任務。能夠用消費類命令改變這些任務的狀態。

被保留 (buried) 的任務能夠用 kick 命令 "踢" 回隊列。

 

應用：

生成者：

require_once 'beanstalk/Client.php';
$config = include 'beanstalk/config.php';
$beanstalk = new Beanstalk\Client($config);
$beanstalk->connect();
$beanstalk->useTube('u_tube'); //使用u_tabe 管道
$param = $argv;
$beanstalk->put(0,0,10,json_encode($param));

 

消費者：

require_once 'beanstalk/Client.php';
$config = include 'beanstalk/config.php';
$Beanstalk= new Beanstalk\Client($config);
$Beanstalk->connect();
$Beanstalk->useTube('u_tube');
$Beanstalk->watch('u_tube');    // 監控u_tube 通道
$Beanstalk->ignore('default');  // 取消默認通道監控

while(true){    // 持續監控
    $job = $Beanstalk->reserve();   // 取出待執行任務，並執行
//    print_r($Beanstalk->statsTube('foo'));
    if($job){
        $Beanstalk->bury($job['id'],'u_tube');  // 任務加入保留隊列
    }
//    $Beanstalk->kick(3);  // 開啓保留隊列
    $Beanstalk->delete($job['id']);     //刪除吧任務
}

 

參考：http://blog.csdn.net/black_ox/article/details/24792489

https://www.cnblogs.com/fuland/p/4245386.html