1.根據選擇的存儲方式，在\application\extra\queue.php這個配置文件中，添加消息隊列對應的驅動配置，例如：
return [
'connector' => 'Redis', // Redis 驅動
'expire' => 60, // 任務的過期時間，默認為60秒; 若要禁用，則設置為 null
'default' => 'default', // 默認的隊列名稱
'host' => '127.0.0.1', // redis 主機ip
'port' => 6379, // redis 端口
'password' => '', // redis 密碼
'select' => 1, // 使用哪一個 db，默認為 db0
'timeout' => 0, // redis連接的超時時間
'persistent' => false, // 是否是長連接
// 'connector' => 'Database', // 數據庫驅動
// 'expire' => 60, // 任務的過期時間，默認為60秒; 若要禁用，則設置為 null
// 'default' => 'default', // 默認的隊列名稱
// 'table' => 'jobs', // 存儲消息的表名，不帶前綴
// 'dsn' => [],
// 'connector' => 'Topthink', // ThinkPHP內部的隊列通知服務平臺 ，本文不作介紹
// 'token' => '',
// 'project_id' => '',
// 'protocol' => 'https',
// 'host' => 'qns.topthink.com',
// 'port' => 443,
// 'api_version' => 1,
// 'max_retries' => 3,
// 'default' => 'default',
// 'connector' => 'Sync', // Sync 驅動，該驅動的實際作用是取消消息隊列，還原為同步執行
];
2.在業務控制器中創建一個新的消息，并推送到helloJobQueue隊列。
新增\application\index\controller\JobTest.php控制器，在該控制器中添加actionWithHelloJob方法，代碼：
namespace app\index\controller;
use think\Queue;
class JobTest
{
/*
* 測試隊列action
* */
public function actionWithHelloJob(){
// 1.當前任務將由哪個類來負責處理。
// 當輪到該任務時，系統將生成一個該類的實例，并調用其 fire 方法
$jobHandlerClassName = 'app\index\job\Hello@fire';
// 2.當前任務歸屬的隊列名稱，如果為新隊列，會自動創建
$jobQueueName = "helloJobQueue";
// 3.當前任務所需的業務數據 . 不能為 resource 類型，其他類型最終將轉化為json形式的字符串
// ( jobData 為對象時，需要在先在此處手動序列化，否則只存儲其public屬性的鍵值對)
$jobData = [ 'name' => 'test'.rand(), 'password'=>rand()] ;
// 4.將該任務推送到消息隊列，等待對應的消費者去執行
$time2wait = strtotime('2018-09-08 11:15:00') - strtotime('now'); // 定時執行
$isPushed = Queue::later($time2wait, $jobHandlerClassName , $jobData , $jobQueueName );
// database 驅動時，返回值為 1|false ; redis 驅動時，返回值為 隨機字符串|false
if( $isPushed !== false ){
echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ"."
";
}else{
echo 'Oops, something went wrong.';
}
}
}
3.編寫Hello消費者類，用來處理helloJobQueue隊列中的任務，新增\application\index\job\Hello.php消費者類，并編寫其fire()方法，代碼：
/**
* 文件路徑： \application\index\job\Hello.php
* 這是一個消費者類，用于處理 helloJobQueue 隊列中的任務
*/
namespace app\index\job;
use think\queue\Job;
use think\Db;
class Hello {
/**
* fire方法是消息隊列默認調用的方法
* @param Job $job 當前的任務對象
* @param array|mixed $data 發布任務時自定義的數據
*/
public function fire(Job $job,$data){
// 如有必要,可以根據業務需求和數據庫中的最新數據,判斷該任務是否仍有必要執行.
$isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
if(!$isJobStillNeedToBeDone){
$job->delete();
return;
}
$isJobDone = $this->doHelloJob($data);
if ($isJobDone) {
//如果任務執行成功， 記得刪除任務
$job->delete();
}else{
if ($job->attempts() > 3) {
//通過這個方法可以檢查這個任務已經重試了幾次了
$job->delete();
// 也可以重新發布這個任務
//$job->release(2); //$delay為延遲時間，表示該任務延遲2秒后再執行
}
}
}
/**
* 有些消息在到達消費者時,可能已經不再需要執行了
* @param array|mixed $data 發布任務時自定義的數據
* @return boolean 任務執行的結果
*/
private function checkDatabaseToSeeIfJobNeedToBeDone($data){
return true;
}
/**
* 根據消息中的數據進行實際的業務處理
* @param array|mixed $data 發布任務時自定義的數據
* @return boolean 任務執行的結果
*/
private function doHelloJob($data) {
// 根據消息中的數據進行實際的業務處理...
// test
Db::name('admin')->insert([
'name'=>$data['name'],
'password'=>$data['password']
]);
return true;
}
}
