安裝 在此我就不再略過TP6的項目創建過程了，大致就是安裝composer工具，安裝成功以后，再使用composer去創建項目即可。 think-queue 安裝 ~~~ composer require topthink/think-queue ~~~ 項目中添加驅動配置 我們需要在安裝好的config下找到 queue.php [](javascript:void(0); "復制代碼") ~~~ <?php return [ 'default' => 'redis', 'connections' => [ 'sync' => [ 'type' => 'sync', ], 'database' => [ 'type' => 'database', 'queue' => 'default', 'table' => 'jobs', 'connection' => null, ], 'redis' => [ 'type' => 'redis', 'queue' => 'default', 'host' => '127.0.0.1', 'port' => 6379, 'password' => '', 'select' => 4, 'timeout' => 0, 'persistent' => false, ], ], 'failed' => [ 'type' => 'none', 'table' => 'failed_jobs', ], ]; ~~~ [](javascript:void(0); "復制代碼") 生產者 [](javascript:void(0); "復制代碼") ~~~ <?php namespace app\controller; use app\BaseController; use think\facade\Queue; class Index extends BaseController { public function queue() { //當前任務將由哪個類來負責處理。 //當輪到該任務時，系統將生成一個該類的實例，并默認調用其 fire 方法 $jobHandlerClassName = 'app\Job\Order'; //當前任務歸屬的隊列名稱，如果為新隊列，會自動創建 //php think queue:work --queue orderJobQueue //php think queue:work --queue orderJobQueue --daemon $jobQueueName = "orderJobQueue"; //數組數據 $orderData = [ 'id' => uniqid(), 'time' => time(), 'message' => 'later message83' ]; //將該任務推送到消息隊列，等待對應的消費者去執行 //這里只是負責將數據添加到相應的隊列名稱的隊列里，消費者與生產者并無聯系 //立即執行 $isPushed = Queue::push($jobHandlerClassName, $orderData, $jobQueueName); //延遲10秒后執行 //$isPushed = Queue::later(10, $jobHandlerClassName, $orderData, $jobQueueName); if ($isPushed !== false) { echo date('Y-m-d H:i:s') . " 隊列添加成功"; } else { echo '隊列添加失敗'; } } } ~~~ [](javascript:void(0); "復制代碼") 消費者 [](javascript:void(0); "復制代碼") ~~~ <?php namespace app\Job; use think\facade\Log; use think\queue\Job; /** * @Title: app\task\job$Order * @Package package_name * @Description: todo(測試訂單消費者) * @author Jack */ class Order { /** * @Title: fire * @Description: todo(fire方法是消息隊列默認調用的方法) * @param Job $job * @param array $data * @author Jack * @throws */ public function fire(Job $job, array $data) { //有些消息在到達消費者時,可能已經不再需要執行了 $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data); if(!$isJobStillNeedToBeDone){ $job->delete(); return; } $jobId = $job->getJobId(); $isJobDone = $this->orders($data, $jobId); if ($isJobDone) { //如果任務執行成功，記得刪除任務 $job->delete(); } else { //通過這個方法可以檢查這個任務已經重試了幾次了 if ($job->attempts() > 3){ Log::error('試了3次了'); $job->delete(); //也可以重新發布這個任務 //print("<info>Hello Job will be availabe again after 2s."."</info>\n"); //$job->release(2); //$delay為延遲時間，表示該任務延遲2秒后再執行 } } } /** * @Title: checkDatabaseToSeeIfJobNeedToBeDone * @Description: todo(有些消息在到達消費者時,可能已經不再需要執行了) * @param array $data * @return boolean * @author Jack * @throws */ private function checkDatabaseToSeeIfJobNeedToBeDone($data) { return true; } /** * @Title: orders * @Description: todo(數據處理) * @param array $data * @author Jack * @throws */ public function orders(array $data, $jobId) { //對訂單進行數據庫操作或其他等等 Log::info(date('Y-m-d H:i:s') . ' - data：' . json_encode($data)); return true; } } ~~~ [](javascript:void(0); "復制代碼") 服務器執行常駐命令 ~~~ php think queue:work --queue orderJobQueue ~~~