[TOC] 注意如果一直執行一個出錯的job,他會一直try,最終把內存爆掉 ~~~ $pheanstalk=new Pheanstalk('ip','11300'); $pheanstalk以下這個變量認為是beanstalkd的實例了 ~~~ # 維護 ## 查看狀態stats ~~~ require __DIR__."/vendor/autoload.php"; use Pheanstalk\Pheanstalk; //ip和端口,由于我使用的是docker,寫容器名就可以了 $pheanstalk=new Pheanstalk('beanstalkd','11300'); print_r($pheanstalk->stats()); ~~~ ~~~ Pheanstalk\Response\ArrayResponse Object ( [_name:Pheanstalk\Response\ArrayResponse:private] => OK [storage:ArrayObject:private] => Array ( [current-jobs-urgent] => 0 # 當前存在優先級的任務是0 [current-jobs-ready] => 0 # 當前處在準備讀取的任務數是0 [current-jobs-reserved] => 0 # 當前正在執行的任務數是0 [current-jobs-delayed] => 0 # 當前處在延遲的任務數是0 [current-jobs-buried] => 0 # 當前處在預留任務數是0 #cmd開頭表示運行過這個命令的累計數量 [cmd-put] => 0 [cmd-peek] => 0 [cmd-peek-ready] => 0 [cmd-peek-delayed] => 0 [cmd-peek-buried] => 0 [cmd-reserve] => 0 [cmd-reserve-with-timeout] => 0 [cmd-delete] => 0 [cmd-release] => 0 [cmd-use] => 0 [cmd-watch] => 0 [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 # 任務超時的次數.就是任務超過時間,被重新放入管道的次數 [total-jobs] => 0 # 用戶累計數量 [max-job-size] => 65535 # 任務字符串大小,不能超過這個,也就是2的16次方 [current-tubes] => 1 # 當前存在的管道數量 [current-connections] => 1 # 當前打開的連接數量 [current-producers] => 0 # 當前有多少個生產者,連接上發送至少一個就算 [current-workers] => 0 # 有多少消費者 [current-waiting] => 0 # 消費者發送了reserved但沒有響應的數量,就一直阻塞在那,等著任務 [total-connections] => 2 #總共的鏈接數 [pid] => 1 #pid [version] => 1.10 #版本 [rusage-utime] => 0.010000 #進程執行用戶代碼的時間 [rusage-stime] => 0.000000 #進程執行內核代碼的時間,這2個主要看負載的 [uptime] => 160 # 從啟動到現在一共運行了多少秒 [binlog-oldest-index] => 0 [binlog-current-index] => 0 [binlog-records-migrated] => 0 [binlog-records-written] => 0 [binlog-max-size] => 10485760 [id] => 1bd5de699bcabfbc [hostname] => 2935a637423d ) ) ~~~ ## 查看管道數量listTubes ~~~ print_r($pheanstalk->listTubes()); ~~~ ~~~ Array ( [0] => default ) ~~~ 目前存在一個管道,這個管道的名字教default是默認管道,一般不使用默認管道,自己創建管道使用 ## 查看具體管道詳情statsTube ~~~ 參數是管道的名字 print_r($pheanstalk->statsTube('default')); ~~~ ~~~ Pheanstalk\Response\ArrayResponse Object ( [_name:Pheanstalk\Response\ArrayResponse:private] => OK [storage:ArrayObject:private] => Array ( [name] => default #當前管道的名字 [current-jobs-urgent] => 0 #當前管道存在任務優先級的任務數 [current-jobs-ready] => 0 # 當前處在準備讀取的任務數 [current-jobs-reserved] => 0 # 當前正在執行的任務數 [current-jobs-delayed] => 0 # 當前處在延遲的任務數 [current-jobs-buried] => 0 # 當前處在預留任務數 [total-jobs] => 0 # 當前有多少個任務數 [current-using] => 1 #當前有多少個用這個管道 [current-watching] => 1 #當前有多少個消費者在使用這個管道 [current-waiting] => 0 # 當前有多少個消費者在等待這個任務 [cmd-delete] => 0 [cmd-pause-tube] => 0 [pause] => 0 [pause-time-left] => 0 ) ) ~~~ ## 使用一個管道useTube ~~~ //useTube表示要使用那個管道,put是放入的意思 $pheanstalk->useTube('default')->put('test'); //查看管道狀態 print_r($pheanstalk->statsTube('default')); ~~~ ~~~ Pheanstalk\Response\ArrayResponse Object ( [_name:Pheanstalk\Response\ArrayResponse:private] => OK [storage:ArrayObject:private] => Array ( [name] => default [current-jobs-urgent] => 0 [current-jobs-ready] => 1 #準備的任務數是1個 [current-jobs-reserved] => 0 [current-jobs-delayed] => 0 [current-jobs-buried] => 0 [total-jobs] => 1 [current-using] => 1 [current-watching] => 1 [current-waiting] => 0 [cmd-delete] => 0 [cmd-pause-tube] => 0 [pause] => 0 [pause-time-left] => 0 ) ) ~~~ ## 查看任務詳情statsJob ~~~ //watch表示監聽管道,reserve取出 $job=$pheanstalk->watch('default')->reserve(); //statsJob查看隊列詳情 $stats=$pheanstalk->statsJob($job); print_r($stats); ~~~ ~~~ Pheanstalk\Response\ArrayResponse Object ( [_name:Pheanstalk\Response\ArrayResponse:private] => OK [storage:ArrayObject:private] => Array ( [id] => 1 #當前任務的id,這個id是唯一的 [tube] => default #當前任務所在的管道 [state] => reserved #當前任務的狀態是什么 [pri] => 1024 # 任務優先級,1024表示默認優先級 [age] => 206 # 表示從創建到現在活了多少秒 [delay] => 0 # 延遲的描述 [ttr] => 60 #當前任務的ttr時間,默認60秒,也可以設置 [time-left] => 59 # 這個任務在reserved維持的秒數,ttr是60,超過就超時了 [file] => 0 # 沒有開啟binlog日志這個就是0,開binlog這就表示0 [reserves] => 1 #當前狀態reserves的次數,超時就重新放入在reserves,再超時 [timeouts] => 0 # ttr超時的次數 [releases] => 0 # 重設任務的狀態 [buries] => 0 # 被預留的次數 [kicks] => 0 # 讓一個任務從預留的狀態提回到releases狀態 ) ) ~~~ ## 根據任務的id來查看任務peek ~~~ //這個數字是任務的id $job=$pheanstalk->peek(1); $stats=$pheanstalk->statsJob($job); print_r($stats); ~~~ 調用put方法的時候也會返回一個任務的id # 生產類 ## putInTube把任務放到管道 ~~~ /** * 第一個參數是管道名字 * 第二個參數是要放入任務的字符串 * 第三參數是優先級,可省略不寫,默認1024 * 返回任務的id */ $id=$pheanstalk->putInTube('default','666','1000'); print_r($id); ~~~ ## put任務放到管道 ~~~ $tube=$pheanstalk->useTube('default'); /** * 第二個參數是任務優先級,0最大,越小越大 [可省略] * 第三個參數是延遲的秒數,表示要延遲多少秒執行 [可省略] * 第四個參數表示超過多少秒就重發,超時重發 [可省略] */ $tube->put('test',4,20,30); ~~~ ## delete刪除任務 讀取到任務,然后把他刪除 ~~~ $job=$pheanstalk->watch('default')->reserve(); print_r($job); $pheanstalk->delete($job); ~~~ ## touch即將延遲的任務續命,讓任務重置ttr時間 如果我們一個任務執行10秒,任務的延遲也是10秒,那這個任務我們一直拿不到,怎么辦呢? 我們讓這個任務等等,這樣就能拿到了 ~~~ $tube=$pheanstalk->useTube('default'); $tube->put('test',0,0,10); $job=$pheanstalk->watch('default')->reserve(); print_r($pheanstalk->statsJob($job)); sleep(9); //給任務續命 $pheanstalk->touch($job); $pheanstalk->delete($job); ~~~ # 消費類 ## watch監聽管道,可以同時監聽多個 ~~~ //watch參數是監聽的管道名字 $job=$pheanstalk->watch('newUsers') ->watch('default'); //列出監聽的管道 $tubes=$pheanstalk->listTubesWatched(); print_r($tubes); ~~~ ## ignore忽略的管道 ~~~ $job=$pheanstalk->watch('newUsers') ->watch('default') ->ignore('default'); //列出監聽的管道 $tubes=$pheanstalk->listTubesWatched(); print_r($tubes); ~~~ ## reserve阻塞的方式獲取任務 ~~~ $job=$pheanstalk->watch('default') ->reserve(); var_dump($job); $pheanstalk->delete($job); ~~~ ~~~ /** * reserve參數 * 設置reserve阻塞的秒數 [可省略] * 超過這個秒數立即返回 */ $job=$pheanstalk->watch('default') ->reserve(3); var_dump($job); $pheanstalk->delete($job); ~~~ ## release把任務重新放回管道 ~~~ $job=$pheanstalk->watch('default')->reserve(); /** * 因為一些元素,比如網絡連接不上 */ $module=false; if (!$module) { //把取出來的任務重新放回管道,任務變為ready狀態 $pheanstalk->release($job); }else{ $pheanstalk->delete($job); } ~~~ ## bury把任務放一邊,預留 ~~~ $job=$pheanstalk->watch('default')->reserve(); /** * 比如我們要發郵件,但是郵件服務器連接不上,就把這個任務放一邊設為預留 */ $pheanstalk->bury($job); ~~~ ## peekBuried,kickJob把任務從預留中取出來,變為ready ~~~ /** * 從管道中取出預留的任務 */ $job=$pheanstalk->peekBuried('default'); /** * 把這個任務變為ready狀態,這樣消費者就能去消費了 */ $pheanstalk->kickJob($job); var_dump($pheanstalk->statsJob($job)); ~~~ ## kick批量的把預留的設為ready ~~~ /** * 從這個管道中,把任務id小于999的預留任務全部變為ready */ $pheanstalk->useTube('default')->kick(999); ~~~ ## peekReady一次性取出所有ready的任務,不阻塞,無法獲取任務會拋出異常 ~~~ /** * 一次性從這個管道中取出ready的任務 */ $job=$pheanstalk->peekReady('default'); var_dump($job); $pheanstalk->delete($job); #相似還有peekDelayed一次性取出Delayed的消息 #需要注意peek的一些方法,不阻塞,無法獲取任務會拋出異常 ~~~ ## pauseTube設置管道延遲多少秒 ~~~ /** * 設置管道延遲10秒 */ $pheanstalk->pauseTube('default',10); $job=$pheanstalk->watch('default')->reserve(); var_dump($job); ~~~ ## resumeTube把管道從延遲中恢復 ~~~ $pheanstalk->resumeTube('default'); ~~~ # 具體使用 ~~~ /** * 生產者 */ $pheanstalk->useTube('default') ->put(222); ~~~ ~~~ /** * 消費者 */ while(true) { $queue = $pheanstalk->watch('default'); if(($job=$queue->reserve(5)) == false) { continue; } //沒有任務的時候reserve會阻塞在這里 $job=$job->reserve(); //getData是從任務中取出數據 echo $job->getData(); //處理具體業務 //這邊要刪除,不刪除的話,超時會重新放入管道 $pheanstalk->delete($job); } ~~~