# Beanstalkd任務隊列 > 高性能離不開異步，異步離不開隊列。 ## Beanstalkd是什么 Beanstalkd 是一個輕量級消息中間件，它最大特點是將自己定位為基于管道 (tube) 和任務 (job) 的工作隊列。 Beanstalkd 支持任務優先級 (priority), 延時 (delay), 超時重發 (time-to-run) 和預留 (buried), 能夠很好的支持分布式的后臺任務和定時任務處理。它的內部采用libevent，服務器-客戶端之間采用類似Memcached的輕量級通訊協議，因此性能很高（enque: 9000 jobs/second, worker: 5200 jobs/second)。 盡管是內存隊列, Beanstalkd 提供了 binlog 機制, 當重啟 beanstalkd 時，當前任務狀態能夠從紀錄的本地 binlog 中恢復。Beanstalkd支持過有9.5 million用戶的Facebook Causes應用。后來開源，現在有PostRank大規模部署和使用，每天處理百萬級任務。Beanstalkd是典型的類Memcached設計，協議和使用方式都是同樣的風格，所以使用過Memcached的用戶會覺得Beanstalkd似曾相識。 Beanstalkd支持的語言有很多，可以參考這里：https://github.com/kr/beanstalkd/wiki/client-libraries ## Beanstalkd設計的核心概念 ### job 一個需要異步處理的任務，是Beanstalkd中的基本單元，需要放在一個tube中。 ### tube 一個有名的任務隊列，用來存儲統一類型的job，是producer和consumer操作的對象。 ### producer Job的生產者，通過put命令來將一個job放到一個tube中。 ### consumer Job的消費者，通過`reserve/release/bury/delete`命令來獲取job或改變job的狀態。 當producer直接put一個job時，job就處于READY狀態，等待consumer來處理，如果選擇延遲put，job就先到DELAYED狀態，等待時間過后才遷移到READY狀態。 consumer獲取了當前READY的job后，該job的狀態就遷移到RESERVED，這樣其他的consumer就不能再操作該job。 當consumer完成該job后，可以選擇delete, release或者bury操作： * delete操作，job從系統消亡，之后不能再獲取； * release操作，可以重新把該job狀態遷移回READY（也可以延遲該狀態遷移操作），使其他的consumer可以繼續獲取和執行該job； * bury操作，可以把該job休眠，等到需要的時候，再將休眠的job kick回READY狀態，也可以delete buride狀態的job。  #### 任務優先級 (priority): 任務 (job) 可以有 0~2^32 個優先級, 0 代表最高優先級。 beanstalkd 采用最大最小堆 (Min-max heap) 處理任務優先級排序, 任何時刻調用 reserve 命令的消費者總是能拿到當前優先級最高的任務, 時間復雜度為 O(logn). #### 延時任務 (delay): 有兩種方式可以延時執行任務 (job): 生產者發布任務時指定延時；或者當任務處理完畢后, 消費者再次將任務放入隊列延時執行 (RELEASE with <delay>)。這種機制可以實現分布式的 Java.util.Timer，這種分布式定時任務的優勢是：如果某個消費者節點故障，任務超時重發 (time-to-run) 能夠保證任務轉移到另外的節點執行。 #### 任務超時重發 (time-to-run): Beanstalkd 把任務返回給消費者以后：消費者必須在預設的 TTR (time-to-run) 時間內發送 delete / release/ bury 改變任務狀態；否則 Beanstalkd 會認為消息處理失敗，然后把任務交給另外的消費者節點執行。如果消費者預計在 TTR (time-to-run) 時間內無法完成任務, 也可以發送 touch 命令, 它的作用是讓 Beanstalkd 從系統時間重新計算 TTR (time-to-run). #### 任務預留 (buried): 如果任務因為某些原因無法執行, 消費者可以把任務置為 buried 狀態讓 Beanstalkd 保留這些任務。管理員可以通過 peek buried 命令查詢被保留的任務，并且進行人工干預。簡單的, kick <n> 能夠一次性把 n 條被保留的任務踢回隊列。 ### Beanstalkd 協議 Beanstalkd 采用類 memcached 協議, 客戶端通過文本命令與服務器交互。 這些命令可以簡單的分成三組: 1. 生產類 - use <tube> / put <priority> <delay> <ttr> [bytes]: 2. 生產者用 use 選擇一個管道 (tube), 然后用 put 命令向管道發布任務 (job). 3. 消費類 - watch <tubes> / reserve / delete <id> / release <id> <priority> <delay> / bury <id> / touch <id> 消費者用 watch 選擇多個管道 (tube), 然后用 reserve 命令獲取待執行的任務，這個命令是阻塞的。客戶端直到有任務可執行才返回。當任務處理完畢后, 消費者可以徹底刪除任務 (DELETE), 釋放任務讓別人處理 (RELEASE), 或者保留 (BURY) 任務。 維護類 - peek job / peek delayed / peek ready / peek buried / kick <n> 用于維護管道內的任務狀態, 在不改變任務狀態的條件下獲取任務。可以用消費類命令改變這些任務的狀態。 被保留 (buried) 的任務可以用 kick 命令 "踢" 回隊列。 > 協議文檔: https://raw.github.com/kr/beanstalkd/master/doc/protocol.txt ??正是有這些有趣的操作和狀態，才可以基于此做出很多意思的應用，比如要實現一個循環隊列，就可以將RESERVED狀態的job休眠掉，等沒有READY狀態的job時再將BURIED狀態的job一次性kick回READY狀態。 ## Beanstalkd有什么不足？ ??Beanstalkd 沒有提供主備同步 + 故障切換機制, 在應用中有成為單點的風險。實際應用中，可以用數據庫為任務 (job) 提供持久化存儲。 和 Memcached 類似, Beanstalkd 依賴 libevent 的單線程事件分發機制, 不能有效利用多核 cpu 的性能。這一點可以通過單機部署多個實例克服。  ??一個Beanstalkd尚無提供刪除一個tube的操作，只能將tube的job依次刪除，并讓Beanstalkd來自動刪除空tube。還有就是Beanstalkd不支持客戶端認證機制（開發者將應用場景定位在局域網）。 ??Beanstalk速度非常快，協議簡單，占用內存空間少，而且支持持久化。唯一的不足是掛了之后恢復慢，3G日志數據恢復了十多分鐘。如何安裝Beanstalkd? ## 安裝 ??使用下面的命令進行安裝，同時查看版本： sudo apt-get install beanstalkd beanstalkd -v beanstalkd 1.9 ??Beanstalkd可以使用以下命令停止和啟動： root@ubuntu-vagrant:/usr/local/nginx/conf# service beanstalkd stop * Stopping in-memory queueing server beanstalkd [ OK ] root@ubuntu-vagrant:/usr/local/nginx/conf# service beanstalkd start * Starting in-memory queueing server beanstalkd root@ubuntu-vagrant:/usr/local/nginx/conf# service beanstalkd start 通過apt-get安裝后的配置文件目錄在/etc/default/beanstalkd，里面描述了Beanstalkd監聽的地址和端口: root@ubuntu-vagrant:/usr/local/nginx/conf# cat /etc/default/beanstalkd ## Defaults for the beanstalkd init script, /etc/init.d/beanstalkd on ## Debian systems. BEANSTALKD_LISTEN_ADDR=127.0.0.1 BEANSTALKD_LISTEN_PORT=11300 # You can use BEANSTALKD_EXTRA to pass additional options. See beanstalkd(1) # for a list of the available options. Uncomment the following line for # persistent job storage. # BEANSTALKD_EXTRA="-b /var/lib/beanstalkd" ## 使用composer組件來管理它 [GitHub - davidpersson/beanstalk: Minimalistic PHP client for beanstalkd.](https://github.com/davidpersson/beanstalk)我們可以使用這個組件來幫助操作beanstalkd。 ### 安裝composer依賴組件 此處我用composer沒有拉下來，換了種方式直接git clone。 git clone https://github.com/davidpersson/beanstalk.git ### 代碼實現 <?php use Beanstalk\Client; // // A sample producer. // $beanstalk = new Client(); // For connection options see the // class documentation. $beanstalk->connect(); $beanstalk->useTube('flux'); // Begin to use tube `'flux'`. $beanstalk->put( 23, // Give the job a priority of 23. 0, // Do not wait to put job into the ready queue. 60, // Give the job 1 minute to run. '/path/to/cat-image.png' // The job's body. ); $beanstalk->disconnect(); // // A sample consumer. // $beanstalk = new Client(); $beanstalk->connect(); $beanstalk->watch('flux'); while (true) { $job = $beanstalk->reserve(); // Block until job is available. // Now $job is an array which contains its ID and body: // ['id' => 123, 'body' => '/path/to/cat-image.png'] // Processing of the job... $result = touch($job['body']); if ($result) { $beanstalk->delete($job['id']); } else { $beanstalk->bury($job['id']); } } // When exiting i.e. on critical error conditions // you may also want to disconnect the consumer. // $beanstalk->disconnect(); ## 控制臺 為了可以可視化的在網頁上查看消息隊列的運行情況，我推薦使用控制臺。 [ptrofimov/beanstalk_console - Packagist](https://packagist.org/packages/ptrofimov/beanstalk_console)我們使用這個項目作為消息隊列的控制臺，直觀的查看隊列任務的執行。 安裝console項目： composer create-project ptrofimov/beanstalk_console 然后給他配置虛擬站點，就可以進入控制臺查看beanstalkd狀態了。  此處因為使用lnmp一鍵腳本搭建環境，踩了幾個坑。有時間和大家分享下。