> think-queue是ThinkPHP官方提供的一個消息隊列服務，是專門支持隊列服務的擴展包。 > think-queue消息隊列適用于大并發或返回結果時間比較長且需要批量操作的第三方接口，可用于短信發送、郵件發送、APP推送。 > think-queue消息隊列可進行發布、獲取、執行、刪除、重發、失敗處理、延遲執行、超時控制等操作。 ## **執行流程圖：**  ## **安裝think-queue擴展** 在命令行切換到你的WEB根目錄并執行下面的命令： ~~~ // 安裝think-queue擴展 composer require topthink/think-queue ~~~ 安裝后在`app/config`下將產生一個queue.php配置文件： ~~~ return [ //驅動類型，可選擇 sync(默認):同步執行，database:數據庫驅動,redis:Redis驅動 'default' => 'redis', 'connections' => [ 'sync' => [ 'type' => 'sync', ], 'database' => [ 'type' => 'database', 'queue' => 'default', 'table' => 'jobs', ], 'redis' => [ 'type' => 'redis', 'queue' => 'default', 'host' => '127.0.0.1', 'port' => 6379, 'password' => '', 'select' => 0, 'timeout' => 0, 'persistent' => false, ], ], 'failed' => [ 'type' => 'none', 'table' => 'failed_jobs', ], ]; ~~~ **一、優化上一章中的SmsTask.php 的邏輯代碼** ~~~ <?php declare (strict_types = 1); namespace app\listener; use think\facade\Queue; class SmsTask { /** * 事件監聽處理 * * @return mixed */ public function handle($event) { var_dump($event->data);//event的data數據即server->task()傳入的數據 /*創建新消息并推送到消息隊列*/ // 當前任務由哪個類負責處理 $job_handler_classname = "app\job\Dismiss"; // 當前隊列歸屬的隊列名稱 $job_queue_name = "dismiss_job_queue"; // 當前任務所需的業務數據 $job_data = ["ts"=>time(), "bizid"=>uniqid(), "params"=>$event->data]; // 將任務推送到消息隊列等待對應的消費者去執行 $is_pushed = Queue::push($job_handler_classname, $job_data, $job_queue_name); // 可以調用 finish 方法通知其他事件類，通知當前異步任務已經完成了（非必須調用） // 參數 $event 是 Swoole\Server\Task 類的一個對象 可以調用 finish 方法觸發 task 任務的 onFinish 事件 $event->finish($event->data); return; } } ~~~ **二、創建消費者類 Dismiss.php 的邏輯代碼** ~~~ <?php // +---------------------------------------------------------------------- // | najing [ 通用后臺管理系統 ] // +---------------------------------------------------------------------- // | Copyright (c) 2020 http://www.najingquan.com All rights reserved. // +---------------------------------------------------------------------- // | Author: 救火隊隊長 // +---------------------------------------------------------------------- namespace app\job; use think\facade\Log; use think\queue\Job; /** * 消費者類 * 用于處理隊列中的任務 */ class Dismiss { /** * fire是消息隊列默認調用的方法 * @param Job $job 當前的任務對象 * @param array|mixed $data 發布任務時自定義的數據 */ public function fire(Job $job, $data) { //有效消息到達消費者時可能已經不再需要執行了 if(!$this->checkJob($data)){ $job->delete(); return; } //執行業務處理 if($this->doJob($data)){ $job->delete();//任務執行成功后刪除 Log::record("job has been down and deleted"); }else{ //檢查任務重試次數 if($job->attempts() > 3){ Log::record("job has been retried more that 3 times"); $job->delete(); } } } /** * 消息在到達消費者時可能已經不需要執行了 * @param array|mixed $data 發布任務時自定義的數據 * @return boolean 任務執行的結果 */ private function checkJob($data) { $ts = $data["ts"]; $bizid = $data["bizid"]; $params = $data["params"]; return true; } /** * 根據消息中的數據進行實際的業務處理 */ private function doJob($data) { // 實際業務流程處理 var_dump($data); Log::record(json_encode($data)); return true; } } ~~~ **三、啟動一個工作進程來處理消息隊列** ~~~ php think queue:work --queue dismiss_job_queue ~~~ **四、瀏覽器訪問注冊方法** 發送短信的任務被異步推送到了隊列里面，然后由消費隊列正常消費：