初識 beanstalkd

簡述

Beanstalkd 是一個輕量級的內存型隊列，利用了和 Memcache 相似的協議。依賴 libevent 單線程事件分發機制, 能夠部署多個實例，可是高併發支持仍是不太友好；

管道

即有名稱的任務隊列，一個服務器有一個或者多個管道，用來儲存統一類型的 job。每一個管道由一個就緒隊列與延遲隊列組成。每一個job全部的狀態遷移在一個管道中完成。消費者能夠監控感興趣的管道，經過發送 watch 指令。消費者也能夠取消監控 tube，經過發送 ignore 命令。經過 list 命令返回全部監控的管道，當客戶端預訂一個job，此 job 可能來自任何一個它監控的管道。

當一個客戶端鏈接上服務器時，客戶端監控的tube 默認爲 default，若是客戶端提交 job 時，沒有使用 use 命令，那麼這些 job 就存於名爲 default 的 tube 中。

管道按需求建立，不管他們在地方被引用到。若是一個管道變爲空和沒有任何客戶端引用，它將會被自動刪除。

Job

任務在隊裏之中被稱做 Job. 一個 Job 在 Beanstalkd 中有如下的生命週期：

• put 將一個任務放置進 tube 中
• deayed 這個任務如今再等待中，須要若干秒才能準備完畢【延遲隊列】
• ready 這個任務已經準備好了，能夠消費了。全部的消費都是要從取 ready 狀態的 job
• reserved 這個任務已經被消費者消費
• release 這個 job 執行失敗了，把它放進 ready 狀態隊列中。讓其餘隊列執行
• bury 這個 job 執行失敗了，但不但願其餘隊列執行，先把它埋起來

安裝

在 Centos7 上經過命令 yum -y install beanstalkd --enablerepo=epel;

其餘系統的安裝在 官網 上查看

查看當前版本號

beanstalkd -v

啓動

常見啓動以下：

beanstalkd -l 0.0.0.0 -p 11300 -b /home/software/binstalkd/binlogs

啓動後對 beanstalkd 的操做可使用 telnet,好比 telnet 127.0.0.1 11300。而後即可以執行 beanstalkd 的各命令，如 stats 查看信息，use, put, watch 等等。

使用場景

• 用做延時隊列：好比能夠用於若是用戶30分鐘內不操做，任務關閉。
• 用做定時任務：好比能夠用於專門的後臺任務。
• 用做異步操做：這是全部消息隊列都最經常使用的，先將任務仍進去，順序執行。
• 用做循環隊列：用release命令能夠循環執行任務，好比能夠作負載均衡任務分發。
• 用做兜底機制：好比一個請求有失敗的機率，能夠用Beanstalk不斷重試，設定超時時間，時間內嘗試到成功爲止。

操做

如下示例代碼是基於 pda/pheanstalk 這個第三方擴展寫的；

查看統計信息

整個 beanstalkd 信息

<?php

require './vendor/autoload.php';

use Pheanstalk\Pheanstalk;

$pheanstalk = new Pheanstalk('127.0.0.1');

# 查看 beanstalkd 當前的狀態信息
var_dump($pheanstalk->stats());

輸出的信息爲如下 ：

'current-jobs-urgent' => '0', // 優先級小於1024狀態爲ready的job數量
   'current-jobs-ready' => '0', // 狀態爲ready的job數量
   'current-jobs-reserved' => '0', // 狀態爲reserved的job數量
   'current-jobs-delayed' => '0', // 狀態爲delayed的job數量
   'current-jobs-buried' => '0', // 狀態爲buried的job數量
   'cmd-put' => '0', // 總共執行put指令的次數
   'cmd-peek' => '0', // 總共執行peek指令的次數
   'cmd-peek-ready' => '0', // 總共執行peek-ready指令的次數
   'cmd-peek-delayed' => '0', // 總共執行peek-delayed指令的次數
   'cmd-peek-buried' => '0', // 總共執行peek-buried指令的次數
   'cmd-reserve' => '0', // 總共執行reserve指令的次數
   'cmd-reserve-with-timeout' => '0',
   'cmd-delete' => '0',
   'cmd-release' => '0',
   'cmd-use' => '0', // 總共執行use指令的次數
   'cmd-watch' => '0', // 總共執行watch指令的次數
   'cmd-ignore' => '0',
   'cmd-bury' => '0',
   'cmd-kick' => '0',
   'cmd-touch' => '0',
   'cmd-stats' => '2',
   'cmd-stats-job' => '0',
   'cmd-stats-tube' => '0',
   'cmd-list-tubes' => '0',
   'cmd-list-tube-used' => '0',
   'cmd-list-tubes-watched' => '0',
   'cmd-pause-tube' => '0',
   'job-timeouts' => '0', // 全部超時的job的總共數量
   'total-jobs' => '0', // 建立的全部job數量
   'max-job-size' => '65535', // job的數據部分最大長度
   'current-tubes' => '1', // 當前存在的tube數量
   'current-connections' => '1', // 當前打開的鏈接數
   'current-producers' => '0', // 當前全部的打開的鏈接中至少執行一次put指令的鏈接數量
   'current-workers' => '0', // 當前全部的打開的鏈接中至少執行一次reserve指令的鏈接數量
   'current-waiting' => '0', // 當前全部的打開的鏈接中執行reserve指令可是未響應的鏈接數量
   'total-connections' => '2', // 總共處理的鏈接數
   'pid' => '3609', // 服務器進程的id
   'version' => '1.10', // 服務器版本號
   'rusage-utime' => '0.000000', // 進程總共佔用的用戶CPU時間
   'rusage-stime' => '0.001478', // 進程總共佔用的系統CPU時間
   'uptime' => '12031', // 服務器進程運行的秒數
   'binlog-oldest-index' => '2', // 開始儲存jobs的binlog索引號
   'binlog-current-index' => '2', // 當前儲存jobs的binlog索引號
   'binlog-records-migrated' => '0',
   'binlog-records-written' => '0', // 累積寫入的記錄數
   'binlog-max-size' => '10485760', // binlog的最大容量
   'id' => '37604ac4305d3b16', // 一個隨機字符串，在beanstalkd進程啓動時產生
   'hostname' => 'localhost.localdomain',

