# 消息隊列 在開發應用程序的時候，你可能需要執行一些任務，例如解析和存儲一些大型文件，這些任務在Web 請求期間需要耗費很長時間才能執行完成。 這時候我們就可以用消息隊列來創建在后臺處理的隊列任務。 通過將時間密集型任務移至隊列，這樣你的應用程序可以以極快的速度響應 Web 請求，并為你的程序提供更好的用戶體驗 傳統的程序執行流程一般是 即時|同步|串行的，在某些場景下，會存在并發低，吞吐量低，響應時間長等問題。在大型系統中，一般會引入消息隊列的組件，將流程中部分任務抽離出來放入消息隊列，并由專門的消費者作針對性的處理，從而降低系統耦合度，提高系統性能和可用性。 一般來說，可以抽離的任務具有以下的特點： ●允許延后|異步|并行處理 （相對于傳統的 即時|同步|串行 的執行方式） ○允許延后：搶購活動時，先快速緩沖有限的參與人數到消息隊列，后續再排隊處理實際的搶購業務； ○允許異步：業務處理過程中的郵件，短信等通知 ○允許并行：用戶支付成功之后，郵件通知，微信通知，短信通知可以由多個不同的消費者并行執行，通知到達的時間不要求先后順序。 ●允許失敗和重試 ○強一致性的業務放入核心流程處理 ○無一致性要求或最終一致即可的業務放入隊列處理 ## 1.配置 1.當前隊列有三種驅動方式 | 驅動方式 | 說明 | | --- | --- | | sync | 同步執行，有新隊列任務則通過事件 Event 來直接觸發執行,不存儲任務,直接執行 | | atabase | 數據庫存儲，新隊列任務數據存儲到數據庫，隊列執行程序再從數據庫中讀取任務數據 | | redis(推薦) | Redis 存儲，新隊列任務數據存儲到 Redis，隊列執行程序再從 Redis 中讀取任務數據 | 2.先打開config下的queue.php文件 ``` <?php return [ 'default' => 'database', 'connections' => [ 'sync' => [ 'type' => 'sync', ], 'database' => [ 'type' => 'database', 'queue' => 'default', 'table' => 'jobs', 'connection' => null, ], 'redis' => [ // 驅動方式 'type' => 'redis', 'queue' => 'default', // 服務器地址 'host' => env('redis.redis_hostname', '127.0.0.1'), // 端口 'port' => env('redis.port', '6379'), // 密碼 'password' => env('redis.redis_password', ''), // 緩存有效期 0表示永久緩存 'expire' => 0 , // 緩存前綴 'prefix' => 'QUERY', // 緩存標簽前綴 'tag_prefix' => 'QUERY:', // 數據庫 0號數據庫 'select' => env('redis.select', 0), // 服務端主動關閉 'timeout' => 0, 'persistent' => false, ], ], 'failed' => [ 'type' => 'database',// none 不記錄失敗任務 database 將失敗任務遷移到失敗任務表 'table' => 'jobs_failed', ], ]; ``` default值:當前啟動的隊列驅動 ## 2.消息的消費和刪除 1.在app/job目錄下創建一個任務消費類 例如: ``` <?php namespace app\job; use core\base\BaseJob; /** * 訂單延時關閉任務 */ class OrderClose extends BaseJob { /** * 消費 * @param $order_id 訂單id * @param $close_time 關閉時間 * @return true */ protected function doJob($order_id, $close_time) { //業務代碼 //...... return true; } } ``` ## 3.消息的創建與推送 在業務控制器中新增一個控制器,創建一條消息,并推送到隊列 1、新增一個控制器\\app\\adminapi\\controller\\test\\JobTest.php,在該控制器中添加 helloWorldJob 方法,實例化剛剛創建的任務類,并調用dispatch函數 ``` <?php namespace app\adminapi\controller\test; use app\job\OrderClose; use core\base\BaseAdminController; use think\Response; /** * 測試隊列任務 */ class JobTest extends BaseAdminController { /** * 創建消息 * @return Response */ public function helloWorldJob() { $order_id = 1;//訂單id $close_time = time();//訂單關閉時間 $is_pushed = OrderClose::dispatch(['order_id' => $order_id, 'close_time' => $close_time]); if( $is_pushed !== false ){ echo date('Y-m-d H:i:s') . " Hello World Job is Pushed."; }else{ echo 'Hello World Job Error.'; } } } ``` dispatch函數的參數介紹 * action 任務消費類內部的方法, 非必填,不填寫或類型為數組的話的話默認會調用doJob函數,如果是數組的話則視為當前值為data參數 * data 數據傳參, 他會將這兒的數組作為可變數量的參數傳遞給對應函數的給定變量參數 * secs 延時執行時間(單位為秒),默認為0, 支持延時執行和定時執行,如果為0就是普通的異步任務,如果傳值則視為當前時間的\*秒后延時執行 * queue\_name 當前任務歸屬的隊列名稱，如果為新隊列，會自動創建,結合命令行使用 * is\_async 默認為true,當前任務是否需要異步執行,最高優先級 2、新建一個路由配置文件\\app\\adminapi\\route\\job.php ``` <?php use think\facade\Route; Route::group('job', function () { //獲取本地插件 Route::get('test', 'test.JobTest/helloWorldJob'); }); ``` 至此，所有的代碼都已準備完畢，在運行消息隊列之前，我們先看一下現在的目錄結構：  ## 4.發布任務 在瀏覽器中訪問 [http://your.project.com/adminapi/job/test](http://your.project.domain/index/job_test/actionWithHelloJob) ,可以看到消息推送成功。  ## 5.處理任務 切換當前終端窗口的目錄到項目niucloud目錄下，執行 ``` php think queue:listen ``` 可以看到執行的結果類似如下:  1. 命令行參數 ``` php think queue:listen \ --queue helloJobQueue \ //監聽的隊列的名稱 --delay 0 \ //如果本次任務執行拋出異常且任務未被刪除時，設置其下次執行前延遲多少秒,默認為0 --memory 128 \ //該進程允許使用的內存上限，以 M 為單位 --sleep 3 \ //如果隊列中無任務，則多長時間后重新檢查 --tries 0 \ //如果任務已經超過重發次數上限，則進入失敗處理邏輯，默認為0 --timeout 60 // work 進程允許執行的最長時間，以秒為單位 ``` 這兒需要重點說明的是 * \--queue 參數 監聽的隊列名稱,結合任務創建時的隊列名來使用 * \--tries 任務失敗幾次后進入失敗數理邏輯,防止任務一直重復執行 2. 為了穩定要放在消息隊列Supervisor下運行 至此，我們成功地完成了一個消息的 創建 -> 推送 -> 消費 -> 刪除 的流程