單個 job 任務的統計信息

var_export($pheanstalk->statsJob($job_4));

執行結果以下：

'id' => '1', // job id
   'tube' => 'test', // job 所在的管道
   'state' => 'reserved', // job 當前的狀態
   'pri' => '1024', // job 的優先級
   'age' => '5222', // 自 job 建立時間爲止 單位：秒
   'delay' => '0',
   'ttr' => '60', // time to run
   'time-left' => '58', // 僅在job狀態爲reserved或者delayed時有意義，當job狀態爲reserved時表示剩餘的超時時間
   'file' => '2', // 表示包含此job的binlog序號，若是沒有開啓它將爲0
   'reserves' => '10', // 表示job被reserved的次數
   'timeouts' => '0', // 表示job處理的超時時間
   'releases' => '1', // 表示job被released的次數
   'buries' => '0', // 表示job被buried的次數
   'kicks' => '0', // 表示job被kiced的次數

管道相關的命令

// 查看有多少個tube
//var_export($pheanstalk->listTubes());

// 在 put 以前預申明要使用的管道，若是管道不存在，即建立
//$pheanstalk->useTube('test');

//設置要監聽的tube
$pheanstalk->watch('test');

//取消對默認tube的監聽，能夠省略
$pheanstalk->ignore('default');

//查看監聽的tube列表
var_export($pheanstalk->listTubesWatched());

//查看test的tube當前的狀態
var_export($pheanstalk->statsTube('test'));

生產者調用的方法

// put 任務 方式一; 返回新 job 的任務標識，整型值；
$pheanstalk->useTube('test')->put(
    'hello, beanstalk, i am job 1', // 任務內容
    23, // 任務的優先級, 默認爲 1024
    0, // 不等待直接放到ready隊列中.
    60 // 處理任務的時間(單位爲秒)
);

// put 任務 方式二； 返回新 job 的任務標識，整型值；
$pheanstalk->putInTube(
    'test', // 管道名稱
    'hello, beanstalk, i am job 2', // 任務內容
    23, // 任務的優先級, 默認爲 1024
    0, // 不等待直接放到ready隊列中. 如值爲 60 表示 60秒；
    60 // 處理任務的時間(單位爲秒)
);

// 給管道里全部新任務設置延遲
$pheanstalk->pauseTube('test', 30);

// 取消管道延遲
$pheanstalk->resumeTube('test');

此處介紹幾個概念：

• 任務優先級
  任務 (job) 能夠有 0~2^32 個優先級, 0 表明最高優先級。 beanstalkd 採用最大最小堆 (Min-max heap) 處理任務優先級排序, 任什麼時候刻調用 reserve 命令的消費者老是能拿到當前優先級最高的任務, 時間複雜度爲 O(logn).
• ttr(time to run, 預設的執行時間)
  消費者必須在預設的 TTR (time-to-run) 時間內發送 delete / release / bury 改變任務狀態；不然 Beanstalkd 會認爲消息處理失敗，狀態改成 ready，而後把任務交給另外的消費者節點執行。若是消費者預計在 TTR (time-to-run) 時間內沒法完成任務, 也能夠發送 touch 命令, 它的做用是讓 Beanstalkd 重置該任務的 time-left 剩餘執行時間.

消費者方法

正常的獲取和執行 Job 流程

// 獲取 test 管道的 job
$job = $pheanstalk->watch('test')->ignore('default')->reserve();
$job_2 = $pheanstalk->reserveFromTube('test');
$job_3 = $pheanstalk->peekReady('test');

// 若是知道 job 的 id， 也能夠
$job_4 = $pheanstalk->peek($id);
// var_export($pheanstalk->statsJob($job_4));

// 獲取下一個延遲時間最短 的 job
$job_5 = $pheanstalk->peekDelayed('test');

// do job .... 這裏省略異常的考慮

// 釋聽任務 讓別人執行
$pheanstalk->release($job);
// 或成功執行完，則刪除任務
//$pheanstalk->delete($job);
// 將任務埋起來，預留
//$pheanstalk->bury($job);

處理 buried 狀態的 Job

// 獲取下一個被埋藏的 job
$job = $pheanstalk->peekBuried('test');
// 將任務狀態從 buried 改成 ready
//$pheanstalk->kickJob($job);
// 批量將指定數目的任務從 buried 改成 ready
$pheanstalk->kick(10);

總結

若是是有優先級/延時任務的需求的話, beanstalkd 是個不錯選擇。若是做爲常規的先進先出隊列來講，以性能和穩定來講 kafka/redis 會是更好的選擇，redis 自己也是全內存，隊列操做 O(1), 而 benastalkd 是 log(n)。redis 也更加成熟和穩定，同時支持本地持久化和主從。

另外有一個加分項是 beanstalkd 做者自己比較活躍，以前提了一個 pr`, 當天就獲得回饋，這也是做爲開源項目選擇一個很重要的因素。

附錄

• beanstalkd 協議